134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
"""Run one finite RF-001 RX child and persist start/final evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
from typing import Mapping
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def atomic_json(path: Path, value: Mapping[str, object]) -> None:
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(
|
|
json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
while block := stream.read(1024 * 1024):
|
|
digest.update(block)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def run_capture(
|
|
executable: Path,
|
|
arguments_path: Path,
|
|
capture_path: Path,
|
|
stdout_path: Path,
|
|
stderr_path: Path,
|
|
started_path: Path,
|
|
result_path: Path,
|
|
) -> int:
|
|
arguments_value = json.loads(arguments_path.read_text(encoding="utf-8-sig"))
|
|
if not isinstance(arguments_value, list) or not all(
|
|
isinstance(item, str) for item in arguments_value
|
|
):
|
|
raise ValueError("arguments JSON must be an array of strings")
|
|
started_at = utc_now()
|
|
child: subprocess.Popen[bytes] | None = None
|
|
exit_code: int | None = None
|
|
runner_error: str | None = None
|
|
try:
|
|
with stdout_path.open("xb") as stdout_stream, stderr_path.open(
|
|
"xb"
|
|
) as stderr_stream:
|
|
child = subprocess.Popen(
|
|
[str(executable), *arguments_value],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=stdout_stream,
|
|
stderr=stderr_stream,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
close_fds=True,
|
|
)
|
|
atomic_json(
|
|
started_path,
|
|
{
|
|
"state": "RUNNING",
|
|
"pid": int(child.pid),
|
|
"started_at": started_at,
|
|
"capture_path": str(capture_path),
|
|
"rf_tx": 0,
|
|
},
|
|
)
|
|
exit_code = int(child.wait())
|
|
except Exception as error: # persisted at the process boundary
|
|
runner_error = f"{type(error).__name__}: {error}"
|
|
finished_at = utc_now()
|
|
capture_size = capture_path.stat().st_size if capture_path.is_file() else 0
|
|
capture_hash: str | None = None
|
|
if capture_path.is_file():
|
|
try:
|
|
capture_hash = sha256_file(capture_path)
|
|
except OSError as error:
|
|
detail = f"{type(error).__name__}: {error}"
|
|
runner_error = detail if runner_error is None else f"{runner_error}; {detail}"
|
|
duration = (
|
|
datetime.fromisoformat(finished_at) - datetime.fromisoformat(started_at)
|
|
).total_seconds()
|
|
atomic_json(
|
|
result_path,
|
|
{
|
|
"pid": int(child.pid) if child is not None else None,
|
|
"started_at": started_at,
|
|
"finished_at": finished_at,
|
|
"duration": duration,
|
|
"exit_code": exit_code,
|
|
"capture_path": str(capture_path),
|
|
"capture_size": capture_size,
|
|
"sha256": capture_hash,
|
|
"runner_error": runner_error,
|
|
"rf_tx": 0,
|
|
},
|
|
)
|
|
return 255 if exit_code is None else exit_code
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--executable", required=True, type=Path)
|
|
parser.add_argument("--arguments-json", required=True, type=Path)
|
|
parser.add_argument("--capture-path", required=True, type=Path)
|
|
parser.add_argument("--stdout-path", required=True, type=Path)
|
|
parser.add_argument("--stderr-path", required=True, type=Path)
|
|
parser.add_argument("--started-status", required=True, type=Path)
|
|
parser.add_argument("--result-status", required=True, type=Path)
|
|
arguments = parser.parse_args()
|
|
raise SystemExit(
|
|
run_capture(
|
|
arguments.executable,
|
|
arguments.arguments_json,
|
|
arguments.capture_path,
|
|
arguments.stdout_path,
|
|
arguments.stderr_path,
|
|
arguments.started_status,
|
|
arguments.result_status,
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|