146 lines
6.7 KiB
Python
146 lines
6.7 KiB
Python
"""Strict configuration for executable material-consumption instances."""
|
|
|
|
from dataclasses import MISSING, 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
|
|
source_mode: str = "direct_mass_rate"
|
|
application_signal_refs: tuple[str, ...] = ()
|
|
erp_workplace: str = ""
|
|
production_order_format: 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")
|
|
allowed = {field.name for field in fields(MaterialCalculationConfig)}
|
|
required = {field.name for field in fields(MaterialCalculationConfig)
|
|
if field.default is MISSING}
|
|
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")
|
|
area = entry.get("source_mode", "direct_mass_rate") == "area_application"
|
|
if area:
|
|
entry.setdefault("rate_signal_ref", "unused")
|
|
missing = required - entry.keys()
|
|
if missing:
|
|
raise CalculationConfigError(f"{prefix}: missing fields: {', '.join(sorted(missing))}")
|
|
if entry.keys() - allowed:
|
|
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,
|
|
)
|
|
mode = entry.get("source_mode", "direct_mass_rate")
|
|
if not isinstance(mode, str) or mode not in {"direct_mass_rate", "area_application"}:
|
|
raise CalculationConfigError("Unsupported material source_mode")
|
|
if area:
|
|
refs = entry.get("application_signal_refs")
|
|
if (not isinstance(refs, list) or not refs
|
|
or any(not isinstance(ref, str) or not ref.strip() for ref in refs)
|
|
or len(set(refs)) != len(refs)):
|
|
raise CalculationConfigError(
|
|
"application_signal_refs must be unique signal strings"
|
|
)
|
|
if entry["rate_signal_ref"] != "unused":
|
|
raise CalculationConfigError("area_application cannot specify rate_signal_ref")
|
|
for name in ("erp_workplace", "production_order_format"):
|
|
if not isinstance(entry.get(name), str) or not entry[name].strip():
|
|
raise CalculationConfigError(f"area_application requires {name}")
|
|
from production_analytics.context import build_enlyze_production_order
|
|
try:
|
|
build_enlyze_production_order("0", entry["production_order_format"])
|
|
except ValueError as exc:
|
|
raise CalculationConfigError("Invalid production_order_format") from exc
|
|
entry["application_signal_refs"] = tuple(refs)
|
|
elif any(name in entry for name in (
|
|
"application_signal_refs", "erp_workplace", "production_order_format",
|
|
)):
|
|
raise CalculationConfigError("Area context fields require area_application")
|
|
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()))
|