66 lines
2.8 KiB
Python
66 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Replay common timestamps from ignored K7 run 00842 captures, without network access."""
|
|
|
|
import argparse
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from production_analytics.calculations import (
|
|
MaterialConsumptionIntegrator,
|
|
MaterialIntegratorConfig,
|
|
MaterialSample,
|
|
)
|
|
|
|
RATE_UUID = "c9d06af5-f6d6-4ede-b6c4-5a98bac77129" # Stundenleistung Anlage [kg/h]
|
|
GATE_UUID = "823867bb-f5d2-40eb-b875-657155addfd0" # Geschwindigkeit Gesamtanlage [m/min]
|
|
CAPTURE_DIRECTORY = Path("data/raw/enlyze")
|
|
|
|
|
|
def read_capture(path: Path, variable_uuid: str) -> dict[datetime, float]:
|
|
"""Read unique, ordered observations; refuse ambiguous duplicate source records."""
|
|
data = json.loads(path.read_text(encoding="utf-8"))["response"]["body"]["data"]
|
|
time_index = data["columns"].index("time")
|
|
value_index = data["columns"].index(variable_uuid)
|
|
values = {}
|
|
previous = None
|
|
for record in data["records"]:
|
|
timestamp = datetime.fromisoformat(record[time_index].replace("Z", "+00:00"))
|
|
if timestamp.utcoffset() is None:
|
|
raise ValueError(f"Naive timestamp in {path}")
|
|
timestamp = timestamp.astimezone(UTC)
|
|
if previous is not None and timestamp <= previous:
|
|
raise ValueError(f"Capture timestamps must be strictly increasing: {path}")
|
|
previous = timestamp
|
|
if record[value_index] is not None:
|
|
values[timestamp] = float(record[value_index])
|
|
return values
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--max-sample-gap-seconds", type=float, default=20.0)
|
|
args = parser.parse_args()
|
|
integrator = MaterialConsumptionIntegrator(
|
|
MaterialIntegratorConfig(0.5, args.max_sample_gap_seconds)
|
|
)
|
|
try:
|
|
rates = read_capture(CAPTURE_DIRECTORY / "k7-00842-throughput.raw.json", RATE_UUID)
|
|
gates = read_capture(CAPTURE_DIRECTORY / "k7-00842-speed.raw.json", GATE_UUID)
|
|
except (OSError, ValueError, KeyError) as error:
|
|
parser.exit(1, f"Cannot replay K7 captures: {error}\n")
|
|
# Dict insertion order preserves validated capture order. No resampling or filling.
|
|
samples = [MaterialSample(t, rate, gates[t]) for t, rate in rates.items() if t in gates]
|
|
if len(samples) < 2:
|
|
parser.exit(1, "Need at least two common non-null samples\n")
|
|
state = integrator.process_many(samples)
|
|
print(f"Common samples: {len(samples)}")
|
|
print(f"Maximum sample gap: {integrator.config.max_sample_gap_seconds:g} s")
|
|
print(f"Integrated running time: {state.integrated_running_seconds / 3600:.9f} h")
|
|
print(f"Cumulative material consumption: {state.cumulative_consumption_kg:.9f} kg")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|