Compile

Compile systems into problems of one mathematical form with fast numeric functions.

BaseCompiler(backend=None, solver=None, observables=())

Bases: BaseConfig[System, Mapping[str, object]]

Compiler of a system into a problem of one mathematical form.

prepare(system) validates the settings and the system and sets report_, the issues it finds, and run(initial_state) raises the errors of the report and sets problem_, the compiled problem, and solver_, the solver the problem runs on. A subclass implements the hooks _issues and _compile, and a compiler of a form the engine does not know names the solver of its problems in _get_solver.

Parameters:

Name	Type	Description	Default
backend	BaseBackend | None	The backend instance, or None for the backend that suits the size of the system.	None
solver	object	The solver the problem runs on, or None for the default solver of its form, whose limits compiling checks.	None
observables	Sequence[ObservableSpec]	The observables to compute at every frame, labelled by their path from the root, such as village/mean_wealth for an observable of the nested system village.	()

Attributes:

Name	Type	Description
system_	System	The prepared system, with the observables of the run.
report_	ValidationReportSpec	The issues of the prepared system.
problem_	BaseProblemSpec	The compiled problem.
solver_	object	The solver the compiled problem runs on: the solver setting, the solver the compiler names, or None for the default solver of the problem's form.

Source code in src/skcomplex/core/engine/compile/_base.py

def __init__(
    self,
    backend: BaseBackend | None = None,
    solver: object = None,
    observables: Sequence[ObservableSpec] = (),
) -> None:
    self.backend = backend
    self.solver = solver
    self.observables = observables

BaseLowering

Bases: BaseConfig[Compilation, LoweredT]

Lowering of one family of laws, which one pass creates, later passes finish, and the compilation keeps.

The constructor stores what the lowering reads, prepare(compilation) sets the prepared attributes the lowering gathers, and run(laws) sets lowered_, the laws lowered. A subclass implements the hooks _prepare and _run.

A lowering that reads new symbols into laws notes their references with _add_reference, reads them back with _reference_of, and _apply_references replaces every spec it noted references for by a copy that holds them, in every field of the compilation.

Attributes:

Name	Type	Description
compilation_	Compilation	The compilation the lowering serves.
added_	dict[int, tuple[KernelSpec, dict[Expression, ReferenceSpec]]]	The spec and the references the lowering added to it, by the identity of the spec.
lowered_	LoweredT	The laws the last run lowered, in the shape of its input.

BasePass

Bases: BaseConfig[Compilation, None]

Step of a compilation that reads fields of the compilation and writes others.

A pass declares the fields it reads and writes through the hooks _get_reads and _get_writes, which order the passes of a compilation. prepare(compilation) takes the compilation the pass serves, and run() calls the hook _run, which reads the fields the pass reads from compilation_ and writes the fields it writes. A pass takes no input to run: the compilation it changes is its prepared target.

Attributes:

Name	Type	Description
compilation_	Compilation	The compilation the pass reads and writes.

Compilation(system, backend=None, solver=None, owners=None, held=())

State of one compilation of a system, whose fields its passes read and write in order.

The constructor stores the inputs of the compilation, and the passes set every other field, from nodes to problem in the order they are listed. A field that no pass has set yet is None. A pass may also declare hidden states and parameters on the layout, which only grows. Kernel specs are frozen: a pass that adds references to a spec builds a copy, which _replace_specs puts in every field that holds the spec.

Parameters:

Name	Type	Description	Default
system	System	The system, with the observables of the run.	required
backend	BaseBackend | None	The backend instance, or None for the backend that suits the size of the system.	None
solver	object	The solver instance, or None for the default solver of the system's form.	None
owners	frozenset[str] | None	The paths of the elements and groups whose laws a partition keeps, or None to keep every law.	None
held	tuple[str, ...]	The paths of the observables that roles bind to and laws read only at synchronisation points.	()

Attributes:

Name	Type	Description
nodes	list[NodeSpec] | None	The elements, groups, and interactions of the system tree, flattened once.
census	CensusSpec | None	The laws every component declares, with their owners.
plan	_PlanSpec | None	The partitions of the system.
layout	DeclarationsSpec | None	The states and parameters of every component, with the blocks of their symbols.
specs	list[tuple[KernelSpec, BlockSpec]] | None	The kernel spec of every component with its block.
collected	BlockLaws | None	The laws of every kernel spec as declared, with its block.
apart	Laws | None	The laws set apart from the kernels of their spec, which a pass of their own compiles.
programs	Laws | None	The linear program laws of every kernel spec, which never reach a kernel of their spec.
lowered	BlockLaws | None	The laws of every kernel spec with complex states split and Markov laws taken out.
markov	list[tuple[KernelSpec, MatrixLawSpec]] | None	The Markov laws of every kernel spec.
matrices	tuple[MarkovKernel, ...] | None	The Markov matrices, checked at the initial parameters.
ordered	Laws | None	The laws with derivatives and lags read through companion states and equations solved.
delayed	Laws | None	The laws with delayed states read through hidden parameters and events taken out.
corrected	Laws | None	The laws with the drift correction of every noise law of a convention.
shifted	Laws | None	The laws with member offset reads lowered to neighbour reads.
drawn	Laws | None	The laws of the owners of the compilation, masked to the rows present, with draws read from parameters.
drawn_apart	Laws | None	The laws set apart of the owners of the compilation.
kinds	dict[str, set[type]] | None	The law classes of every owner.
draws	DrawKernel | None	The kernel of the draws of the laws.
form	FormKind | None	The form of the problem.
reduced	Laws | None	The laws with every reduction read through its placeholder.
reduced_apart	Laws | None	The laws set apart, with every reduction read through its placeholder.
reductions	ReductionScheduleSpec | None	The reductions the laws read.
kernels	LoweredKernelsSpec | None	The kernels of every law kind.
functions	_ProblemFunctionsSpec | None	The numeric functions, the equations, and the backends of the problem.
jacobian	JacobianKernel | None	The exact Jacobian of the drift, or None.
problem	ProblemSpec | None	The compiled problem.
delays	list[_DelayedStateSpec] | None	The delayed states with their hidden parameters.
placeholders	dict[tuple[str, int, str, int, str, ReadKind], Expression] | None	The placeholder of every delayed state read.
events	list[tuple[KernelSpec, CrossingLaw]] | None	The events of every kernel spec.
renewals	list[BoundRenewalSpec] | None	The renewal laws with the hidden states that count them.
outputs	dict[int, list[OutputSpec]] | None	The outputs of the laws that a pass of a formalism lowers itself, by the identity of the law, which the kernels take in place of the outputs of the law's value.
written	dict[tuple[str, int], list[tuple[Expression, Expression]]] | None	The operators of fields as written, by state and column, with their stencils.
limits	list[float] | None	The stable steps of explicit field operators.
volume_limits	list[DiffusiveLimitSpec] | None	The stable steps of finite volumes, as functions of the parameters.
linear	list[LinearTermSpec] | None	The linear terms with constant coefficients of the rates of field states.
volumes	tuple[FiniteVolumeKernel, ...] | None	The conservation laws of fields.
symmetries	tuple[SymmetrySpec, ...] | None	The mirror symmetries of the field states.
constrained	tuple[NDArray[intp], NDArray[float64]] | None	The indices and values of the fixed nodes of meshes.
encounters	tuple[EncounterKernel, ...] | None	The encounter laws, lowered for their members.
transfer	TransferStateSpec | None	The state of the lowering of flows and delays, which one pass creates and a later pass finishes.
transfer_network	TransferNetworkSpec | None	The flows and stocks of the system.
rules	list[tuple[KernelSpec, tuple[RuleLaw, ...]]] | None	The rules of every kernel spec, taken out of the reduced laws.
request_jumps	dict[int, tuple[KernelSpec, TopologyRequestSpec]] | None	The jumps that request topology changes, by the identity of the law, with their spec and request.
topology	TopologyKernel | None	The topology changes of the networks, or None.
done	set[object] | None	The passes that ran.

Source code in src/skcomplex/core/engine/compile/_compilation.py

def __init__(
    self,
    system: System,
    backend: BaseBackend | None = None,
    solver: object = None,
    owners: frozenset[str] | None = None,
    held: tuple[str, ...] = (),
) -> None:
    self.system = system
    self.backend = backend
    self.solver = solver
    self.owners = owners
    self.held = held

FlattenPass

Bases: BasePass

Pass that flattens the system tree into its nodes.

HilbertSpaceError

Bases: SkcomplexError

An operator system has a Hilbert space larger than its declared limit.

LayoutPass

Bases: BasePass

Pass that declares every component in tree order and checks that no two variables share a symbol.

ObservableEvaluator(problem)

Evaluator of every observable of a problem from a state vector, a parameter vector, and the agents.

Preparing compiles every observable into a getter, in declaration order, the prepared getters_.

Parameters:

Name	Type	Description	Default
problem	BaseProblemSpec	The compiled problem.	required

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.core.engine.compile import ObservableEvaluator, PassCompiler
>>> from skcomplex.core.specs import ObservableSpec
>>> from skcomplex.core.components.system import System
>>> from skcomplex.systems.biology.common.populations import PopulationElement
>>> system = System(
...     elements=[GroupElement('herd', element=PopulationElement, size=2, values={'n': [2.0, 6.0]})]
... )
>>> total = ObservableSpec('total', group='herd', reduce='sum', variable='n')
>>> problem = PassCompiler(observables=[total]).prepare(system).run().problem_
>>> ObservableEvaluator(problem).prepare().evaluate(0.0, problem.initial_state, problem.initial_parameters)
{'total': 8.0}

Source code in src/skcomplex/core/engine/compile/_observables.py

def __init__(self, problem: BaseProblemSpec) -> None:
    self.problem = problem

evaluate(t, x, p, agents_of=None)

Return the value of every observable by path.

Parameters:

Name	Type	Description	Default
t	float	The time.	required
x	Array	The state vector.	required
p	Array	The parameter vector.	required
agents_of	AgentsFunction | None	A function that returns the views of the agents of an agent group by path, or None without agents.	None

Returns:

Name	Type	Description
values	dict[str, object]	A mapping from observable path to its value.

Source code in src/skcomplex/core/engine/compile/_observables.py

def evaluate(self, t: float, x: Array, p: Array, agents_of: AgentsFunction | None = None) -> dict[str, object]:
    """Return the value of every observable by path.

    Args:
        t:
            The time.

        x:
            The state vector.

        p:
            The parameter vector.

        agents_of:
            A function that returns the views of the agents of an agent group by path, or `None` without agents.

    Returns:
        values:
            A mapping from observable path to its value.
    """
    done: dict[str, object] = {'__agents__': agents_of or (lambda group: None)}
    for path, get in self.getters_.items():
        done[path] = get(t, x, p, done)
    del done['__agents__']
    return done

prepare()

Compile every observable into a getter, in declaration order.

Returns:

Name	Type	Description
evaluator	Self	The evaluator, with the getter of every observable path in getters_.

Raises:

Type	Description
ValidationError	Two observables share a path, an observable has neither an expression nor a group, or an expression reads a symbol it cannot.

Source code in src/skcomplex/core/engine/compile/_observables.py

def prepare(self) -> Self:
    """Compile every observable into a getter, in declaration order.

    Returns:
        evaluator:
            The evaluator, with the getter of every observable path in `getters_`.

    Raises:
        ValidationError:
            Two observables share a path, an observable has neither an expression nor a group, or an expression
            reads a symbol it cannot.
    """
    self.getters_: dict[str, ObservableFunction] = {}
    if isinstance(self.problem, OperatorProblemSpec):
        self._scalars = {}
        self.getters_ = self._operator_getters(self.problem)
        return self
    self._scalars = self._scalar_sources()
    for scope, observable in self.problem.observables:
        path = self._join(scope, observable.label)
        if path in self.getters_:
            error_msg = f'Two observables have the path {path}. Give observables of one system distinct labels.'
            raise ValidationError(error_msg)
        if observable.function is not None:
            self.getters_[path] = self._function_getter(scope, path, observable.function)
            continue
        if observable.group is None and observable.expression is None:
            error_msg = (
                f'ObservableSpec {path} has neither an expression, a function, nor a group. Give one of them.'
            )
            raise ValidationError(error_msg)
        observed = self.problem.observed if isinstance(self.problem, ProblemSpec) else {}
        if observable.group is None:
            self.getters_[path] = self._expression_getter(scope, path, observable)
        elif path in observed:
            self.getters_[path] = self._observed_getter(observed[path], observable)
        else:
            self.getters_[path] = self._reduction_getter(scope, path, observable)
    self._add_law_records()
    return self

OperatorCompiler(backend=None, solver=None, observables=(), max_dimension=DEFAULT_MAX_DIMENSION)

Bases: BaseCompiler

Compiler of a system holding operator elements into a problem of the operator form.

Parameters:

Name	Type	Description	Default
backend	BaseBackend | None	The backend requested for the problem, which the operator form does not use.	None
solver	object	The solver the problem runs on. A solver with a largest dimension of its own limits the Hilbert space.	None
observables	Sequence[ObservableSpec]	The observables to compute at every frame, labelled by their path from the root.	()
max_dimension	int	The largest dimension of the Hilbert space, for a solver without a largest dimension of its own.	DEFAULT_MAX_DIMENSION

Examples:

>>> from skcomplex.core.components.system import System
>>> from skcomplex.core.engine.compile import OperatorCompiler
>>> from skcomplex.systems.biology.common.populations import PopulationElement
>>> OperatorCompiler().prepare(System(elements=[PopulationElement('rabbits')])).report_.ok
False

Source code in src/skcomplex/core/engine/compile/passes/_operator_compiler.py

def __init__(
    self,
    backend: BaseBackend | None = None,
    solver: object = None,
    observables: Sequence[ObservableSpec] = (),
    max_dimension: int = DEFAULT_MAX_DIMENSION,
) -> None:
    super().__init__(backend, solver, observables)
    self.max_dimension = max_dimension

PassCompiler(passes=None, backend=None, solver=None, observables=(), partition='plan')

Bases: BaseCompiler

General compiler of a system, which runs passes over a compilation in the order of the fields they share.

prepare(system) validates the system: it compares the requirements every law declares with the states it targets, the space it lives on, and the system's time, checks that one solver runs the laws of every partition, and checks the solver and the backend against their declared support. A state of a fractional order needs an order in (0, 1), continuous time, and a solver and a backend that integrate fractional orders. It reports every issue at once in report_, each with its path and fix, and keeps the compilation planned so far in compilation_, which run continues from. Elements and groups are compiled once each, whatever the size of a group. Laws written for one member are evaluated for every member, and pair interactions are evaluated for every pair within their cutoff.

Parameters:

Name	Type	Description	Default
passes	Sequence[BasePass] | None	The passes, in the order they are registered, or None for every pass of the library. The compiler runs copies of them.	None
backend	BaseBackend | None	The backend instance, or None for NumbaBackend on systems whose groups hold at least 1000 members in total and NumpyBackend on the others.	None
solver	object	The solver the problem runs on, or None for the default solver of its form.	None
observables	Sequence[ObservableSpec]	The observables to compute at every frame, labelled by their path from the root.	()
partition	str	'plan' to compile one problem, or one partition per planned subsystem when no single form runs the system, or 'subsystems' to compile one partition for the elements of the root and one per child system, so a co-simulation can advance them with separate solvers.	'plan'

Attributes:

Name	Type	Description
passes_	tuple[BasePass, ...]	The copies of the passes the compiler runs.
compilation_	Compilation | None	The compilation planned so far, or None when the system tree cannot be flattened or trains a nested system.
learning_	_LearningCompiler | None	The compiler of a system that holds a TrainingInteraction, which compiles its nested system into the inner problem of a LearningProblemSpec, or None for other systems.

Examples:

>>> from skcomplex.core.components.system import System
>>> from skcomplex.core.engine.compile import PassCompiler
>>> from skcomplex.systems.biology.common.populations import ExponentialGrowthInteraction, PopulationElement
>>> system = System(
...     elements=[PopulationElement('rabbits', n=10)],
...     interactions=[ExponentialGrowthInteraction('growth', target='rabbits', r=0.5)],
... )
>>> compiler = PassCompiler().prepare(system)
>>> compiler.report_.ok
True
>>> compiler.run().problem_.equations
{'rabbits.n': n_rabbits*r_growth}

Source code in src/skcomplex/core/engine/compile/_pass_compiler.py

def __init__(
    self,
    passes: Sequence[BasePass] | None = None,
    backend: BaseBackend | None = None,
    solver: object = None,
    observables: Sequence[ObservableSpec] = (),
    partition: str = 'plan',
) -> None:
    super().__init__(backend, solver, observables)
    self.passes = passes
    self.partition = partition

ProblemPass

Bases: BasePass

Pass that assembles the problem of the general parts of a compilation.

RewritingCompiler(backend=None, solver=None, observables=())

Bases: BaseCompiler

Compiler of a system of words or patches into a problem of the rewriting form.

The backend, the solver, and the initial state do not change a problem of the rewriting form.

Examples:

>>> from skcomplex.core.components.system import System
>>> from skcomplex.core.engine.compile import RewritingCompiler
>>> from skcomplex.systems.biology.common.populations import PopulationElement
>>> RewritingCompiler().prepare(System(elements=[PopulationElement('rabbits')])).report_.ok
False

Source code in src/skcomplex/core/engine/compile/_base.py

def __init__(
    self,
    backend: BaseBackend | None = None,
    solver: object = None,
    observables: Sequence[ObservableSpec] = (),
) -> None:
    self.backend = backend
    self.solver = solver
    self.observables = observables