# SPDX-FileCopyrightText: 2026 John C. Travers and contributors
#
# SPDX-License-Identifier: GPL-3.0-or-later
"""Recipes: a simulation as versioned TOML, read, built, run and saved (SEP 41, S1).
A recipe is read, checked against the parameter metadata, built into a
:class:`~solprop.driver.simulation.Simulation`, run, written back and exported as a
plain Python script. It is one-way: objects are built from it and never turned back
into tables, so a saved recipe always comes from a recipe or from plain values
(:meth:`Recipe.from_keywords`).
"""
import copy
import dataclasses
import importlib.resources
import math
import warnings
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Final, Literal, cast
import numpy as np
from solprop.exceptions import SolpropError, SolpropWarning
from solprop.output.schema import RecipeRecord
from solprop.recipe.document import (
FORMAT,
NONE,
Document,
decode_integer,
dump_text,
encode_integer,
parse_text,
unknown_key,
validate_document,
)
from solprop.recipe.errors import RecipeError
from solprop.recipe.metadata import (
OBJECT_ROUTES,
REQUIRED,
SETTINGS_KEYWORDS,
Parameter,
parameters,
surface,
)
from solprop.recipe.paths import format_path, get_path, parse_path, set_path
from solprop.recipe.quantity import Unit, format_quantity, parse_quantity
from solprop.recipe.registry import REGISTRY, BuildContext, coerce, decode_keyword
if TYPE_CHECKING:
from numpy.typing import ArrayLike
from solprop.driver.scan import Scan, ScanPoint
from solprop.driver.simulation import Simulation
from solprop.grid import TimeGrid
from solprop.log import ProgressChoice, ProgressReporter, StopFlag
from solprop.output.schema import Output
from solprop.recipe.pins import Pin
from solprop.recipe.preview import InputPreview, ModeSetPreview
from solprop.tools import DesignRoots
__all__ = [
"ROUTES",
"Recipe",
"RecipeBuilder",
"RecipeIssue",
"RecipeScan",
"resume_file",
]
ROUTES: Final = ("gnlse", "capillary", "waveguide", "radial", "free", "planar")
"""The routes a recipe of format 1 describes (SEP 41).
The keyword routes ``gnlse``, ``capillary`` and ``waveguide``, and the object routes
(:data:`~solprop.recipe.metadata.OBJECT_ROUTES`), whose keywords are tables of grids,
medium, responses, beam and settings.
"""
_EXAMPLES: Final = "examples"
"""The package directory that holds the example recipes."""
_SCAN_EXECUTION: Final = ("fields", "field_planes")
"""The ``[execution]`` keys ``run_scan`` takes, rather than each point's settings."""
[docs]
@dataclass(frozen=True)
class RecipeIssue:
"""One problem :meth:`Recipe.check` found, placed at the parameter it concerns.
Attributes
----------
message : str
What is wrong, in the words of the refusal or warning.
parameter : str or tuple of str or None
The path of the value at fault, or the paths of a conflict; ``None`` when the
issue concerns no single value.
severity : {'error', 'warning'}
Whether the recipe cannot be built, or builds with a warning.
rule : str or None
The identifier of the combination rule a refusal enforces, where it is one.
Examples
--------
>>> from solprop.recipe import RecipeIssue
>>> RecipeIssue("must be positive", "pressure", "error").severity
'error'
"""
message: str
parameter: str | tuple[str, ...] | None
severity: Literal["error", "warning"]
rule: str | None = None
[docs]
class Recipe:
"""A simulation written as TOML: read, checked, built, run, saved and exported.
Construct one with :meth:`load`, :meth:`loads`, :meth:`from_dict`,
:meth:`from_keywords` or :meth:`example`, not directly.
Parameters
----------
document : collections.abc.Mapping
The recipe's tables, as :func:`tomllib.loads` reads them.
text : str or None, optional
The text it was read from, kept so that an unedited recipe saves verbatim.
base : pathlib.Path or None, optional
The directory its data files resolve against.
allow_outside : bool, optional
Whether a data file may lie outside `base`.
Attributes
----------
route : str
The route the recipe builds.
Examples
--------
>>> from solprop.recipe import Recipe
>>> recipe = Recipe.loads('''
... [recipe]
... format = 1
... route = "gnlse"
...
... [build]
... gamma = 0.11
... length = "1 cm"
... betas = [0, 0, "-0.01183 ps^2/m"]
... wavelength = "835 nm"
... wavelength_limits = ["500 nm", "2 um"]
... trange = "1 ps"
... width = "100 fs"
... peak_power = "9.8 W"
... shape = "sech"
... noise = "none"
... ''')
>>> recipe.route, recipe.get("length")
('gnlse', '1 cm')
>>> recipe.set("length", "2 cm")
>>> recipe.get("length")
'2 cm'
"""
def __init__(
self,
document: Mapping[str, object],
*,
text: str | None = None,
base: Path | None = None,
allow_outside: bool = False,
) -> None:
"""Adopt a checked document."""
self._document: Document = validate_document(document)
self._text = text
self._base = base
self._allow_outside = allow_outside
# Data files a run's output embedded, by the name the recipe refers to them by,
# and the generated name each is filed under (from_output).
self._embedded: dict[str, bytes] = {}
self._filed: dict[str, str] = {}
# The authored recipe a resolved one was solved from (resolve), whose text a
# run records beside the resolved one.
self._authored: Recipe | None = None
self.route = str(self._document["recipe"]["route"])
if self.route not in ROUTES:
raise unknown_key(
self.route, ROUTES, where="the routes a recipe describes", path="route"
)
self._check_keys()
# ------------------------------------------------------------------ construction
[docs]
@classmethod
def loads(
cls, text: str, *, base: Path | None = None, allow_outside: bool = False
) -> "Recipe": # noqa: UP037 - the class, checked at run time
"""Read a recipe from TOML text.
Parameters
----------
text : str
The recipe.
base : pathlib.Path or None, optional
The directory its data files resolve against; ``None`` allows none.
allow_outside : bool, optional
Whether a data file may lie outside `base`.
Returns
-------
Recipe
The recipe, which saves as `text` until it is edited.
"""
return cls(parse_text(text), text=text, base=base, allow_outside=allow_outside)
[docs]
@classmethod
def load(cls, path: str | Path, *, allow_outside: bool = False) -> "Recipe": # noqa: UP037 - the class, checked at run time
"""Read a recipe file.
Parameters
----------
path : str or pathlib.Path
The ``.toml`` file.
allow_outside : bool, optional
Whether a data file it refers to may lie outside its directory.
Returns
-------
Recipe
The recipe, whose data files resolve against the file's directory.
"""
path = Path(path)
text = path.read_text(encoding="utf-8")
return cls.loads(text, base=path.resolve().parent, allow_outside=allow_outside)
[docs]
@classmethod
def from_dict(
cls, document: Mapping[str, object], *, base: Path | None = None
) -> "Recipe": # noqa: UP037 - the class, checked at run time
"""Build a recipe from its tables.
Parameters
----------
document : collections.abc.Mapping
The tables, as TOML would read them.
base : pathlib.Path or None, optional
The directory its data files resolve against.
Returns
-------
Recipe
The recipe.
"""
return cls(copy.deepcopy(dict(document)), base=base)
[docs]
@classmethod
def from_keywords(
cls, route: str, *, title: str | None = None, **keywords: object
) -> "Recipe": # noqa: UP037 - the class, checked at run time
"""Write a route's keyword call as a recipe, from plain values only.
Parameters
----------
route : str
The route, one of :data:`ROUTES`.
title : str or None, optional
The recipe's title.
**keywords : object
The route's build keywords, as numbers in their SI unit (pressure in bar),
strings, booleans, and lists and mappings of these.
Returns
-------
Recipe
The recipe.
Raises
------
RecipeError
If a value is an object or a callable, which a recipe holds only as a table
with a ``type`` key, or through the Python export.
Examples
--------
>>> from solprop.recipe import Recipe
>>> recipe = Recipe.from_keywords(
... "capillary", radius=125e-6, length=1.0, gas="He", pressure=1.2,
... wavelength=800e-9, fwhm=7.5e-15, energy=275e-6,
... )
>>> recipe.get("radius")
0.000125
"""
build = {name: _plain(value, name) for name, value in keywords.items()}
header: dict[str, object] = {"format": FORMAT, "route": route}
if title is not None:
header["title"] = title
return cls({"recipe": header, "build": build})
[docs]
@classmethod
def examples(cls) -> tuple[str, ...]:
"""Return the names of the example recipes solprop ships.
Returns
-------
tuple of str
The names :meth:`example` takes, sorted.
Examples
--------
>>> from solprop.recipe import Recipe
>>> "rdw" in Recipe.examples()
True
"""
root = importlib.resources.files("solprop.recipe").joinpath(_EXAMPLES)
return tuple(
sorted(
entry.name.removesuffix(".toml")
for entry in root.iterdir()
if entry.name.endswith(".toml")
)
)
[docs]
@classmethod
def example(cls, name: str) -> "Recipe": # noqa: UP037 - checked at run time
"""Load one of the example recipes solprop ships, each a gallery example's.
Parameters
----------
name : str
The example's name, one of :meth:`examples`.
Returns
-------
Recipe
The recipe.
Raises
------
RecipeError
If no example has that name.
"""
if name not in cls.examples():
raise unknown_key(name, cls.examples(), where="the example recipes")
root = importlib.resources.files("solprop.recipe").joinpath(_EXAMPLES)
with importlib.resources.as_file(root.joinpath(f"{name}.toml")) as path:
return cls.load(path)
[docs]
@classmethod
def from_output(cls, path: str | Path, *, resolved: bool = False) -> "Recipe": # noqa: UP037 - the class, checked at run time
"""Read back the recipe a run's output records (SEP 41, S2).
Parameters
----------
path : str or pathlib.Path
The HDF5 output of a run, or a scan's file, which records the authored
recipe only.
resolved : bool, optional
Read the resolved recipe, which rebuilds the run as it ran, rather than
the authored one, for editing.
Returns
-------
Recipe
The recipe, holding the data files the output embedded.
Raises
------
RecipeError
If the file records no recipe, being a file from a script, or its recipe
is of a newer format.
"""
import h5py # noqa: PLC0415 - only to tell a scan file from a run's
from solprop.output.hdf5 import open_file # noqa: PLC0415 - layer above
from solprop.processing.scans import ScanOutput # noqa: PLC0415 - layer above
with h5py.File(path, "r") as file:
scan = "scan_schema_version" in file.attrs
if scan:
# Through the scan reader's own check of the schema its records declare.
record = ScanOutput(path).recipe
else:
with open_file(path) as handler:
record = handler.recipe
if record is None:
msg = (
f"{path} records no recipe: it was written from a script, which is "
"what resumes or rebuilds it"
)
raise RecipeError(msg)
text = record.resolved if resolved else record.authored
recipe = cls.loads(text)
for name, entry in record.files.items():
recipe._filed[entry.path] = name
if entry.data is not None:
recipe._embedded[entry.path] = entry.data
return recipe
# ------------------------------------------------------------------ the values
[docs]
@property
def title(self) -> str | None:
"""Return the recipe's title.
Returns
-------
str or None
The ``[recipe]`` table's title, or ``None``.
"""
title = self._document["recipe"].get("title")
return None if title is None else str(title)
[docs]
def to_dict(self) -> Document:
"""Return a copy of the recipe's tables.
Returns
-------
dict
The five tables that are present, as TOML would read them.
"""
return copy.deepcopy(self._document)
def __eq__(self, other: object) -> bool:
"""Compare two recipes by their values, not their text or spelling.
Parameters
----------
other : object
The other recipe.
Returns
-------
bool
Whether the tables are equal.
"""
if not isinstance(other, Recipe):
return NotImplemented
return self._canonical_document() == other._canonical_document()
def _canonical_document(self) -> Document:
"""Return the tables with each quantity spelled in its display unit.
Returns
-------
dict
The document a canonical save writes, so that two spellings of one value,
``"1 cm"`` and ``0.01``, compare equal.
"""
document = self.to_dict()
table = self._table()
document["build"] = {
name: _canonical(table[name], value)
for name, value in document["build"].items()
}
return document
__hash__ = None # type: ignore[assignment]
def __repr__(self) -> str:
"""Return the route and the title.
Returns
-------
str
For example ``Recipe(capillary, 'RDW in a helium capillary')``.
"""
title = "" if self.title is None else f", {self.title!r}"
return f"Recipe({self.route}{title})"
[docs]
def get(self, path: str) -> object:
"""Return a value of the ``[build]`` table as the recipe writes it.
Parameters
----------
path : str
A parameter path, as ``"pressure"`` or ``"pulses[1].energy"``.
Returns
-------
object
The value as written: a number, a quantity string, a list or a table.
"""
return get_path(self._document["build"], path)
[docs]
def set(self, path: str, value: object) -> None:
"""Set a value of the ``[build]`` table, checked against the metadata.
Parameters
----------
path : str
A parameter path; the keyword it starts with must be the route's.
value : object
A number in the keyword's unit, a quantity string, a boolean, a string,
or a list or table of these.
Raises
------
RecipeError
If the keyword is not the route's, or the value does not decode.
"""
steps = parse_path(path)
build = set_path(self._document["build"], steps, _plain(value, path))
keyword = format_path(steps[:1])
context = self._context()
self._decode(keyword, build[keyword], _deferred(context))
self._document["build"] = build
self._text = None
# ------------------------------------------------------------------ checking
def _table(self) -> Mapping[str, Parameter]:
"""Return the route's parameter metadata.
Returns
-------
collections.abc.Mapping
:class:`~solprop.recipe.metadata.Parameter` by keyword.
"""
return parameters(self.route)
def _check_keys(self) -> None:
"""Refuse a build or execution key the route does not have.
Raises
------
RecipeError
Naming the key and the nearest valid one.
"""
table = self._table()
for key in self._document["build"]:
if key in SETTINGS_KEYWORDS and self.route in OBJECT_ROUTES:
msg = (
f"on the {self.route} route a recipe gives {key} in the settings "
f"table, as settings.{key}"
)
raise RecipeError(msg, path=key)
if key not in table:
raise unknown_key(
key,
list(table),
where=f"the {self.route} route's keywords",
path=key,
)
def _decode(self, name: str, value: object, context: BuildContext) -> object:
"""Return one build keyword's value as the constructor takes it.
Parameters
----------
name : str
The keyword.
value : object
Its value as written.
context : BuildContext
The data files' directory and the route's grid.
Returns
-------
object
The decoded value.
"""
table = self._table()
if name not in table:
raise unknown_key(
name, list(table), where=f"the {self.route} route's keywords", path=name
)
decoded = decode_keyword(table[name], value, context)
return coerce(surface(self.route), name, decoded)
[docs]
def check(self) -> list[RecipeIssue]:
"""Find what stops the recipe building, or makes it warn (SEP 41, S10).
Every value is decoded against the metadata first, each problem reported at
its path. If none is refused, the simulation is built and translated, which is
where the refusals that depend on several values are found, stopping before a
backend is constructed; the first refusal is reported with its parameter, and
every warning of solprop's with its own. It costs a translation: call it on a
pause in typing, not on every keystroke.
Returns
-------
list of RecipeIssue
The issues, errors first; empty when the recipe builds cleanly.
"""
issues: list[RecipeIssue] = []
context = self._context()
for name, value in _with_pulse_placeholders(self._document).items():
try:
self._decode(name, value, _deferred(context))
except SolpropError as error:
# An object built from a table refuses by its own argument's name.
issues.append(_issue(error, name))
pinned = set(self._document.get("pins", {}))
missing = [
name
for name, entry in self._table().items()
if entry.default is REQUIRED
and name not in self._document["build"]
and name not in pinned
]
issues.extend(
RecipeIssue(f"{name} is required", name, "error") for name in missing
)
if issues:
return issues
try:
resolved = self.resolve()
except SolpropError as error:
return [_issue(error)]
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always", SolpropWarning)
try:
resolved.build().translate()
except SolpropError as error:
issues.append(_issue(error))
issues.extend(
RecipeIssue(
str(record.message),
getattr(record.message, "parameter", None),
"warning",
)
for record in caught
if isinstance(record.message, SolpropWarning)
)
return issues
# ------------------------------------------------------------------ building
[docs]
def keywords(
self,
*,
grid: "TimeGrid | None" = None, # noqa: UP037 - checked at run time
) -> dict[str, object]:
"""Return the route's build keywords as the constructor takes them.
Parameters
----------
grid : solprop.grid.TimeGrid or None, optional
The route's grid, for a value built on it; ``None`` leaves such values
out.
Returns
-------
dict
Keyword arguments for the route's ``Simulation.from_*``.
"""
build = self._document["build"]
if grid is None and self.route in OBJECT_ROUTES and "grid" in build:
# The time grid is a keyword of its own here: what is built on it, the
# settings' absorber, is built on that grid, in one pass.
grid = cast(
"TimeGrid", self._decode("grid", build["grid"], self._context())
)
context = self._context(grid)
out: dict[str, object] = {}
for name, value in build.items():
if name == "grid" and self.route in OBJECT_ROUTES:
out[name] = grid
continue
if grid is None and _needs_grid(value):
continue
out[name] = self._decode(name, value, context)
return out
[docs]
def build(self) -> "Simulation": # noqa: UP037 - checked at run time
"""Build the simulation the recipe describes, without running it.
Pins are solved first (:meth:`resolve`).
Returns
-------
solprop.driver.simulation.Simulation
The simulation, with ``[execution]``'s settings applied, carrying this
recipe, which its first run records in its output.
"""
if self._document.get("pins"):
return self.resolve().build()
builder = surface(self.route)
simulation = builder(**self.keywords())
if self.route not in OBJECT_ROUTES and any(
_needs_grid(value) for value in self._document["build"].values()
):
simulation = builder(**self.keywords(grid=simulation._base_grid))
execution = self._execution()
if execution:
simulation.settings = dataclasses.replace(simulation.settings, **execution)
# The simulation carries its recipe, which its first run records (S2).
simulation._recipe = self
return simulation
# ------------------------------------------------------------------ pins
[docs]
def resolve(self) -> "Recipe": # noqa: UP037 - the class, checked at run time
"""Return the recipe with each pin replaced by its solved value.
The pin on the guide or fill is solved first, on the launch mode at the
launch position; the pins on pulses then read the mode it settled
(:mod:`solprop.recipe.pins`). A run of a pinned recipe records both texts:
the authored one with its pins, and the resolved one.
Returns
-------
Recipe
A recipe without ``[pins]``, its pinned paths set in ``[build]`` as
numbers in their keywords' units; a copy of this one when it has no pins.
Raises
------
RecipeError
On the pinned path, if a pin does not read, or its target is not met in
its search range (naming the attainable range).
Examples
--------
>>> from solprop.recipe import Recipe
>>> recipe = Recipe.example("soliton_self_compression")
>>> document = recipe.to_dict()
>>> del document["build"]["energy"]
>>> document["pins"] = {"energy": {"target": "soliton_order", "order": 4}}
>>> pinned = Recipe.from_dict(document)
>>> f"{pinned.resolve().get('energy') * 1e6:.1f} uJ"
'126.1 uJ'
"""
from solprop.recipe.pins import read_pins # noqa: PLC0415 - cycle
pins = read_pins(self)
working = self._copy(("pins",))
for pin in pins:
if pin.stage == "pulse":
working._document["build"] = set_path(
working._document["build"],
parse_path(pin.path),
_placeholder(pin.path),
)
for pin in pins:
value = _pick(pin, working._solve(pin))
working._document["build"] = set_path(
working._document["build"], parse_path(pin.path), value
)
working._text = None
working._authored = self if pins else None
return working
[docs]
def pin_roots(self, path: str) -> "DesignRoots": # noqa: UP037 - TYPE_CHECKING
"""Return every root of one pin's target, for a front end to choose among.
A pin on a pulse reads the guide as its own pin (if any) settles it.
Parameters
----------
path : str
The pinned path, as ``[pins]`` writes it.
Returns
-------
solprop.tools.DesignRoots
The roots over the pin's search range, or none with the attainable range.
Raises
------
RecipeError
If the path is not pinned, or a pin does not read.
"""
from solprop.recipe.pins import read_pins # noqa: PLC0415 - cycle
pins = read_pins(self)
steps = format_path(parse_path(path))
chosen = next((pin for pin in pins if pin.path == steps), None)
if chosen is None:
msg = "this path is not pinned"
raise RecipeError(msg, path=path)
working = self._copy(("pins",))
for pin in pins:
if pin.stage == "pulse":
working._document["build"] = set_path(
working._document["build"],
parse_path(pin.path),
_placeholder(pin.path),
)
if chosen.stage == "pulse":
for pin in pins:
if pin.stage == "guide":
working._document["build"] = set_path(
working._document["build"],
parse_path(pin.path),
_pick(pin, working._solve(pin)),
)
return working._solve(chosen)
# ------------------------------------------------------------------ previews
[docs]
def modes(self, z: float | None = None) -> "ModeSetPreview": # noqa: UP037 - TYPE_CHECKING
"""Return the mode set a guided route would build, before the run (S18).
Labels, the effective index, ``beta`` to ``beta3``, loss, effective area and
nonlinear coefficient over the grid's band, and the zero-dispersion
wavelength; pins are solved first. The cost is the
mode solve (:func:`solprop.recipe.preview.modes`).
Parameters
----------
z : float or None, optional
The position, m; ``None`` for the launch position.
Returns
-------
solprop.recipe.preview.ModeSetPreview
The modes.
Examples
--------
>>> from solprop.recipe import Recipe
>>> preview = Recipe.example("rdw").modes()
>>> [mode.label for mode in preview.modes]
['HE1,1:even', 'HE1,2:even']
"""
from solprop.recipe.preview import modes # noqa: PLC0415 - layer D
return modes(self, z)
def _copy(self, drop: tuple[str, ...]) -> "Recipe": # noqa: UP037 - checked at run time
"""Return a copy without some top-level tables, its data files kept.
Parameters
----------
drop : tuple of str
The tables to leave out.
Returns
-------
Recipe
The copy.
"""
document = {
name: table for name, table in self.to_dict().items() if name not in drop
}
recipe = Recipe(document, base=self._base, allow_outside=self._allow_outside)
recipe._embedded = self._embedded
recipe._filed = self._filed
return recipe
def _solve(self, pin: "Pin") -> "DesignRoots": # noqa: UP037 - TYPE_CHECKING
"""Solve one pin on this (unpinned) recipe's keywords.
Parameters
----------
pin : solprop.recipe.pins.Pin
The pin.
Returns
-------
solprop.tools.DesignRoots
Its roots.
"""
from solprop.recipe.pins import ( # noqa: PLC0415 - cycle
design_target,
solve_pin,
solve_pulse,
)
keywords = self.keywords()
if any(_needs_grid(value) for value in self._document["build"].values()):
probe = self._copy(())
if pin.stage == "guide" and pin.search is not None:
low, high = pin.search
probe._document["build"][pin.path] = (
math.sqrt(low * high) if low > 0.0 else 0.5 * (low + high)
)
keywords = self.keywords(grid=probe.build()._base_grid)
if pin.stage == "pulse":
keywords.pop(pin.path, None)
return solve_pulse(
self.route,
keywords,
pin.path,
cast("float", pin.keys["order"]),
pin.search,
)
target = design_target(pin, keywords)
return solve_pin(
self.route,
keywords,
pin.path,
target,
cast("tuple[float, float]", pin.search),
)
[docs]
def record(
self,
simulation: "Simulation", # noqa: UP037 - checked at run time
) -> RecipeRecord:
"""Return the record a run built from this recipe keeps in its output.
Called by :meth:`~solprop.driver.simulation.Simulation.run` at the run's
start, with the settings it starts with.
Parameters
----------
simulation : solprop.driver.simulation.Simulation
The simulation this recipe built.
Returns
-------
solprop.output.schema.RecipeRecord
The authored and resolved texts, and the data files
(:mod:`solprop.recipe.provenance`).
"""
from solprop.recipe.provenance import record # noqa: PLC0415 - cycle
return record(
self, simulation, base=self._base, allow_outside=self._allow_outside
)
def _execution(self) -> dict[str, object]:
"""Return the ``[execution]`` table's values as the settings take them.
Returns
-------
dict
The values, ``"none"`` read as ``None`` where a keyword takes it.
"""
table = parameters("execution")
out: dict[str, object] = {}
for name, value in self._document.get("execution", {}).items():
out[name] = (
None
if table[name].none and value == NONE
else tuple(value)
if isinstance(value, list)
else value
)
return out
[docs]
def run(
self,
output: "str | Path | Output | None" = None, # noqa: UP037 - run time
**run_keywords: object,
) -> "Output": # noqa: UP037 - checked at run time
"""Build the simulation and run it, with ``[execution]`` applied.
Parameters
----------
output : str, pathlib.Path, Output or None, optional
Where to write, as :meth:`~solprop.driver.simulation.Simulation.run`
takes it; ``None`` keeps the run in memory.
**run_keywords : object
Further keywords of :meth:`~solprop.driver.simulation.Simulation.run`,
such as ``progress`` or ``resume``.
Returns
-------
solprop.output.schema.Output
The run's output.
"""
return self.build().run(output, **run_keywords) # type: ignore[arg-type]
# ------------------------------------------------------------------ scanning
[docs]
def scan(self) -> "RecipeScan": # noqa: UP037 - the class, checked at run time
"""Return the scan the ``[scan]`` table declares, ready for ``run_scan``.
``[scan]`` requires ``name``, which names the durable files and derives the
points' seeds, and may give ``base_seed``; its other keys are parameter paths,
each with a list of values or ``{start, stop, num, spacing}``, spacing
``"linear"`` (the default) or ``"log"``. Values are read in the unit of the
keyword each path ends on.
Returns
-------
RecipeScan
The :class:`~solprop.driver.scan.Scan`, a picklable builder that sets each
point's values, passes ``point.seed`` and builds, and the keywords
:func:`~solprop.driver.scan.run_scan` takes from ``[execution]``.
Raises
------
RecipeError
If the recipe has no ``[scan]`` table, its ``name`` is missing, or a path
or a value does not read.
Examples
--------
>>> from solprop.recipe import Recipe
>>> recipe = Recipe.example("soliton")
>>> document = recipe.to_dict()
>>> document["scan"] = {"name": "power", "peak_power": ["5 W", "10 W"]}
>>> scan = Recipe.from_dict(document).scan()
>>> scan.scan.shape, scan.scan.variables["peak_power"].tolist()
((2,), [5.0, 10.0])
"""
from solprop.driver.scan import Scan # noqa: PLC0415 - layer D
table = dict(self._document.get("scan", {}))
if not table:
msg = "the recipe has no [scan] table"
raise RecipeError(msg)
name = table.pop("name", None)
if not isinstance(name, str):
msg = "[scan] needs a name, which names the scan's files and its seeds"
raise RecipeError(msg, path="name")
base_seed = table.pop("base_seed", None)
scan = Scan(
name,
base_seed=None if base_seed is None else decode_integer(base_seed),
)
for path, values in table.items():
scan.add(path, cast("ArrayLike", self._axis(path, values)))
document = self.to_dict()
document.pop("scan", None)
# What `run_scan` takes is its own, and a builder that sets it is refused; the
# rest of the table is the points', which the builder's document keeps.
execution = dict(document.pop("execution", {}))
keywords: dict[str, object] = {
key: tuple(value) if isinstance(value, list) else value
for key, value in execution.items()
if key in _SCAN_EXECUTION
}
kept = {
key: value for key, value in execution.items() if key not in _SCAN_EXECUTION
}
if kept:
document["execution"] = kept
builder = RecipeBuilder(
dump_text(document),
None if self._base is None else str(self._base),
self._allow_outside,
)
# The scan file keeps the authored recipe, so that it reopens with its setup.
text = self.to_text()
keywords["recipe"] = RecipeRecord(FORMAT, text, text, {})
return RecipeScan(scan, builder, keywords)
def _axis(self, path: str, values: object) -> list[object]:
"""Return a scan axis's values, read in the unit of the path's keyword.
Parameters
----------
path : str
The parameter path.
values : object
A list, or ``{start, stop, num, spacing}``.
Returns
-------
list
The values.
Raises
------
RecipeError
If the path's keyword is not the route's, or the values do not read.
"""
steps = parse_path(path)
keyword = format_path(steps[:1])
table = self._table()
if keyword not in table:
raise unknown_key(
keyword,
list(table),
where=f"the {self.route} route's keywords",
path=path,
)
unit = self._unit_at(path)
if isinstance(values, Mapping):
start = float(cast("float", _number_in(values.get("start"), unit, path)))
stop = float(cast("float", _number_in(values.get("stop"), unit, path)))
num = decode_integer(values.get("num"), path=path)
spacing = values.get("spacing", "linear")
if spacing not in ("linear", "log"):
msg = f"spacing is 'linear' or 'log', not {spacing!r}"
raise RecipeError(msg, path=path)
axis = (
np.geomspace(start, stop, num)
if spacing == "log"
else np.linspace(start, stop, num)
)
return list(axis.tolist())
if not isinstance(values, list):
msg = "a scan axis is a list of values or {start, stop, num, spacing}"
raise RecipeError(msg, path=path)
return [_number_in(value, unit, path) for value in values]
def _unit_at(self, path: str) -> str | None:
"""Return the unit of the keyword a path ends on.
Parameters
----------
path : str
The parameter path.
Returns
-------
str or None
The unit of a top-level keyword's number form, or of a registered type's
keyword; ``None`` for a value without one.
"""
steps = parse_path(path)
if len(steps) == 1:
entry = self._table()[format_path(steps)]
return next(
(form.unit for form in entry.forms if form.kind == "number"), None
)
parent = get_path(self._document["build"], steps[:-1])
if isinstance(parent, Mapping) and str(parent.get("type")) in REGISTRY:
unit = REGISTRY[str(parent["type"])].units.get(
format_path(steps[-1:]).lstrip(".")
)
return unit if isinstance(unit, str) else None
return None
# ------------------------------------------------------------------ writing
[docs]
def to_text(self) -> str:
"""Return the recipe's TOML text.
Returns
-------
str
The text it was read from while it is unedited, comments included;
otherwise canonical TOML, each quantity in its keyword's display unit.
"""
if self._text is not None:
return self._text
return dump_text(self._canonical_document())
def _context(self, grid: "TimeGrid | None" = None) -> BuildContext: # noqa: UP037
"""Return the context the recipe's values are decoded in.
Parameters
----------
grid : solprop.grid.TimeGrid or None, optional
The route's grid, for a value built on it.
Returns
-------
BuildContext
The data files' directory, the embedded files and the grid.
"""
return BuildContext(
base=self._base,
allow_outside=self._allow_outside,
grid=grid,
embedded=self._embedded,
# The planar route's slab channels carry the run's fill.
mode_fill=self.route == "planar",
)
[docs]
def save(self, path: str | Path) -> None:
"""Write the recipe to a file.
A recipe read back from a run's output (:meth:`from_output`) writes the data
files the output embedded beside it, under the generated names they were
filed by, and refers to them by those names; a file the output only referred
to stays a reference, which :meth:`check` reports when it is missing.
Parameters
----------
path : str or pathlib.Path
The ``.toml`` file to write.
"""
path = Path(path)
if not self._embedded:
path.write_text(self.to_text(), encoding="utf-8")
return
for original, data in self._embedded.items():
(path.parent / self._filed[original]).write_bytes(data)
document = self._canonical_document()
document["build"] = cast(
"dict[str, object]", _renamed(document["build"], self._filed)
)
path.write_text(dump_text(document), encoding="utf-8")
[docs]
def to_python(self, output: str | None = None) -> str:
"""Return a plain Python script that makes the same run.
The route's ``prop_*`` call when every key is on its surface, otherwise
``Simulation.from_*``, the settings and ``run``: keyword arguments in SI, with
a comment per keyword stating the recipe's value where it was a quantity.
Parameters
----------
output : str or None, optional
The file the script's run writes; ``None`` keeps it in memory.
Returns
-------
str
The script.
"""
from solprop.recipe.export import script # noqa: PLC0415 - only here
return script(self, output)
def _deferred(context: BuildContext) -> BuildContext:
"""Return a context for checking values built on the grid before it exists.
Parameters
----------
context : BuildContext
The recipe's context.
Returns
-------
BuildContext
The same, with a stand-in grid, so a factory on the grid can be checked.
"""
from solprop.grid import EnvelopeGrid # noqa: PLC0415 - only for the stand-in
grid = EnvelopeGrid.from_limits(1.0, 800e-9, (400e-9, 2e-6), 200e-15)
return dataclasses.replace(context, grid=grid)
def _needs_grid(value: object) -> bool:
"""Return whether a value holds a factory built on the route's grid.
Parameters
----------
value : object
A value as written.
Returns
-------
bool
True if a table in it names a factory whose context is the grid.
"""
if isinstance(value, Mapping):
factory = REGISTRY.get(str(value.get("type")))
if factory is not None and "grid" in factory.context:
return True
return any(_needs_grid(item) for item in value.values())
if isinstance(value, list):
return any(_needs_grid(item) for item in value)
return False
def _issue(error: SolpropError, keyword: str | None = None) -> RecipeIssue:
"""Return the issue a refusal makes.
Parameters
----------
error : solprop.exceptions.SolpropError
The refusal.
keyword : str or None, optional
The build keyword whose value was refused, named when the refusal names no
recipe path of its own.
Returns
-------
RecipeIssue
Its message, parameter and rule, as an error.
"""
if isinstance(error, RecipeError):
return RecipeIssue(error.message, error.parameter, "error", error.rule)
parameter = keyword if keyword is not None else error.parameter
return RecipeIssue(str(error), parameter, "error", error.rule)
def _plain(value: object, path: str) -> object:
"""Return a value as a recipe holds it, refusing an object or a callable.
Parameters
----------
value : object
A Python value.
path : str
Its path, for the refusal.
Returns
-------
object
The value, with tuples as lists and complex numbers as ``{re, im}``.
Raises
------
RecipeError
If the value is an object or a callable.
"""
if value is None:
return NONE
if isinstance(value, bool | int | float | str):
return value
if isinstance(value, complex):
return {"re": value.real, "im": value.imag}
if isinstance(value, Sequence):
return [_plain(item, f"{path}[{index}]") for index, item in enumerate(value)]
if isinstance(value, Mapping):
return {str(key): _plain(item, f"{path}.{key}") for key, item in value.items()}
kind = "a callable" if callable(value) else f"a {type(value).__name__}"
msg = (
f"{kind} cannot be written into a recipe: give it as a table with a type key "
"(see the recipe guide), or write the run in Python"
)
raise RecipeError(msg, path=path)
def _canonical(entry: Parameter, value: object) -> object:
"""Spell a value in its keyword's display unit, for a canonical recipe.
Parameters
----------
entry : Parameter
The keyword's metadata.
value : object
The value as written.
Returns
-------
object
A quantity string in the display unit for a number of a keyword that has
one, element by element for a list; a two-row list and anything else as it
stands.
"""
unit = next((form.unit for form in entry.forms if form.unit is not None), None)
if entry.display is None or unit is None:
return value
if isinstance(value, list):
# The rows of a two-row form, (z, p), each take their own unit.
rows = any(isinstance(item, list) for item in value)
return value if rows else [_canonical(entry, item) for item in value]
if isinstance(value, bool) or not isinstance(value, int | float | str):
return value
if isinstance(value, str) and value == NONE:
return value
number = parse_quantity(value, Unit(unit), path=entry.name)
if not math.isfinite(number): # pragma: no cover - refused by parse_quantity
return value
return format_quantity(number, Unit(unit), entry.display)
def _renamed(value: object, names: Mapping[str, str]) -> object:
"""Return a value with its data-file references renamed.
Parameters
----------
value : object
A value as the recipe writes it.
names : collections.abc.Mapping
The new name of each file, by the name the recipe refers to it by.
Returns
-------
object
A copy, each ``file`` that `names` holds replaced.
"""
if isinstance(value, Mapping):
out = {key: _renamed(item, names) for key, item in value.items()}
if "sha256" in out and isinstance(out.get("file"), str):
out["file"] = names.get(str(out["file"]), out["file"])
return out
if isinstance(value, list):
return [_renamed(item, names) for item in value]
return value
def _with_pulse_placeholders(document: Document) -> dict[str, object]:
"""Return ``[build]`` with each pulse pin at its placeholder, for decoding.
A pulse in ``pulses`` whose energy is pinned cannot be built without one, and the
pin is solved only after the values are checked.
Parameters
----------
document : dict
The recipe's tables.
Returns
-------
dict
The ``[build]`` table, unchanged where no pulse pin is unset.
"""
from solprop.recipe.pins import PULSE_PINS # noqa: PLC0415 - cycle
build = document["build"]
for path in document.get("pins", {}):
steps = parse_path(path)
name = format_path(steps[-1:]).lstrip(".")
if len(steps) > 1 and name in PULSE_PINS:
try:
build = set_path(build, steps, _placeholder(path))
except RecipeError:
continue # the pin's own check names what is wrong with it
return build
def _placeholder(path: str) -> float:
"""Return the value a pulse pin stands at while the guide is solved.
Parameters
----------
path : str
The pulse pin's path.
Returns
-------
float
Its placeholder, in SI.
"""
from solprop.recipe.pins import placeholder # noqa: PLC0415 - cycle
return placeholder(path)
def _pick(pin: "Pin", roots: "DesignRoots") -> float: # noqa: UP037 - TYPE_CHECKING
"""Return a pin's chosen root, an unmet target refused on its path.
Parameters
----------
pin : solprop.recipe.pins.Pin
The pin.
roots : solprop.tools.DesignRoots
Its roots.
Returns
-------
float
The root its ``root`` key chooses.
Raises
------
RecipeError
Naming the attainable range when there is no root, or an index past them.
"""
try:
return roots.pick(pin.root)
except SolpropError as error:
raise RecipeError(str(error), path=pin.path) from error
[docs]
def resume_file(
path: str | Path,
mode: bool | Literal["interpolated"] = True,
*,
node_budget: int | None = None,
progress: "ProgressReporter | ProgressChoice | None" = None, # noqa: UP037
stop: "StopFlag | None" = None, # noqa: UP037 - checked at run time
) -> "Output": # noqa: UP037 - checked at run time
"""Continue a run from its output file alone, rebuilding it from its recipe.
The simulation is rebuilt from the resolved recipe the file records (SEP 41, S2)
and run into the same file with ``resume=mode``, so the resume gate decides by the
hashes of the stored and rebuilt records, as for any resume (ADR-066). A file
written from a script records no recipe; its script resumes it.
Resuming a file runs the setup it holds. A file received from someone else does
not choose its own limit: the recipe's ``node_budget`` is ignored, and the caller's
applies.
Parameters
----------
path : str or pathlib.Path
The output file.
mode : bool or {'interpolated'}, optional
As :meth:`~solprop.driver.simulation.Simulation.run`'s ``resume`` takes it.
node_budget : int or None, optional
The node-buffer budget of a capillary run; ``None`` takes solprop's default.
progress : object, optional
Where progress goes, as :meth:`~solprop.driver.simulation.Simulation.run`
takes it.
stop : object, optional
The run's stop flag, any object with an ``is_set()`` method.
Returns
-------
solprop.output.schema.Output
The handler the run continued in.
Raises
------
RecipeError
If the file records no recipe, or one of a newer format.
"""
recipe = Recipe.from_output(path, resolved=True)
document = recipe.to_dict()
document["build"].pop("node_budget", None)
if node_budget is not None:
document["build"]["node_budget"] = node_budget
rebuilt = Recipe(document)
rebuilt._embedded = recipe._embedded
rebuilt._filed = recipe._filed
return rebuilt.build().run(
Path(path),
resume=mode,
progress=progress,
stop=stop,
)
def _number_in(value: object, unit: str | None, path: str) -> object:
"""Return one scan value as a number in its unit, or as it stands.
Parameters
----------
value : object
The value as written.
unit : str or None
The keyword's unit.
path : str
Its path.
Returns
-------
object
A float in `unit` for a number or quantity string; anything else unchanged.
"""
if (
unit is None
or isinstance(value, bool)
or not isinstance(value, int | float | str)
):
return value
return parse_quantity(value, Unit(unit), path=path)
[docs]
@dataclass(frozen=True)
class RecipeBuilder:
"""Build one scan point's simulation from a recipe; picklable for pool executors.
Attributes
----------
text : str
The recipe, without its ``[scan]`` and ``[execution]`` tables.
base : str or None
Its directory, for data files.
allow_outside : bool
Whether a data file may lie outside it.
"""
text: str
base: str | None
allow_outside: bool
def __call__(self, point: "ScanPoint") -> "Simulation": # noqa: UP037
"""Set the point's values, pass its seed, and build.
Parameters
----------
point : solprop.driver.scan.ScanPoint
The point.
Returns
-------
solprop.driver.simulation.Simulation
Its simulation.
"""
recipe = Recipe.loads(
self.text,
base=None if self.base is None else Path(self.base),
allow_outside=self.allow_outside,
)
for path, value in point.values.items():
if path != "seed":
recipe.set(path, value)
if point.seed is not None and "seed" not in recipe.to_dict()["build"]:
recipe.set("seed", encode_integer(point.seed))
return recipe.build()
[docs]
@dataclass(frozen=True)
class RecipeScan:
"""A recipe's scan, ready for :func:`~solprop.driver.scan.run_scan`.
Attributes
----------
scan : solprop.driver.scan.Scan
The points.
build : callable
The picklable builder of each point's simulation.
keywords : dict
The ``run_scan`` keywords: ``fields`` and ``field_planes`` from the recipe's
``[execution]``, which ``run_scan`` takes itself, and ``recipe``, which the
scan file records.
Examples
--------
``run_scan(scan.scan, scan.build, output="scan.h5", **scan.keywords)`` runs it.
"""
scan: "Scan" # noqa: UP037 - TYPE_CHECKING only
build: RecipeBuilder
keywords: dict[str, object]