vee
133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
"""Run the opt-in CH1 driver/body-CAN front-end HIL suite."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SRC = ROOT / "src"
|
|
SEAT_SRC = ROOT / "domains" / "seat" / "src"
|
|
for source in (SRC, SEAT_SRC):
|
|
if str(source) not in sys.path:
|
|
sys.path.insert(0, str(source))
|
|
|
|
from cognitive_agent.config_loader import load_local_env # noqa: E402
|
|
from cognitive_agent.infrastructure.can.hardware import VehicleCanSettings # noqa: E402
|
|
from cognitive_agent.testing.ch1_hil import ( # noqa: E402
|
|
Ch1HilRunner,
|
|
JsonApiClient,
|
|
load_suite,
|
|
observe_stationary_ch1,
|
|
require_explicit_hil_arm,
|
|
suite_cases,
|
|
suite_rejections,
|
|
validate_ui_contract,
|
|
)
|
|
|
|
|
|
DEFAULT_SUITE = ROOT / "tests" / "fixtures" / "ch1_vehicle_frontend_hil_suite.json"
|
|
DEFAULT_OUTPUT = ROOT / "artifacts" / "hil" / "ch1_vehicle_frontend_hil_latest.json"
|
|
def passive_reader(token: str | None):
|
|
"""Build one fresh observer that opens CH1 only and never transmits."""
|
|
|
|
settings = VehicleCanSettings.from_env(ROOT)
|
|
settings.authorize(token)
|
|
if settings.channel != 1 or settings.required_channel != 1:
|
|
raise RuntimeError("passive preflight refused a non-CH1 channel")
|
|
|
|
def read() -> dict:
|
|
bridge = settings.bridge([settings.channel_config()])
|
|
try:
|
|
bus = bridge.channel_bus(1)
|
|
return observe_stationary_ch1(bus, timeout_seconds=5.0)
|
|
finally:
|
|
bridge.close()
|
|
|
|
return read
|
|
|
|
|
|
def offline_validate(suite_path: Path) -> int:
|
|
suite = load_suite(suite_path)
|
|
validate_ui_contract(
|
|
"self_orchestration",
|
|
(ROOT / "apps" / "web-console" / "app.js").read_text(encoding="utf-8"),
|
|
)
|
|
validate_ui_contract(
|
|
"self_extension",
|
|
(ROOT / "apps" / "web-console" / "self-extension.js").read_text(
|
|
encoding="utf-8"
|
|
),
|
|
)
|
|
counts = {
|
|
module: {
|
|
"total": len(suite_cases(suite, module)),
|
|
"physical": sum(
|
|
item.get("hil", True) for item in suite_cases(suite, module)
|
|
),
|
|
"contract_only": sum(
|
|
not item.get("hil", True) for item in suite_cases(suite, module)
|
|
),
|
|
"offline_rejections": len(suite_rejections(suite, module)),
|
|
}
|
|
for module in ("self_orchestration", "self_extension")
|
|
}
|
|
print(json.dumps({"status": "validated_offline", "cases": counts}, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="CH1-only driver/body-CAN front-end HIL verification"
|
|
)
|
|
parser.add_argument("--suite", type=Path, default=DEFAULT_SUITE)
|
|
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
|
parser.add_argument("--base-url", default="http://127.0.0.1:8000")
|
|
parser.add_argument("--offline-validate", action="store_true")
|
|
parser.add_argument(
|
|
"--confirm",
|
|
default="",
|
|
help="Live mode requires the exact text CH1_DRIVER_ONLY.",
|
|
)
|
|
args = parser.parse_args()
|
|
if args.offline_validate:
|
|
return offline_validate(args.suite)
|
|
|
|
# Loading the repository launch environment does not arm HIL by itself;
|
|
# the dedicated environment and exact CLI confirmation are both mandatory.
|
|
load_local_env(ROOT / ".env", override=True)
|
|
require_explicit_hil_arm(os.environ, confirmation=args.confirm)
|
|
suite = load_suite(args.suite)
|
|
token = os.getenv("COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN") or None
|
|
client_id = os.getenv("COGNITIVE_AGENT_CH1_HIL_CLIENT_ID") or None
|
|
cookie = os.getenv("COGNITIVE_AGENT_CH1_HIL_COOKIE") or None
|
|
api = JsonApiClient(args.base_url, cookie=cookie)
|
|
reader = passive_reader(token)
|
|
|
|
runner = Ch1HilRunner(
|
|
suite,
|
|
api,
|
|
reader,
|
|
confirmation=args.confirm,
|
|
approval_token=token,
|
|
hardware_client_id=client_id,
|
|
)
|
|
report = runner.run()
|
|
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(
|
|
json.dumps(report, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
print(json.dumps(report["summary"], ensure_ascii=False))
|
|
print(str(args.output.resolve()))
|
|
return 0 if report["summary"]["failed"] == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|