80 lines
3.2 KiB
Python
80 lines
3.2 KiB
Python
"""Helpers for writing runtime configs and preprocessing bundles."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import suppress
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from python_app.models.run_config_model import RunConfigModel, parse_combos_from_text
|
|
from python_app.orchestration.preprocess_assets import (
|
|
PREPROCESS_ASSET_KEYS,
|
|
PREPROCESS_ASSET_SPECS,
|
|
preprocess_asset_model,
|
|
runtime_preprocess_asset_keys,
|
|
)
|
|
from python_app.storage.npz_store import NpzStore
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConfigWriter:
|
|
"""Write runtime artifacts consumed by C++ processes."""
|
|
|
|
def __init__(self, runtime_dir: Path) -> None:
|
|
"""Create writer rooted at runtime directory."""
|
|
self._runtime_dir = runtime_dir
|
|
self._runtime_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def prepare_preprocess_bundles(
|
|
self,
|
|
store: NpzStore,
|
|
radar_key: str,
|
|
config: RunConfigModel,
|
|
) -> None:
|
|
"""Export selected preprocess sets into runtime bundles and update config paths."""
|
|
for key in PREPROCESS_ASSET_KEYS:
|
|
preprocess_asset_model(config, key).bundle_path = ""
|
|
|
|
for key in runtime_preprocess_asset_keys(config):
|
|
spec = PREPROCESS_ASSET_SPECS[key]
|
|
asset = preprocess_asset_model(config, key)
|
|
bundle_path = self._runtime_dir / spec.runtime_filename
|
|
logger.debug("Exporting preprocess bundle %s (set=%s) -> %s", key, asset.set_name, bundle_path)
|
|
store.export_set_bundle(spec.set_kind, radar_key, asset.set_name, bundle_path)
|
|
asset.bundle_path = str(bundle_path)
|
|
|
|
def write(self, config: RunConfigModel, output_path: Path) -> Path:
|
|
"""Atomically write the run configuration JSON file.
|
|
|
|
Dump to a sibling ``.tmp``, flush + ``fsync`` to durably commit the bytes on
|
|
the SD-card-backed Pi, then ``os.replace()`` atomically onto the destination.
|
|
A C++ consumer can therefore never observe a half-written config (which would
|
|
abort it with an opaque JSON parse error), even across a crash or power loss.
|
|
"""
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
# allow_nan=False: a stray NaN/Infinity must fail loudly here in Python
|
|
# rather than serialize to a non-standard token that aborts every C++
|
|
# consumer at startup with an opaque JSON parse error.
|
|
serialized = json.dumps(config.to_dict(), indent=2, allow_nan=False)
|
|
tmp_path = output_path.with_suffix(output_path.suffix + ".tmp")
|
|
try:
|
|
with open(tmp_path, "w", encoding="utf-8") as handle:
|
|
handle.write(serialized)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(tmp_path, output_path)
|
|
except Exception:
|
|
logger.exception("Failed to write run config to %s; removing temp file", output_path)
|
|
# Never leave a half-written .tmp behind on a write/fsync failure.
|
|
with suppress(OSError):
|
|
tmp_path.unlink()
|
|
raise
|
|
logger.debug("Wrote run config to %s (%d bytes)", output_path, len(serialized))
|
|
return output_path
|
|
|
|
|
|
__all__ = ["ConfigWriter", "parse_combos_from_text"]
|