Claude (homelab agent)andClaude Opus 5.5
4f31784a49
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
"""Scanner health: systemd unit state, HackRF presence, recording rate."""
|
|
import glob
|
|
import os
|
|
import subprocess
|
|
import time
|
|
from datetime import datetime
|
|
|
|
import json
|
|
|
|
from radio.config import REC_DIR, SCANNER_UNIT, TR_DIR
|
|
from radio.recordings import FILE_RE, cls_index, list_days
|
|
|
|
_unit = {"t": 0, "v": "unknown"}
|
|
|
|
|
|
def unit_state():
|
|
"""`systemctl is-active` of the scanner, cached for 5 s."""
|
|
now = time.time()
|
|
if now - _unit["t"] > 5:
|
|
try:
|
|
v = subprocess.run(["systemctl", "is-active", SCANNER_UNIT], capture_output=True,
|
|
text=True, timeout=5).stdout.strip()
|
|
except (subprocess.SubprocessError, OSError):
|
|
v = "unknown"
|
|
_unit.update(t=now, v=v)
|
|
return _unit["v"]
|
|
|
|
|
|
def hackrf_present():
|
|
"""HackRF One on the USB bus (1d50:6089), read from sysfs: no tools, works for any user."""
|
|
for d in glob.glob("/sys/bus/usb/devices/*/"):
|
|
try:
|
|
if open(d + "idVendor").read().strip() == "1d50" and open(d + "idProduct").read().strip() == "6089":
|
|
return True
|
|
except OSError:
|
|
continue
|
|
return False
|
|
|
|
|
|
def status():
|
|
st = unit_state()
|
|
last, n_hour, n_voice_hour = None, 0, 0
|
|
now = time.time()
|
|
cutoff = now - 3600
|
|
for day in list_days()[:2]:
|
|
ddir = os.path.join(REC_DIR, day)
|
|
idx = cls_index(day)
|
|
for sub in ("", "noise/"):
|
|
try:
|
|
names = [n for n in os.listdir(os.path.join(ddir, sub) if sub else ddir) if FILE_RE.match(n)]
|
|
except OSError:
|
|
continue
|
|
for n in names:
|
|
try:
|
|
mt = os.stat(os.path.join(ddir, sub + n)).st_mtime
|
|
except OSError:
|
|
continue
|
|
if mt > now + 60: # bogus clock (pre-NTP) files
|
|
continue
|
|
last = mt if last is None or mt > last else last
|
|
if mt >= cutoff:
|
|
n_hour += 1
|
|
if not sub and (idx.get(n) or {}).get("c") == "voice":
|
|
n_voice_hour += 1
|
|
hrf = hackrf_present()
|
|
age = now - last if last else None
|
|
# the unit is "active" even while it restarts every few seconds without a HackRF: judge by the radio itself
|
|
problem = ("HackRF не подключён" if not hrf else "сканер не запущен" if st != "active"
|
|
else "сканер не пишет" if age is None or age > 1800 else None)
|
|
return {
|
|
"scanner": st,
|
|
"hackrf": hrf,
|
|
"problem": problem,
|
|
"ok": problem is None,
|
|
"last_recording": datetime.fromtimestamp(last).astimezone().isoformat(timespec="seconds") if last else None,
|
|
"last_recording_age_s": max(0, int(now - last)) if last else None,
|
|
"recordings_last_hour": n_hour,
|
|
"voice_last_hour": n_voice_hour,
|
|
"transcribe": transcribe_status(),
|
|
}
|
|
|
|
|
|
def transcribe_status():
|
|
"""sdr-transcribe queue: pending, done, whether Whisper on the desktop answers; stale = worker not running."""
|
|
try:
|
|
with open(os.path.join(TR_DIR, "status.json"), encoding="utf-8") as fh:
|
|
s = json.load(fh)
|
|
except (OSError, ValueError):
|
|
return None
|
|
s["stale"] = time.time() - s.get("ts", 0) > 300
|
|
return s
|