155 lines
4.8 KiB
Python
155 lines
4.8 KiB
Python
"""Publish a model-free READY/COMPLETE smoke sequence to the RX result branch."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from dataclasses import replace
|
|
from datetime import datetime, timedelta, timezone
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from typing import Callable
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from tools.rf_handoff_channel import ( # noqa: E402
|
|
CHANNEL_TEST_TASK_ID,
|
|
COMPLETE_STATE,
|
|
READY_STATE,
|
|
GitResultChannel,
|
|
PublishOutcome,
|
|
RxHandoffResult,
|
|
)
|
|
|
|
|
|
def _git_head(repository: Path) -> str:
|
|
return subprocess.run(
|
|
["git", "rev-parse", "HEAD"],
|
|
cwd=repository,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
check=True,
|
|
).stdout.strip()
|
|
|
|
|
|
def run_smoke(
|
|
repository: Path,
|
|
*,
|
|
remote: str = "origin",
|
|
branch: str = "rx-node",
|
|
source_head: str | None = None,
|
|
hold_seconds: float = 7.0,
|
|
sleep_function: Callable[[float], None] = time.sleep,
|
|
) -> tuple[PublishOutcome, PublishOutcome, RxHandoffResult]:
|
|
"""Publish fake READY then COMPLETE; never access SDR or open a TX gate."""
|
|
|
|
repository = repository.resolve()
|
|
observed_source_head = source_head or _git_head(repository)
|
|
channel = GitResultChannel(repository, remote=remote, branch=branch)
|
|
channel.fetch()
|
|
|
|
with tempfile.TemporaryDirectory(prefix="rf-handoff-smoke-") as directory:
|
|
fake_capture = Path(directory) / "model_free_fake_capture.txt"
|
|
started = datetime.now(timezone.utc)
|
|
fake_capture.write_text(
|
|
"RF_HANDOFF_CHANNEL_TEST_001\nREADY\nRF TX = 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
published = datetime.now(timezone.utc)
|
|
ready = RxHandoffResult(
|
|
schema_version=1,
|
|
task_id=CHANNEL_TEST_TASK_ID,
|
|
state=READY_STATE,
|
|
rx_ready_for_tx=True,
|
|
process_alive=True,
|
|
file_growing=True,
|
|
capture_pid=max(1, os.getpid()),
|
|
published_at=published.isoformat(),
|
|
capture_started_at=started.isoformat(),
|
|
expected_capture_end=(started + timedelta(seconds=120)).isoformat(),
|
|
capture_path=str(fake_capture.resolve()),
|
|
common_clock_verified=False,
|
|
source_head=observed_source_head,
|
|
rf_tx=0,
|
|
exit_code=None,
|
|
file_size=None,
|
|
duration_seconds=None,
|
|
overruns=None,
|
|
usb_errors=None,
|
|
file_errors=None,
|
|
sha256=None,
|
|
clkin_after="NOT_RUN",
|
|
rf_tx_observed="UNKNOWN",
|
|
)
|
|
ready_outcome = channel.publish(ready)
|
|
sleep_function(hold_seconds)
|
|
|
|
fake_capture.write_text(
|
|
"RF_HANDOFF_CHANNEL_TEST_001\nCOMPLETE\nRF TX = 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
finished = datetime.now(timezone.utc)
|
|
content = fake_capture.read_bytes()
|
|
complete = replace(
|
|
ready,
|
|
state=COMPLETE_STATE,
|
|
rx_ready_for_tx=False,
|
|
process_alive=False,
|
|
file_growing=False,
|
|
published_at=finished.isoformat(),
|
|
exit_code=0,
|
|
file_size=len(content),
|
|
duration_seconds=(finished - started).total_seconds(),
|
|
overruns=0,
|
|
usb_errors=0,
|
|
file_errors=0,
|
|
sha256=hashlib.sha256(content).hexdigest(),
|
|
)
|
|
complete_outcome = channel.publish(complete)
|
|
return ready_outcome, complete_outcome, complete
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--repository", type=Path, default=ROOT)
|
|
parser.add_argument("--remote", default="origin")
|
|
parser.add_argument("--result-branch", default="rx-node")
|
|
parser.add_argument("--source-head")
|
|
parser.add_argument("--hold-seconds", type=float, default=7.0)
|
|
return parser
|
|
|
|
|
|
def main() -> None:
|
|
arguments = build_parser().parse_args()
|
|
if arguments.hold_seconds < 0:
|
|
raise SystemExit("--hold-seconds cannot be negative")
|
|
ready, complete, result = run_smoke(
|
|
arguments.repository,
|
|
remote=arguments.remote,
|
|
branch=arguments.result_branch,
|
|
source_head=arguments.source_head,
|
|
hold_seconds=arguments.hold_seconds,
|
|
)
|
|
print("TASK_ID =")
|
|
print(CHANNEL_TEST_TASK_ID)
|
|
print(f"READY PUBLISH = {ready.status}")
|
|
print(f"READY REMOTE HEAD = {ready.remote_head}")
|
|
print(f"COMPLETE PUBLISH = {complete.status}")
|
|
print(f"COMPLETE REMOTE HEAD = {complete.remote_head}")
|
|
print(f"FINAL STATE = {result.state}")
|
|
print("REAL RX = NOT_RUN")
|
|
print("RF TX = 0")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|