128 lines
5.1 KiB
Python
128 lines
5.1 KiB
Python
"""Configuration for reusable channel-level extruder consumption."""
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from production_analytics.calculations.config import CalculationConfigError, finite_number
|
|
from production_analytics.service.channel_material import ChannelDefinition
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ChannelMaterialCalculationConfig:
|
|
id: str
|
|
machine_ref: str
|
|
max_sample_gap_seconds: float
|
|
percentage_tolerance: float
|
|
channels: tuple[ChannelDefinition, ...]
|
|
material_names: dict[str, str]
|
|
|
|
|
|
def load_channel_material_calculation(path: str | Path) -> ChannelMaterialCalculationConfig:
|
|
try:
|
|
document = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, yaml.YAMLError) as exc:
|
|
raise CalculationConfigError("Cannot read channel material configuration") from exc
|
|
if (
|
|
not isinstance(document, dict)
|
|
or set(document) != {"calculations"}
|
|
or not isinstance(document["calculations"], list)
|
|
):
|
|
raise CalculationConfigError("Configuration must contain only calculations")
|
|
if len(document["calculations"]) != 1:
|
|
raise CalculationConfigError("Channel configuration requires one calculation")
|
|
entry = document["calculations"][0]
|
|
required = {
|
|
"id",
|
|
"type",
|
|
"version",
|
|
"machine_ref",
|
|
"max_sample_gap_seconds",
|
|
"percentage_tolerance",
|
|
"extruders",
|
|
}
|
|
if (
|
|
not isinstance(entry, dict)
|
|
or set(entry) - (required | {"material_names"})
|
|
or required - set(entry)
|
|
):
|
|
raise CalculationConfigError("Invalid channel calculation fields")
|
|
if entry["type"] != "channel_material_consumption" or entry["version"] != "1":
|
|
raise CalculationConfigError("Unsupported channel calculation type or version")
|
|
if not all(isinstance(entry[x], str) and entry[x].strip() for x in ("id", "machine_ref")):
|
|
raise CalculationConfigError("id and machine_ref must be non-empty strings")
|
|
channels = []
|
|
extruders = entry["extruders"]
|
|
if not isinstance(extruders, list) or not extruders:
|
|
raise CalculationConfigError("extruders must be a non-empty list")
|
|
for extruder in extruders:
|
|
if not isinstance(extruder, dict) or set(extruder) != {
|
|
"id",
|
|
"total_rate_signal_ref",
|
|
"channels",
|
|
}:
|
|
raise CalculationConfigError("Invalid extruder fields")
|
|
if not all(
|
|
isinstance(extruder[x], str) and extruder[x].strip()
|
|
for x in ("id", "total_rate_signal_ref")
|
|
):
|
|
raise CalculationConfigError("Extruder id and total signal must be strings")
|
|
rows = extruder["channels"]
|
|
if not isinstance(rows, list) or len(rows) != 7:
|
|
raise CalculationConfigError("Each extruder requires exactly seven channels")
|
|
for row in rows:
|
|
allowed = {
|
|
"id",
|
|
"status_signal_ref",
|
|
"percentage_signal_ref",
|
|
"screw_speed_signal_ref",
|
|
"material_number_signal_ref",
|
|
}
|
|
if (
|
|
not isinstance(row, dict)
|
|
or set(row) - allowed
|
|
or {"id", "status_signal_ref", "percentage_signal_ref", "screw_speed_signal_ref"}
|
|
- set(row)
|
|
):
|
|
raise CalculationConfigError("Invalid dosing channel fields")
|
|
if not all(
|
|
isinstance(row[x], str) and row[x].strip()
|
|
for x in (
|
|
"id",
|
|
"status_signal_ref",
|
|
"percentage_signal_ref",
|
|
"screw_speed_signal_ref",
|
|
)
|
|
):
|
|
raise CalculationConfigError("Dosing signal references must be non-empty strings")
|
|
material = row.get("material_number_signal_ref")
|
|
if material is not None and (not isinstance(material, str) or not material.strip()):
|
|
raise CalculationConfigError("material signal must be a string or null")
|
|
channels.append(
|
|
ChannelDefinition(
|
|
extruder["id"],
|
|
row["id"],
|
|
extruder["total_rate_signal_ref"],
|
|
row["status_signal_ref"],
|
|
row["percentage_signal_ref"],
|
|
row["screw_speed_signal_ref"],
|
|
material,
|
|
)
|
|
)
|
|
if len({(c.extruder, c.channel) for c in channels}) != len(channels):
|
|
raise CalculationConfigError("Extruder/channel identities must be unique")
|
|
names = entry.get("material_names", {})
|
|
if not isinstance(names, dict) or any(
|
|
not isinstance(k, str) or not isinstance(v, str) for k, v in names.items()
|
|
):
|
|
raise CalculationConfigError("material_names must map strings to strings")
|
|
return ChannelMaterialCalculationConfig(
|
|
entry["id"],
|
|
entry["machine_ref"],
|
|
finite_number(entry["max_sample_gap_seconds"], "max_sample_gap_seconds", positive=True),
|
|
finite_number(entry["percentage_tolerance"], "percentage_tolerance"),
|
|
tuple(channels),
|
|
names,
|
|
)
|