particula.execution.diagnostics

diagnostics

Write closed resident diagnostics into caller-owned Warp arrays.

This concrete direct-import-only module has no callback registration or package export. Registrations execute in this fixed order: gas-concentration snapshot, saturation-ratio snapshot, total species mass, particle-number concentration, latent heat energy, and conservation residual. Matrix operations use (B, S) wp.float64 arrays; particle number uses a (B,) wp.float64 array.

Total species mass is V[b] * (Σp(m[b, p, s] * c[b, p]) + g[b, s]) in kg. Particle number is Σp(c[b, p]) in m^-3. Latent energy copies signed whole-call P2-finalized energy in J. The residual is total_mass - baseline_total_mass - source_ledger + sink_ledger in kg; source and sink ledgers are nonnegative accumulated extensive-mass inputs. Execution validates caller-owned same-device bindings without host readback, synchronization, transfer, allocation, or physics mutation. Empty matrix operations are write-free for B == 0 or S == 0; particle number is write-free only for B == 0.

ResidentDiagnosticOperation

Bases: str, Enum

Enumerate the closed resident diagnostic operations in launch order.

ResidentDiagnosticRegistration dataclass

ResidentDiagnosticRegistration(operation: ResidentDiagnosticOperation, output: object, energy_transfer: object | None = None, baseline_total_mass: object | None = None, source_ledger: object | None = None, sink_ledger: object | None = None)

Bind one closed diagnostic operation to caller-owned Warp arrays.

Attributes:

• operation (ResidentDiagnosticOperation) –

  Exact closed operation that selects the diagnostic reduction.

• output (object) –

  Caller-owned Warp float64 output validated by the executor.

• energy_transfer (object | None) –

  Required (B, S) signed whole-call energy input in J for latent-energy output; forbidden otherwise.

• baseline_total_mass (object | None) –

  Required (B, S) extensive mass baseline in kg for residual output; forbidden otherwise.

• source_ledger (object | None) –

  Required nonnegative extensive source ledger for the residual in kg; forbidden otherwise.

• sink_ledger (object | None) –

  Required nonnegative extensive sink ledger for residual in kg; forbidden otherwise.

__post_init__

__post_init__() -> None

Validate the exact closed diagnostic operation.

Raises:

• TypeError –

  If operation is not an exact supported operation.

• ValueError –

  If required accounting inputs are missing or forbidden accounting inputs are supplied for operation.

Source code in particula/execution/diagnostics.py

def __post_init__(self) -> None:
    """Validate the exact closed diagnostic operation.

    Raises:
        TypeError: If ``operation`` is not an exact supported operation.
        ValueError: If required accounting inputs are missing or forbidden
            accounting inputs are supplied for ``operation``.
    """
    if type(self.operation) is not ResidentDiagnosticOperation:
        raise TypeError(
            "operation must be an exact ResidentDiagnosticOperation."
        )
    inputs = (
        self.energy_transfer,
        self.baseline_total_mass,
        self.source_ledger,
        self.sink_ledger,
    )
    if self.operation is ResidentDiagnosticOperation.LATENT_HEAT_ENERGY:
        if self.energy_transfer is None or any(
            item is not None for item in inputs[1:]
        ):
            raise ValueError("latent energy requires only energy_transfer.")
    elif (
        self.operation is ResidentDiagnosticOperation.CONSERVATION_RESIDUAL
    ):
        if (
            any(item is None for item in inputs[1:])
            or self.energy_transfer is not None
        ):
            raise ValueError(
                "residual requires baseline, source, and sink ledgers."
            )
    elif any(item is not None for item in inputs):
        raise ValueError("diagnostic operation forbids accounting inputs.")

ResidentDiagnosticsExecutor

Execute an already-bound closed diagnostics plan without transfers.

Validation preserves caller ownership and rejects output or accounting-input aliases with resident primaries, published sidecars, or diagnostic outputs. Execution dispatches the six canonical registrations without host readback, synchronization, transfer, allocation, or physics mutation. Matrix registrations are write-free for empty (B, S) schemas; particle number remains writable for (B, 0).

execute

execute(plan: object) -> None

Validate and dispatch each registration in declared order.

Matrix schemas complete without their writer launch when B == 0 or S == 0. Particle number still launches for (B, 0) because its (B,) output exists. Successful launches are asynchronous; callers synchronize before inspecting outputs on the host.

Parameters:

• plan (object) –

  Exact plan selecting the sources and caller-owned outputs.

Raises:

• TypeError –

  If plan is not an exact diagnostics plan.

• ValueError –

  If its bindings or output metadata are invalid.

Source code in particula/execution/diagnostics.py

def execute(self, plan: object) -> None:
    """Validate and dispatch each registration in declared order.

    Matrix schemas complete without their writer launch when ``B == 0`` or
    ``S == 0``. Particle number still launches for ``(B, 0)`` because its
    ``(B,)`` output exists. Successful launches are asynchronous; callers
    synchronize before inspecting outputs on the host.

    Args:
        plan: Exact plan selecting the sources and caller-owned outputs.

    Raises:
        TypeError: If ``plan`` is not an exact diagnostics plan.
        ValueError: If its bindings or output metadata are invalid.
    """
    plan = self.validate(plan)
    self._execute_validated(plan)

validate

validate(plan: object) -> ResidentDiagnosticsPlan

Validate one exact diagnostics plan without dispatching a kernel.

Parameters:

• plan (object) –

  Candidate concrete diagnostics plan.

Returns:

• ResidentDiagnosticsPlan –

  The unchanged, exact validated plan.

Raises:

• TypeError –

  If plan is not an exact diagnostics plan.

• ValueError –

  If the plan's graph, bindings, or registration protocol is invalid.

Source code in particula/execution/diagnostics.py

def validate(self, plan: object) -> ResidentDiagnosticsPlan:
    """Validate one exact diagnostics plan without dispatching a kernel.

    Args:
        plan: Candidate concrete diagnostics plan.

    Returns:
        The unchanged, exact validated plan.

    Raises:
        TypeError: If ``plan`` is not an exact diagnostics plan.
        ValueError: If the plan's graph, bindings, or registration protocol
            is invalid.
    """
    if type(plan) is not ResidentDiagnosticsPlan:
        raise TypeError("plan must be an exact ResidentDiagnosticsPlan.")
    self._validate(plan)
    return plan

ResidentDiagnosticsPlan dataclass

ResidentDiagnosticsPlan(session: ResidentSession, registry: object, graph: ResolvedProcessGraph, schedule: ResolvedTimestepSchedule, node: ProcessNode, registrations: tuple[ResidentDiagnosticRegistration, ...])

Bind ordered closed diagnostics to one resident graph and schedule.

Attributes:

• session (ResidentSession) –

  Exact active resident session that owns diagnostic sources.

• registry (object) –

  Exact registry pinned to session.

• graph (ResolvedProcessGraph) –

  Resolver-produced graph containing node by identity.

• schedule (ResolvedTimestepSchedule) –

  Matching resolved schedule that ends with node.

• node (ProcessNode) –

  Canonical diagnostics process node.

• registrations (tuple[ResidentDiagnosticRegistration, ...]) –

  Exact canonical tuple of the six ordered closed operation and output bindings, validated by the executor.

__post_init__

__post_init__() -> None

Validate exact types for the resident diagnostics binding.

Structural graph, lifecycle, and output validation is deferred to the executor so plan construction does not inspect Warp-array metadata.

Raises:

• TypeError –

  If a carrier or registration has an inexact type.

Source code in particula/execution/diagnostics.py

def __post_init__(self) -> None:
    """Validate exact types for the resident diagnostics binding.

    Structural graph, lifecycle, and output validation is deferred to the
    executor so plan construction does not inspect Warp-array metadata.

    Raises:
        TypeError: If a carrier or registration has an inexact type.
    """
    from particula.execution.gpu_resources import GPUResourceRegistry

    if type(self.session) is not ResidentSession:
        raise TypeError("session must be an exact ResidentSession.")
    if type(self.registry) is not GPUResourceRegistry:
        raise TypeError("registry must be an exact GPUResourceRegistry.")
    if type(self.graph) is not ResolvedProcessGraph:
        raise TypeError("graph must be an exact ResolvedProcessGraph.")
    if type(self.schedule) is not ResolvedTimestepSchedule:
        raise TypeError(
            "schedule must be an exact ResolvedTimestepSchedule."
        )
    if type(self.node) is not ProcessNode:
        raise TypeError("node must be an exact ProcessNode.")
    if type(self.registrations) is not tuple or not all(
        type(item) is ResidentDiagnosticRegistration
        for item in self.registrations
    ):
        raise TypeError(
            "registrations must be exact "
            "ResidentDiagnosticRegistration tuple."
        )