src/physics.py

"""Clean-room synthetic pump physics for a public edge-inference baseline.

All constants are illustrative engineering proxies. They are not copied from
vendor curves, CAD, BOMs, plant telemetry, or the private pump repository.
"""

from __future__ import annotations

import numpy as np

FEATURE_NAMES = (
    "speed_fraction",
    "static_head_m",
    "system_k",
    "voltage_fraction",
    "ambient_temp_c",
    "inlet_pressure_bar",
)

TARGET_NAMES = (
    "flow_m3h",
    "total_head_m",
    "input_power_kw",
    "winding_temp_c",
    "npsh_margin_m",
    "efficiency_fraction",
)

# Public, illustrative baseline constants—not equipment specifications.
SHUTOFF_HEAD_M = 132.0
PUMP_CURVE_COEFF = 0.58
BEP_FLOW_M3H = 11.5
DEMO_THERMAL_REVIEW_C = 70.0
DEMO_OVERLOAD_REVIEW_KW = 3.2


def evaluate(features: np.ndarray) -> np.ndarray:
    """Evaluate a generic pump/system intersection and thermal proxy.

    Args:
        features: float array shaped ``(..., 6)`` in FEATURE_NAMES order.

    Returns:
        Float array shaped ``(..., 6)`` in TARGET_NAMES order.
    """

    x = np.asarray(features, dtype=np.float64)
    if x.shape[-1] != len(FEATURE_NAMES):
        raise ValueError(f"expected {len(FEATURE_NAMES)} features, got {x.shape[-1]}")

    speed, static_head, system_k, voltage, ambient, inlet_pressure = np.moveaxis(x, -1, 0)
    available_head = np.maximum(SHUTOFF_HEAD_M * speed**2 - static_head, 0.0)
    flow = np.sqrt(available_head / np.maximum(PUMP_CURVE_COEFF + system_k, 1e-6))
    total_head = static_head + system_k * flow**2

    bep_flow = np.maximum(BEP_FLOW_M3H * speed, 0.1)
    relative_offset = (flow - bep_flow) / bep_flow
    pump_efficiency = np.clip(
        0.74 - 0.34 * relative_offset**2 - 0.05 * (1.0 - speed),
        0.18,
        0.76,
    )
    motor_efficiency = np.clip(
        0.90 - 0.12 * (1.0 - voltage) ** 2 - 0.05 * (1.0 - speed) ** 2,
        0.65,
        0.92,
    )
    efficiency = pump_efficiency * motor_efficiency

    hydraulic_power_kw = 9.80665 * (flow / 3600.0) * total_head
    input_power_kw = hydraulic_power_kw / np.maximum(efficiency, 0.1) + 0.10 * speed

    low_flow_penalty = 12.0 * np.clip(1.0 - flow / np.maximum(0.55 * bep_flow, 0.1), 0.0, 1.0)
    undervoltage_penalty = 60.0 * np.clip(0.90 - voltage, 0.0, 0.20)
    winding_temp_c = ambient + 7.2 * input_power_kw + low_flow_penalty + undervoltage_penalty

    npsh_required_m = 1.4 + 0.018 * flow**2
    npsh_available_m = 10.197 * inlet_pressure + 2.0
    npsh_margin_m = npsh_available_m - npsh_required_m

    return np.stack(
        (flow, total_head, input_power_kw, winding_temp_c, npsh_margin_m, efficiency),
        axis=-1,
    )


def anomaly_flags(features: np.ndarray, outputs: np.ndarray) -> list[str]:
    """Return deterministic advisory flags for each row.

    These thresholds are test/demo policy, not a certified protection layer.
    """

    x = np.asarray(features, dtype=np.float64)
    y = np.asarray(outputs, dtype=np.float64)
    rows: list[str] = []
    for feature, target in zip(x.reshape(-1, 6), y.reshape(-1, 6), strict=True):
        speed, _, _, voltage, _, _ = feature
        flow, _, power, winding, npsh_margin, _ = target
        flags: list[str] = []
        if speed > 0.35 and flow < 0.20:
            flags.append("no_flow_risk")
        if npsh_margin < 1.0:
            flags.append("cavitation_risk")
        if winding > DEMO_THERMAL_REVIEW_C:
            flags.append("thermal_risk")
        if voltage < 0.90:
            flags.append("undervoltage")
        if power > DEMO_OVERLOAD_REVIEW_KW:
            flags.append("overload_risk")
        rows.append("|".join(flags) if flags else "normal")
    return rows


def sample_features(count: int, seed: int = 20260802) -> np.ndarray:
    """Sample the frozen synthetic operating envelope."""

    rng = np.random.default_rng(seed)
    columns = (
        rng.uniform(0.35, 1.05, count),
        rng.uniform(5.0, 120.0, count),
        rng.uniform(0.03, 0.80, count),
        rng.uniform(0.82, 1.08, count),
        rng.uniform(5.0, 50.0, count),
        rng.uniform(0.15, 2.50, count),
    )
    return np.stack(columns, axis=-1).astype(np.float32)