67 lines
2.7 KiB
Python
67 lines
2.7 KiB
Python
"""CLI for the isolated Brain local-event vertical slice."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from yovision_brain.config import ConfigError
|
|
from yovision_brain.decode import DecoderError
|
|
from yovision_brain.events import JsonLinesSink
|
|
from yovision_brain.input import InputError
|
|
from yovision_brain.integration.event_export import build_event_export_sink
|
|
from yovision_brain.rules import RuleConfigError
|
|
|
|
from .runner import run_pipeline
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Run the isolated Brain internal-event pipeline")
|
|
parser.add_argument("--config", required=True, help="explicit Brain-internal JSON configuration")
|
|
parser.add_argument("--output", default="-", help="JSON Lines output file, or - for stdout")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
config_path = Path(args.config)
|
|
try:
|
|
raw = json.loads(config_path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
print(json.dumps({"status": "error", "message": "Brain internal config cannot be read"}), file=sys.stderr)
|
|
return 2
|
|
if not isinstance(raw, dict):
|
|
print(json.dumps({"status": "error", "message": "Brain internal config must be an object"}), file=sys.stderr)
|
|
return 2
|
|
|
|
stream = sys.stdout
|
|
owned_stream = None
|
|
try:
|
|
export_sink = build_event_export_sink(raw.get("event_export"), base_dir=config_path.parent)
|
|
if export_sink is None and args.output != "-":
|
|
try:
|
|
owned_stream = Path(args.output).open("w", encoding="utf-8", newline="\n")
|
|
except OSError:
|
|
print(json.dumps({"status": "error", "message": "event output cannot be opened"}), file=sys.stderr)
|
|
return 2
|
|
stream = owned_stream
|
|
sink = export_sink if export_sink is not None else JsonLinesSink(stream)
|
|
summary = run_pipeline(raw, sink, base_dir=config_path.parent)
|
|
except (ConfigError, DecoderError, InputError, RuleConfigError, RuntimeError, ValueError) as exc:
|
|
print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr)
|
|
return 3
|
|
except KeyboardInterrupt:
|
|
print(json.dumps({"status": "cancelled"}), file=sys.stderr)
|
|
return 130
|
|
finally:
|
|
if owned_stream is not None:
|
|
owned_stream.close()
|
|
print(json.dumps({"status": summary.status, "frames": summary.frames, "detections": summary.detections, "events": summary.events}, sort_keys=True), file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|