"""Disinfection unit operations: UV dose-response and chlorine CT / log-removal.
Effluent-permit work needs disinfection sizing and a pathogen log-removal credit.
The reaction *models* ``uv_h2o2`` / ``ozone_bromate`` model the oxidation
chemistry, but neither is a disinfection *unit op* that reduces a pathogen
indicator in the flowsheet. This module adds two, plus the credit physics behind
them:
* :class:`UVUnit` -- a UV reactor. The **dose** is the average fluence rate times
the exposure time (``V/Q``), corrected for the water's UV transmittance (UVT);
a log-linear **dose-response** (with optional tailing) gives the
log-inactivation.
* :class:`ChlorineContactUnit` -- a chlorine contact tank. The applied chlorine
**residual** is a dynamic state that decays first-order; the **CT** (residual ×
the T10 contact time) drives a CT-based log-removal credit. T10 comes from a
baffling factor (``T10 = baffling·V/Q``) or, for a non-ideal contactor, from a
measured/simulated residence-time distribution (:func:`t10_from_rtd`, reusing
:mod:`aquakin.utils.rtd`).
Both **pass the process (ASM) stream through unchanged** -- disinfection does not
materially change COD/N/P at this fidelity -- and reduce the **indicator-organism
density carried on the stream** (``Stream.org``, the disinfection analogue of the
temperature scalar), falling back to the unit's design ``inlet_density`` when the
inlet carries none. So the indicator is tracked through the flowsheet (mixers
flow-weight it) and the disinfection unit applies the log-inactivation, matching
the behaviour of the commercial process simulators.
The credit physics is exposed as pure, AD-clean functions (``uv_dose`` /
``uv_log_inactivation`` / ``ct_value`` / ``ct_log_removal`` / ``t10_from_baffling``
/ ``t10_from_rtd``) so a contactor can be sized or a credit computed standalone.
Units: the standalone functions are unit-agnostic (the caller keeps them
consistent). The units convert the plant residence time ``V/Q`` (in the plant's
time unit -- days for the BSM family) to seconds for the UV dose (fluence rate in
mW/cm², dose in mJ/cm²); the chlorine T10/CT stay in the plant time unit.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
import jax.numpy as jnp
from aquakin.plant._constants import EPS_Q, SECONDS_PER_DAY
from aquakin.plant.streams import Stream
if TYPE_CHECKING: # pragma: no cover
from aquakin.core.model import CompiledModel
# --- UV credit physics ------------------------------------------------------
def uvt_intensity_factor(uvt: jnp.ndarray | None, uvt_ref: jnp.ndarray | None) -> jnp.ndarray:
"""First-order UV-transmittance correction on the average fluence rate.
The average intensity in a UV reactor falls as the water absorbs more UV. With
``uvt`` / ``uvt_ref`` the % transmittance (per cm) at operating / reference
conditions, this returns the linear ratio ``uvt / uvt_ref`` -- a first-order
correction; an exact treatment needs the reactor geometry and a
dose-distribution (UVDGM) model. ``uvt is None`` returns 1 (no correction)."""
if uvt is None or uvt_ref is None:
return 1.0
return jnp.asarray(uvt) / jnp.asarray(uvt_ref)
[docs]
def uv_dose(
intensity: jnp.ndarray,
exposure_time: jnp.ndarray,
*,
uvt: jnp.ndarray | None = None,
uvt_ref: jnp.ndarray | None = None,
) -> jnp.ndarray:
"""UV dose (fluence) = average fluence rate × exposure time.
``intensity`` is the average fluence rate (e.g. mW/cm²) and ``exposure_time``
the contact time in the matching time unit (seconds → dose in mJ/cm²),
optionally scaled by the UVT correction :func:`uvt_intensity_factor`."""
return jnp.asarray(intensity) * uvt_intensity_factor(uvt, uvt_ref) * jnp.asarray(exposure_time)
[docs]
def uv_log_inactivation(
dose: jnp.ndarray, d10: jnp.ndarray, *, max_log: jnp.ndarray | None = None
) -> jnp.ndarray:
"""Log-inactivation from a UV dose by a log-linear dose-response.
``log10(N/N0) = dose / d10`` where ``d10`` is the dose for one log of
inactivation (mJ/cm²/log). Optionally capped at ``max_log`` to represent
tailing (a resistant subpopulation)."""
log = jnp.asarray(dose) / d10
if max_log is not None:
log = jnp.minimum(log, max_log)
return log
# --- chlorine credit physics ------------------------------------------------
[docs]
def t10_from_baffling(
volume: jnp.ndarray, flow: jnp.ndarray, baffling_factor: jnp.ndarray
) -> jnp.ndarray:
"""T10 contact time from the baffling factor: ``T10 = baffling·V/Q``.
The baffling factor is the ratio of the T10 (the time below which 10 % of the
flow exits) to the mean hydraulic residence time ``V/Q`` -- ~0.1 for an
unbaffled tank, ~0.7 for a well-baffled one, 1.0 for ideal plug flow."""
return baffling_factor * jnp.asarray(volume) / (jnp.asarray(flow) + EPS_Q)
[docs]
def t10_from_rtd(t: jnp.ndarray, C: jnp.ndarray) -> jnp.ndarray:
"""T10 contact time from a measured/simulated residence-time distribution.
Reuses :func:`aquakin.utils.rtd.percentile_time` (the 10th percentile -- the
``q=0.10`` quantile -- of the cumulative residence-time distribution) -- the
non-ideal-contactor alternative to a baffling-factor estimate."""
from aquakin.utils.rtd import percentile_time
return percentile_time(t, C, 0.10)
[docs]
def ct_value(residual: jnp.ndarray, t10: jnp.ndarray) -> jnp.ndarray:
"""CT = disinfectant residual × T10 contact time (e.g. mg·min/L)."""
return jnp.asarray(residual) * jnp.asarray(t10)
[docs]
def ct_log_removal(
ct: jnp.ndarray, ct_per_log: jnp.ndarray, *, max_log: jnp.ndarray | None = None
) -> jnp.ndarray:
"""Log-removal from a CT value by a CT-based credit (Chick–Watson form).
``log = CT / ct_per_log`` where ``ct_per_log`` is the CT that earns one log of
inactivation (from the regulatory CT tables for the target organism, pH and
temperature). Optionally capped at ``max_log``."""
log = jnp.asarray(ct) / ct_per_log
if max_log is not None:
log = jnp.minimum(log, max_log)
return log
def _apply_log_removal(org_in, log_inactivation):
"""Reduce an indicator density by a log-inactivation: ``N = N0·10^(-log)``."""
return org_in * 10.0 ** (-log_inactivation)
# --- UV reactor unit --------------------------------------------------------
[docs]
@dataclass
class UVUnit:
"""A UV disinfection reactor (stateless): dose-driven log-inactivation.
Passes the process stream through unchanged and reduces the indicator-organism
density (``Stream.org``, else ``inlet_density``) by the dose-response
log-inactivation. The dose is ``intensity · exposure · UVT-factor`` with the
exposure time the (baffling-scaled) residence time ``V/Q`` converted to
seconds.
Parameters
----------
name : str
model : CompiledModel
volume : float
Reactor volume (m³); the exposure time is ``baffling_factor·V/Q``.
intensity : float
Average UV fluence rate (mW/cm²) at the reference UVT.
d10 : float
Dose for one log of inactivation (mJ/cm²/log).
uvt, uvt_ref : float, optional
Operating / reference UV transmittance (% per cm) for the first-order
intensity correction (:func:`uvt_intensity_factor`). Both ``None`` = no
correction.
max_log : float, optional
Cap on the log-inactivation (tailing). ``None`` = uncapped.
baffling_factor : float
T10/HRT short-circuiting factor on the exposure time. Default 1.0.
inlet_density : float
Indicator density used when the inlet stream carries none (``org`` is
``None``). Default 0.
input_port, output_port : str
"""
name: str
model: CompiledModel
volume: float
intensity: float
d10: float
uvt: float | None = None
uvt_ref: float | None = None
max_log: float | None = None
baffling_factor: float = 1.0
inlet_density: float = 0.0
input_port: str = "in"
output_port: str = "out"
# ----- stateless ------------------------------------------------------
@property
def state_size(self) -> int:
return 0
@property
def input_ports(self) -> list[str]:
return [self.input_port]
@property
def output_ports(self) -> list[str]:
return [self.output_port]
def initial_state(self) -> jnp.ndarray:
return jnp.zeros((0,))
def rhs(self, t, state, inputs, params, signals=None) -> jnp.ndarray:
return jnp.zeros((0,))
# ----- behaviour ------------------------------------------------------
[docs]
def exposure_seconds(self, flow) -> jnp.ndarray:
"""Exposure time (s): the baffling-scaled residence ``V/Q`` (plant time
unit, days) converted to seconds."""
hrt_days = self.baffling_factor * self.volume / (jnp.asarray(flow) + EPS_Q)
return hrt_days * SECONDS_PER_DAY
[docs]
def log_inactivation(self, flow) -> jnp.ndarray:
"""The UV log-inactivation at the given throughflow."""
dose = uv_dose(
self.intensity, self.exposure_seconds(flow), uvt=self.uvt, uvt_ref=self.uvt_ref
)
return uv_log_inactivation(dose, self.d10, max_log=self.max_log)
def compute_outputs(self, t, state, inputs, params, signals=None) -> dict:
s_in = inputs[self.input_port]
org_carried = s_in.scalars.get("org")
org_in = org_carried if org_carried is not None else self.inlet_density
org_out = _apply_log_removal(org_in, self.log_inactivation(s_in.Q))
return {
self.output_port: Stream(
Q=s_in.Q, C=s_in.C, model=self.model, scalars={**s_in.scalars, "org": org_out}
)
}
def flow_outputs(self, input_flows: dict, params, ctx=None) -> dict:
return {self.output_port: input_flows[self.input_port]}
# --- chlorine contact unit --------------------------------------------------