prodigy.analysis.session¶

Analysis sessions¶

An AnalysisSession bundles everything a single analysis run needs to know up front: the program, the input distribution, the configuration and one decision per while loop (which loop strategy to use, with which parameters). Instruction handlers no longer call input() themselves; they ask the session’s DecisionProvider:

InteractiveDecisions asks on the terminal with the historic prompts. The CLI uses it, so prodigy main behaves as before.
FixedDecisions answers from a dictionary (e.g. parsed from JSON). The web GUI, a language server or prodigy analyze use it. A missing decision raises DecisionRequired instead of blocking.

While a session runs, it also records what happened: query results with sound intervals (see prodigy.analysis.result), what each loop strategy produced and certified, warnings, progress and optionally the distribution after each top-level instruction. AnalysisSession.run() returns an AnalysisResult whose to_json() is the exchange format for front ends.

Code that calls compute_discrete_distribution() or the handlers directly (tests, scripts, older code) does not need a session: if none is active, current_session() hands out a detached interactive one, which reproduces the old behaviour.

token_positions(source, pattern)[source]¶

1-based (line, column) of every match of pattern outside of comments.

Parameters:

source (str)
pattern (Pattern[str])

Return type:

List[Tuple[int, int]]

class LoopInfo(id, cond, depth, function=None, line=None, column=None)[source]¶

Bases: object

A while loop of the analysed program, identified by its position in the text.

Parameters:

id (int)
cond (str)
depth (int)
function (str | None)
line (int | None)
column (int | None)

id: int¶: Index of the loop in textual order (0-based); the key for decisions.

cond: str¶: The loop guard as pGCL text.

depth: int¶: Number of enclosing while loops.

function: str | None = None¶: Name of the enclosing function declaration, if any.

line: int | None = None¶

column: int | None = None¶

describe()[source]¶

Return type:: str

to_json()[source]¶

Return type:: Dict[str, Any]

class LoopRegistry(program, source=None)[source]¶

Bases: object

Assigns stable ids to the while loops of a program (textual pre-order).

Parameters:

program (Program | None)
source (str | None)

loops: List[LoopInfo]¶

lookup(instr)[source]¶

Parameters:: instr (WhileInstr)
Return type:: LoopInfo | None

query_instructions(program)[source]¶

Parameters:: program (Program)
Return type:: List[SkipInstr | AbortInstr | WhileInstr | IfInstr | AsgnInstr | ChoiceInstr | LoopInstr | TickInstr | ObserveInstr | QueryInstr | ProbabilityQueryInstr | ExpectationInstr | PlotInstr | PrintInstr | OptimizationQuery]

class LoopDecision(strategy, params=dict())[source]¶

Bases: object

Which strategy to use for a loop, and its parameters (raw, not yet parsed).

Parameters:

strategy (str)
params (Dict[str, Any])

strategy: str¶

params: Dict[str, Any]¶

static parse_spec(spec)[source]¶

Parses the command-line form STRATEGY[:name=value,name=value].

Example: unroll:iterations=100 or invariant:invariant_file=inv.pgcl.

Parameters:: spec (str)
Return type:: LoopDecision

static from_json(data)[source]¶

Parameters:: data (str | Mapping[str, Any] | LoopDecision)
Return type:: LoopDecision

to_json()[source]¶

Return type:: Dict[str, Any]

DEFAULT_KEY = '*'¶: Key for the decision that applies to every loop without an own decision.

parse_decisions(data)[source]¶

Parses {"0": {...}, "1": "unroll:iterations=10", "*": {...}}.

Parameters:: data (Mapping[Any, Any] | None)
Return type:: Dict[int | str, LoopDecision]

validate_decisions(decisions, loops=None)[source]¶

Problems with decisions (unknown strategies, unknown loops, bad parameters).

Parameters:

decisions (Mapping[int | str, LoopDecision])
loops (LoopRegistry | None)

Return type:

List[str]

exception DecisionRequired(message, *, loop=None, strategy=None, param=None)[source]¶

Bases: Exception

A non-interactive analysis reached a choice that was not decided up front.

Parameters:

message (str)
loop (LoopInfo | None)
strategy (str | None)
param (str | None)

to_json()[source]¶

Return type:: Dict[str, Any]

class DecisionProvider(decisions=None)[source]¶

Bases: object

Answers the questions an analysis may have while it runs.

Parameters:: decisions (Mapping[Any, Any] | None)

interactive: bool = False¶

decisions: Dict[int | str, LoopDecision]¶

preset(loop)[source]¶

Parameters:: loop (LoopInfo | None)
Return type:: LoopDecision | None

loop_decision(session, loop, instruction)[source]¶

Parameters:

session (AnalysisSession)
loop (LoopInfo | None)
instruction (WhileInstr)

Return type:

LoopDecision

ask_param(session, task, spec)[source]¶

Parameters:

session (AnalysisSession)
task (LoopTask)
spec (ParamSpec)

Return type:

Any

choose_candidate(task, candidates)[source]¶

Parameters:

task (LoopTask)
candidates (List[str])

Return type:

int

next_step()[source]¶

Called after each instruction in step-wise mode.

Return type:: None

continue_plot(probability)[source]¶

Parameters:: probability (float)
Return type:: bool

class InteractiveDecisions(decisions=None)[source]¶

Bases: DecisionProvider

Asks on the terminal, with the prompts the CLI has always used.

Decisions given up front (e.g. prodigy main --loop 0=unroll:iterations=10) are used without asking.

Parameters:: decisions (Mapping[Any, Any] | None)

interactive: bool = True¶

loop_decision(session, loop, instruction)[source]¶

ask_param(session, task, spec)[source]¶

next_step()[source]¶

Called after each instruction in step-wise mode.

Return type:: None

continue_plot(probability)[source]¶

Parameters:: probability (float)
Return type:: bool

class FixedDecisions(decisions=None)[source]¶

Bases: DecisionProvider

Answers only from decisions given up front; never blocks.

Parameters:: decisions (Mapping[Any, Any] | None)

interactive: bool = False¶

loop_decision(session, loop, instruction)[source]¶

ask_param(session, task, spec)[source]¶

class LoopRecord(loop, cond, strategy, params, invocations=0, kind='exact', residual=0, time=0.0, certificates=list(), messages=list(), progress=None, failed=None, inside_checks=0, suppressed_messages=0)[source]¶

Bases: object

Everything one loop produced during a run (a loop may be entered several times).

Parameters:

loop (LoopInfo | None)
cond (str)
strategy (str)
params (Dict[str, Any])
invocations (int)
kind (str)
residual (Any)
time (float)
certificates (List[Dict[str, Any]])
messages (List[str])
progress (Tuple[int, int | None] | None)
failed (str | None)
inside_checks (int)
suppressed_messages (int)

loop: LoopInfo | None¶

cond: str¶

strategy: str¶

params: Dict[str, Any]¶

invocations: int = 0¶

kind: str = 'exact'¶

residual: Any = 0¶

time: float = 0.0¶

certificates: List[Dict[str, Any]]¶

messages: List[str]¶

progress: Tuple[int, int | None] | None = None¶

failed: str | None = None¶: Error message if the strategy failed (e.g. a refuted invariant).

inside_checks: int = 0¶: How many of the invocations happened inside the check of another loop’s invariant.

suppressed_messages: int = 0¶

MAX_MESSAGES = 4¶

add_message(message)[source]¶

Parameters:: message (str)
Return type:: None

to_json()[source]¶

Return type:: Dict[str, Any]

current_session()[source]¶

The session of the running analysis, or a detached interactive one.

Return type:: AnalysisSession

class AnalysisSession(program, *, source=None, config=None, provider=None, decisions=None, input_distribution=None, allow_files=True, capture_output=False, output_callback=None, progress_callback=None, record_steps=False, coefficient_limit=30)[source]¶

Bases: object

One analysis run with explicit decisions; see the module documentation.

Parameters:

program (Program | None)
source (str | None)
provider (DecisionProvider | None)
decisions (Mapping[Any, Any] | None)
input_distribution (Any)
allow_files (bool)
capture_output (bool)
output_callback (Callable[[str], None] | None)
progress_callback (Callable[[Dict[str, Any]], None] | None)
record_steps (bool)
coefficient_limit (int)

loop_records: Dict[Any, LoopRecord]¶

queries: List[Any]¶

steps: List[Dict[str, Any]]¶

warnings: List[str]¶

classmethod detached()[source]¶

Session for code that runs handlers without setting up a session.

Return type:: AnalysisSession

classmethod interactive(program, *, source=None, config=None, decisions=None, **kwargs)[source]¶

Parameters:

program (Program)
source (str | None)
decisions (Mapping[Any, Any] | None)

Return type:

AnalysisSession

activate()[source]¶

Makes this the current session (see current_session()).

Return type:: Iterator[AnalysisSession]

decide_loop(loop, instruction)[source]¶

Parameters:

loop (LoopInfo | None)
instruction (WhileInstr)

Return type:

LoopDecision

ask_param(task, spec)[source]¶

Parameters:

task (LoopTask)
spec (ParamSpec)

Return type:

Any

resolve_alias(task, alias, raw)[source]¶

Turns an alternative parameter (e.g. a file name) into the main parameter.

Parameters:

task (LoopTask)
alias (ParamSpec)
raw (Any)

Return type:

Any

property scope: _Scope¶

property silent: bool¶

sub_analysis(reason)[source]¶

Runs nested analyses whose distributions are not the program’s (second-order test distributions, invariant candidates, …). Their loops are still decided as usual, but neither their queries nor their residual mass count for the program. The yielded scope tells afterwards whether everything in it was exact.

Parameters:: reason (str)
Return type:: Iterator[_Scope]

on_normalize(error_prob)[source]¶

Called right before the distribution is conditioned on the observations.

Return type:: None

property collect_details: bool¶: Whether to compute extra data for front ends (coefficient tables etc.).

property bound_kind: str¶

property residual: Any¶

run_loop(task)[source]¶

Runs task.strategy and books its outcome (used by the while handler).

Parameters:: task (LoopTask)
Return type:: LoopOutcome

on_instruction(instruction)[source]¶

Called for every instruction before it is executed (see compute_semantics).

Parameters:: instruction (Any)
Return type:: None

loop_query_bounds(kind, expression)[source]¶

Bounds that the strategy of the loop right before this query computed for it.

Parameters:

kind (str)
expression (str)

Return type:

Tuple[Any, Any] | None

note(task, message)[source]¶

Parameters:

task (LoopTask)
message (str)

Return type:

None

warn(message)[source]¶

Parameters:: message (str)
Return type:: None

progress(task, done, total, message='')[source]¶

Parameters:

task (LoopTask)
done (int)
total (int | None)
message (str)

Return type:

None

next_query_position()[source]¶

Return type:: Tuple[int | None, int | None]

record_query(query)[source]¶

Books a QueryResult (ignored in sub-analyses).

Return type:: None

record_step(instruction, distribution, error_prob)[source]¶

Parameters:: instruction (SkipInstr | AbortInstr | WhileInstr | IfInstr | AsgnInstr | ChoiceInstr | LoopInstr | TickInstr | ObserveInstr | QueryInstr | ProbabilityQueryInstr | ExpectationInstr | PlotInstr | PrintInstr | OptimizationQuery)
Return type:: None

run()[source]¶: Runs the analysis; never raises for analysis errors (they end up in the result).