Add continuous material polling runner

src/production_analytics/calculations/config.py

@@ -0,0 +1,110 @@
+"""Strict configuration for executable material-consumption instances."""
+
+from dataclasses import dataclass, fields
+from math import isfinite
+from pathlib import Path
+
+import yaml
+
+
+class CalculationConfigError(ValueError):
+    """An operator-readable configuration failure, without input values."""
+
+
+class _UniqueLoader(yaml.SafeLoader):
+    pass
+
+
+def _mapping(loader, node):
+    result = {}
+    for key_node, value_node in node.value:
+        key = loader.construct_object(key_node)
+        if not isinstance(key, str) or key in result:
+            raise CalculationConfigError("YAML mapping keys must be unique strings")
+        result[key] = loader.construct_object(value_node)
+    return result
+
+
+_UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _mapping)
+
+
+def finite_number(value: object, name: str, *, positive: bool = False) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise CalculationConfigError(f"{name} must be a finite number")
+    try:
+        number = float(value)
+    except (OverflowError, ValueError) as exc:
+        raise CalculationConfigError(f"{name} must be a finite number") from exc
+    if not isfinite(number) or (positive and number <= 0):
+        requirement = "finite and greater than zero" if positive else "finite"
+        raise CalculationConfigError(f"{name} must be {requirement}")
+    return number
+
+
+@dataclass(frozen=True, slots=True)
+class MaterialCalculationConfig:
+    id: str
+    type: str
+    version: str
+    machine_ref: str
+    rate_signal_ref: str
+    gate_signal_ref: str
+    gate_threshold: float
+    max_sample_gap_seconds: float
+    output_metric: str
+    output_unit: str
+    group_by: str
+
+
+def load_material_calculation(
+    path: str | Path, calculation_id: str | None = None,
+) -> MaterialCalculationConfig:
+    try:
+        with Path(path).open(encoding="utf-8") as stream:
+            document = yaml.load(stream, Loader=_UniqueLoader)
+    except (OSError, UnicodeError) as exc:
+        raise CalculationConfigError("Cannot read calculation configuration file") from exc
+    except yaml.YAMLError as exc:
+        raise CalculationConfigError("Invalid YAML in calculation configuration") from exc
+    if not isinstance(document, dict) or set(document) != {"calculations"}:
+        raise CalculationConfigError("Configuration must contain only 'calculations'")
+    entries = document["calculations"]
+    if not isinstance(entries, list) or not entries:
+        raise CalculationConfigError("calculations must be a non-empty list")
+    required = {field.name for field in fields(MaterialCalculationConfig)}
+    instances = {}
+    for index, entry in enumerate(entries):
+        prefix = f"calculations[{index}]"
+        if not isinstance(entry, dict):
+            raise CalculationConfigError(f"{prefix} must be a mapping")
+        if "type" in entry and entry["type"] != "material_consumption":
+            raise CalculationConfigError(f"{prefix}.type: only 'material_consumption' is supported")
+        missing = required - entry.keys()
+        if missing:
+            raise CalculationConfigError(f"{prefix}: missing fields: {', '.join(sorted(missing))}")
+        if entry.keys() - required:
+            raise CalculationConfigError(f"{prefix}: unknown fields (check spelling)")
+        for name in sorted(required - {"gate_threshold", "max_sample_gap_seconds"}):
+            if not isinstance(entry[name], str) or not entry[name].strip():
+                raise CalculationConfigError(f"{prefix}.{name} must be a non-empty string")
+        for name, expected in (
+            ("type", "material_consumption"), ("version", "1"),
+            ("output_metric", "material_consumption"), ("output_unit", "kg"),
+            ("group_by", "production_order"),
+        ):
+            if entry[name] != expected:
+                raise CalculationConfigError(f"{prefix}.{name}: only {expected!r} is supported")
+        entry["gate_threshold"] = finite_number(entry["gate_threshold"], f"{prefix}.gate_threshold")
+        entry["max_sample_gap_seconds"] = finite_number(
+            entry["max_sample_gap_seconds"], f"{prefix}.max_sample_gap_seconds", positive=True,
+        )
+        if entry["id"] in instances:
+            raise CalculationConfigError("Calculation ids must be unique")
+        instances[entry["id"]] = MaterialCalculationConfig(**entry)
+    if calculation_id is not None:
+        if calculation_id not in instances:
+            raise CalculationConfigError("Requested calculation id was not found")
+        return instances[calculation_id]
+    if len(instances) != 1:
+        raise CalculationConfigError("Multiple calculations: specify --calculation-id")
+    return next(iter(instances.values()))

src/production_analytics/cli/__main__.py

@@ -1,4 +1,4 @@
-"""Command-line entry point for safe, read-only API exploration."""
+"""Command-line entry point for exploration and production polling."""
 
 from __future__ import annotations
 
@@ -70,6 +70,14 @@ def _parser() -> argparse.ArgumentParser:
         default=Path("secrets/enlyze.env"),
         help="Local dotenv-style secret file (default: secrets/enlyze.env)",
     )
+    run = namespaces.add_parser("run", help="Foreground production runners")
+    runners = run.add_subparsers(dest="command", required=True)
+    material = runners.add_parser("material-poll", help="Continuously poll material consumption")
+    material.add_argument("--config", type=Path, required=True)
+    material.add_argument("--calculation-id")
+    material.add_argument("--poll-interval-seconds", type=float, default=10.0)
+    material.add_argument("--state-directory", type=Path, default=Path("data/state/material"))
+    material.add_argument("--secrets-file", type=Path, default=Path("secrets/enlyze.env"))
     return parser
 
 
@@ -83,8 +91,36 @@ def _write_fixture(directory: Path, name: str, payload: object) -> Path:
     return destination
 
 
+def _run_material(args: argparse.Namespace) -> int:
+    from production_analytics.calculations.config import (
+        CalculationConfigError,
+        load_material_calculation,
+    )
+    from production_analytics.enlyze.exploration import ConfigurationError
+    from production_analytics.service.material_runtime import build_material_runner
+
+    try:
+        calculation = load_material_calculation(args.config, args.calculation_id)
+        runner = build_material_runner(
+            calculation, poll_interval_seconds=args.poll_interval_seconds,
+            state_directory=args.state_directory, secrets_file=args.secrets_file,
+        )
+        runner.run()
+    except (CalculationConfigError, ConfigurationError) as exc:
+        print(f"Config/startup error: {exc}", file=sys.stderr)
+        return 2
+    except Exception as exc:
+        print(f"Startup/runtime error: {type(exc).__name__}", file=sys.stderr)
+        return 1
+    except KeyboardInterrupt:
+        return 0
+    return 0
+
+
 def main(argv: Sequence[str] | None = None) -> int:
     args = _parser().parse_args(argv)
+    if args.namespace == "run":
+        return _run_material(args)
     if args.namespace != "enlyze" or args.command not in {"raw", "timeseries"}:
         return 2
 

src/production_analytics/service/material_runner.py

@@ -0,0 +1,71 @@
+"""Sequential foreground polling with injectable time and output."""
+
+import sys
+import time
+from collections.abc import Callable
+from datetime import UTC, datetime
+from typing import TextIO
+
+from production_analytics.calculations.config import finite_number
+from production_analytics.enlyze.exploration import ConfigurationError, ExplorationError
+from production_analytics.service.material_polling import MaterialPollingService
+
+
+class MaterialStateError(RuntimeError):
+    """State operation failed; message contains only the operation and error class."""
+
+
+def utc_now() -> datetime:
+    return datetime.now(UTC)
+
+
+class MaterialPollingRunner:
+    def __init__(
+        self, service: MaterialPollingService, *, machine_id: str,
+        poll_interval_seconds: float, clock: Callable[[], datetime] = utc_now,
+        sleep: Callable[[float], None] = time.sleep,
+        stdout: TextIO | None = None, stderr: TextIO | None = None,
+    ) -> None:
+        self.service = service
+        self.machine_id = machine_id
+        self.interval = finite_number(poll_interval_seconds, "poll interval", positive=True)
+        self.clock = clock
+        self.sleep = sleep
+        self.stdout = stdout if stdout is not None else sys.stdout
+        self.stderr = stderr if stderr is not None else sys.stderr
+
+    def run(self) -> None:
+        try:
+            while True:
+                now = self.clock()
+                if now.tzinfo is None or now.utcoffset() is None:
+                    raise ConfigurationError("Runner clock must return timezone-aware timestamps")
+                now = now.astimezone(UTC)
+                prefix = f"{now.isoformat()} machine={self.machine_id!r}"
+                try:
+                    result = self.service.poll_once(now=now)
+                except ConfigurationError:
+                    raise
+                except Exception as exc:
+                    if isinstance(exc, MaterialStateError):
+                        detail = str(exc)
+                    elif isinstance(exc, ExplorationError):
+                        detail = f"ENLYZE/gateway: {type(exc).__name__}"
+                    else:
+                        detail = f"gateway/poll cycle: {type(exc).__name__}"
+                    # Do not echo arbitrary exception messages or response bodies.
+                    print(f"{prefix} error {detail}", file=self.stderr, flush=True)
+                else:
+                    if result is None:
+                        detail = "no open Production Run / no eligible polling window"
+                    else:
+                        detail = (
+                            f"production_order={result.run.production_order!r} "
+                            f"consumption_kg={result.state.cumulative_consumption_kg:.9f} "
+                            f"running_seconds={result.state.integrated_running_seconds:.3f}"
+                        )
+                    print(f"{prefix} {detail}", file=self.stdout, flush=True)
+                # Fixed delay after completion, including errors; never catch up or overlap.
+                self.sleep(self.interval)
+        except KeyboardInterrupt:
+            print("Material polling stopped", file=self.stdout, flush=True)

src/production_analytics/service/material_runtime.py

@@ -0,0 +1,78 @@
+"""Compose the configured live material polling application."""
+
+import math
+import os
+import tempfile
+from pathlib import Path
+from urllib.parse import urlsplit
+
+from production_analytics.calculations.config import MaterialCalculationConfig, finite_number
+from production_analytics.enlyze.exploration import (
+    ConfigurationError,
+    ExplorationClient,
+    ExplorationSettings,
+    load_secret_file,
+)
+from production_analytics.enlyze.gateway import EnlyzeApiGateway
+from production_analytics.service.material_polling import (
+    MaterialPollingService,
+    MaterialPollingState,
+)
+from production_analytics.service.material_runner import MaterialPollingRunner, MaterialStateError
+from production_analytics.service.material_state_store import JsonMaterialStateStore
+
+
+class _ReportingStateStore:
+    def __init__(self, store: JsonMaterialStateStore) -> None:
+        self.store = store
+
+    def load(self, machine_id: str, production_order: str) -> MaterialPollingState | None:
+        try:
+            return self.store.load(machine_id, production_order)
+        except Exception as exc:
+            raise MaterialStateError(f"state loading: {type(exc).__name__}") from exc
+
+    def save(
+        self, machine_id: str, production_order: str, state: MaterialPollingState,
+    ) -> None:
+        try:
+            self.store.save(machine_id, production_order, state)
+        except Exception as exc:
+            raise MaterialStateError(f"state saving: {type(exc).__name__}") from exc
+
+
+def build_material_runner(
+    calculation: MaterialCalculationConfig, *, poll_interval_seconds: float,
+    state_directory: Path, secrets_file: Path,
+) -> MaterialPollingRunner:
+    finite_number(poll_interval_seconds, "poll interval", positive=True)
+    environment = dict(os.environ)
+    environment.update(load_secret_file(secrets_file))
+    settings = ExplorationSettings.from_environment(environment)
+    parsed = urlsplit(settings.base_url)
+    if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username:
+        raise ConfigurationError(
+            "ENLYZE_BASE_URL must be an HTTP(S) server URL without credentials"
+	)
+    if not math.isfinite(settings.timeout_seconds):
+        raise ConfigurationError("ENLYZE_HTTP_TIMEOUT_SECONDS must be finite")
+    try:
+        state_directory.mkdir(parents=True, exist_ok=True)
+        with tempfile.TemporaryFile(dir=state_directory):
+            pass
+    except OSError as exc:
+        raise ConfigurationError(
+            f"State directory startup check failed: {type(exc).__name__}"
+        ) from exc
+    service = MaterialPollingService(
+        gateway=EnlyzeApiGateway(ExplorationClient(settings)),
+        state_store=_ReportingStateStore(JsonMaterialStateStore(state_directory)),
+        machine_id=calculation.machine_ref,
+        rate_variable_id=calculation.rate_signal_ref,
+        gate_variable_id=calculation.gate_signal_ref,
+        gate_threshold=calculation.gate_threshold,
+        max_sample_gap_seconds=calculation.max_sample_gap_seconds,
+    )
+    return MaterialPollingRunner(
+        service, machine_id=calculation.machine_ref, poll_interval_seconds=poll_interval_seconds,
+    )