aquakin.PlantSolution#

class aquakin.PlantSolution(t, state, plant, events_log=None, _requested_time_unit=None)[source]#

Bases: object

Result of Plant.solve().

Holds the integrated unit states (state), not the inter-unit streams: the plant integrates states only, so an effluent / recycle stream is reconstructed on demand from the saved states via stream() (the whole sweep is reconstructed once and cached on the solution). Per-unit concentrations are read directly with C_named() / unit_state().

Variables:

t (jnp.ndarray) – Save times, shape (n_t,).
state (jnp.ndarray) – Full plant state, shape (n_t, total_state_size). This is the raw integrated state. Where a unit’s model sets clip_negative_states (on by default for ASM1, so the activated-sludge reactors in BSM1/BSM2), entries may be small transient negatives – the max(x, 0) clamp is applied only when evaluating that unit’s reaction rates, not to the saved state. A normal numerical transient, not a solver/model error; clip with jnp.maximum(state, 0.0) for display if needed.
plant (Plant) – The plant that produced this solution; retained for accessor methods.
events_log (list of (float, str), optional) – When the solve used events=, the fired events in order as (time, name). None for a plain solve.
_requested_time_unit (str, optional) – The unit t was rescaled into when Plant.solve() was called with an explicit time_unit=, or None for the plant’s native unit. Backs the time_unit property; a declared field (rather than an attribute set after construction) so it participates in repr/eq.

Parameters:

t (jnp.ndarray)
state (jnp.ndarray)
plant (Plant)
events_log (list | None)
_requested_time_unit (str | None)

__init__(t, state, plant, events_log=None, _requested_time_unit=None)#

Parameters:

t (jnp.ndarray)
state (jnp.ndarray)
plant (Plant)
events_log (list | None)
_requested_time_unit (str | None)

Return type:

None

Methods

`C_named`(unit_name, species)	Return a single species' trajectory in a single unit.
`C_named_many`(unit_name, species)	Trajectories of several species in one unit, as `{name: array}`.
`__init__`(t, state, plant[, events_log, ...])
`available_streams`()	The `"unit.port"` output endpoints `Plant.stream()` accepts (a convenience alias for `plant.list_ports()`).
`final_named`(unit_name[, species])	Values at the last save time for one unit, as `{name: float}`.
`plot`(unit_name[, species, ax])	Plot one unit's species trajectories over time.
`stream`(endpoint[, params])	Reconstruct one output stream's trajectory -- `sol.stream("effluent")`.
`to_csv`([path_or_buf, units_in_columns])	Write one unit's trajectory to CSV (delegates to `to_dataframe()`).
`to_dataframe`(unit, *[, units_in_columns])	Return one unit's state trajectory as a pandas `DataFrame`.
`unit_state`(unit_name)	Return the trajectory of one unit's state, shape `(n_t, unit.state_size)`.

Attributes

`events_log`
`final_state`	The full plant state at the last save time, shape `(total_state_size,)`.
`time_unit`	The time unit of `t` (`"s"`, `"d"`, .
`t`
`state`
`plant`

property time_unit: str | None#

The time unit of t ("s", "d", … or None).

The plant’s native unit (Plant.time_unit), or the time_unit= passed to Plant.solve() when the times were reported in that unit.

property final_state: Array#

The full plant state at the last save time, shape (total_state_size,).

The last row of state. Pass it to Plant.states_by_unit() to read per-unit pieces without a trailing [-1] index on a 2-D trajectory.

unit_state(unit_name)[source]#

Return the trajectory of one unit’s state, shape (n_t, unit.state_size). See plant.list_units() for the names.

Parameters:: unit_name (str)
Return type:: Array

available_streams()[source]#

The "unit.port" output endpoints Plant.stream() accepts (a convenience alias for plant.list_ports()).

Return type:: list[str]

stream(endpoint, params=None)[source]#

Reconstruct one output stream’s trajectory – sol.stream("effluent").

A convenience for plant.stream(sol, endpoint, params) (the plant is carried on the solution). Inter-unit streams are not stored – the plant integrates unit states, so the effluent/recycle streams are recomputed from the saved states; the whole sweep is reconstructed once and cached on this solution, so repeated calls for different ports are cheap. endpoint is a semantic name ("effluent", "ras", …) or a "unit.port" (see available_streams() / plant.list_streams).

Parameters:

endpoint (str)
params (Array | None)

C_named(unit_name, species)[source]#

Return a single species’ trajectory in a single unit.

Only valid for units whose state is a concentration vector in the unit’s model (CSTRs and other kinetic units). See plant.list_units() / plant.list_species(unit) for the valid names.

Parameters:

unit_name (str)
species (str)

Return type:

Array

C_named_many(unit_name, species)[source]#

Trajectories of several species in one unit, as {name: array}.

The multi-species companion to C_named() – read a handful of a unit’s species in one call (sol.C_named_many("tank5", ["SNH", "SNO"])) instead of a slice each. An unknown unit/species raises the same hinted KeyError C_named() gives.

Parameters:: unit_name (str)
Return type:: dict[str, Array]

final_named(unit_name, species=None)[source]#

Values at the last save time for one unit, as {name: float}.

The reporting shortcut for a steady-state value: instead of float(sol.C_named("tank5", "SNH")[-1]) per species, sol.final_named("tank5", ["SNH", "SNO"]) returns them in one dict. With species=None (default) every species of the unit’s model is returned (see Plant.list_species()). Values are plain Python floats (a post-processing read on an already-solved solution – use C_named(unit, sp)[-1] if you need a differentiable last value).

Parameters:: unit_name (str)
Return type:: dict[str, float]

plot(unit_name, species=None, *, ax=None, **kwargs)[source]#

Plot one unit’s species trajectories over time.

The plant analogue of BatchSolution.plot: a thin matplotlib wrapper so sol.plot("tank5", "SNH") needs no manual C_named / unit / axis-label boilerplate. The x-axis is labelled with the plant’s time unit; a single-species plot labels the y-axis with that species’ units.

Parameters:

unit_name (str) – A concentration-vector unit (see Plant.list_species()).
species (str or iterable of str, optional) – Species to plot; a single name, an iterable (legended), or None for every species of the unit’s model.
ax (matplotlib.axes.Axes, optional) – Axes to draw on; a new one is created if omitted.
**kwargs – Forwarded to ax.plot.

Return type:

matplotlib.axes.Axes

Raises:

ImportError – If matplotlib is not installed (pip install aquakin[plot]).
KeyError – For an unknown unit / species, or a non-concentration unit (hinted).

to_dataframe(unit, *, units_in_columns=False)[source]#

Return one unit’s state trajectory as a pandas DataFrame.

The plant integrates a heterogeneous flat state across many units, so a single whole-plant table is not meaningful; pick one unit. For a kinetic unit (one with a model) the columns are species names; for any other unit they are generic state_0..state_{n-1} columns.

Parameters:

unit (str) – Name of the unit whose trajectory to tabulate.
units_in_columns (bool, optional) – If True, append " [unit]" to each species column label (kinetic units only); otherwise units are stored in df.attrs["units"].

Returns:

Time-indexed (t), one row per save time.

Return type:

pandas.DataFrame

Raises:

KeyError – If unit is not a unit of this plant.
ImportError – If pandas (an optional dependency) is not installed.

to_csv(path_or_buf=None, *, unit, units_in_columns=True, **kwargs)[source]#

Write one unit’s trajectory to CSV (delegates to to_dataframe()).

unit is required (the plant state is per-unit). units_in_columns defaults to True so the file is self-describing. Extra keyword arguments are forwarded to pandas.DataFrame.to_csv.

Parameters:

unit (str)
units_in_columns (bool)