mirror of
https://github.com/DeeJanuz/frametop.git
synced 2026-10-06 08:00:09 +02:00
DeeJanuzandClaude Opus 5.5
3ebae6a88f
After the headset wakes, XRService sometimes fails to load the colour module's VCINT FPGA image; then only the two side cameras run, without the IR light, and the tracker finds no hands. hands/camcheck.py reads XRService's log, the video nodes it holds and ft-camd's ring, and says ok, degraded or unknown. The recorder won't start while degraded (--ignore-cameras overrides it), offers a confirmed SteamVR restart, and stops the first hand-size step when the tracker sees no hand at all. ft-camwatch (unit file only, not enabled) follows the log, notifies, and with CAMWATCH_AUTO_RESTART=1 restarts SteamVR when the headset isn't worn and nothing else uses VR. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
233 lines
10 KiB
Python
233 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for hands/ft-camwatch: its decision logic with made-up inputs (worn or not, VR apps,
|
|
remote viewers, cooldown, once per failure), following a growing log, and its settings. It
|
|
never notifies or restarts anything here.
|
|
|
|
python3 hands/tests/test_camwatch.py
|
|
"""
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
HANDS = os.path.dirname(HERE)
|
|
sys.path.insert(0, HANDS)
|
|
sys.path.insert(0, HERE)
|
|
from test_camcheck import CLOSE, FAIL, GOOD_OPEN, RESUME, RESUME_OK, START # noqa: E402
|
|
|
|
loader = importlib.machinery.SourceFileLoader("camwatch", os.path.join(HANDS, "ft-camwatch"))
|
|
spec = importlib.util.spec_from_loader("camwatch", loader)
|
|
cw = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(cw)
|
|
|
|
ON = dict(cw.DEFAULTS, CAMWATCH_AUTO_RESTART="1", CAMWATCH_IDLE_S="60", CAMWATCH_COOLDOWN_MIN="30")
|
|
OFF = dict(cw.DEFAULTS)
|
|
IDLE = {"steamvr": True, "worn": False, "vr_apps": [], "remote": []}
|
|
WORN = dict(IDLE, worn=True)
|
|
F1, F2 = "a.log@17:02:47", "b.log@18:10:00"
|
|
|
|
|
|
def kinds(acts):
|
|
return [k for k, _ in acts]
|
|
|
|
|
|
class Decide(unittest.TestCase):
|
|
def setUp(self):
|
|
self.mem = cw.new_memory()
|
|
|
|
def d(self, now, failure, env=IDLE, conf=ON):
|
|
return cw.decide(now, failure, env, self.mem, conf)
|
|
|
|
def test_nothing_wrong(self):
|
|
self.assertEqual(self.d(0, "", {}), [])
|
|
self.assertEqual(self.mem["pending"], "")
|
|
|
|
def test_notify_once(self):
|
|
acts = self.d(0, F1, WORN, OFF)
|
|
self.assertEqual(kinds(acts), ["log", "notify", "log"])
|
|
self.assertIn("CAMWATCH_AUTO_RESTART=1", acts[2][1])
|
|
self.assertEqual(self.d(2, F1, WORN, OFF), []) # same failure, same reason: quiet
|
|
self.assertEqual(self.d(4, F1, IDLE, OFF), []) # off: never restarts, whatever the state
|
|
self.assertEqual(self.d(1000, F1, IDLE, OFF), [])
|
|
|
|
def test_notify_off(self):
|
|
acts = self.d(0, F1, WORN, dict(OFF, CAMWATCH_NOTIFY="0"))
|
|
self.assertNotIn("notify", kinds(acts))
|
|
|
|
def test_never_while_worn(self):
|
|
self.d(0, F1, WORN)
|
|
for t in range(2, 4000, 2):
|
|
self.assertNotIn("restart", kinds(self.d(t, F1, WORN)))
|
|
self.assertIn("worn", self.mem["why"])
|
|
|
|
def test_restart_after_idle_then_once(self):
|
|
self.d(0, F1, WORN)
|
|
self.assertNotIn("restart", kinds(self.d(10, F1, IDLE))) # just taken off: wait
|
|
self.assertIn("waiting", self.mem["why"])
|
|
self.assertNotIn("restart", kinds(self.d(69, F1, IDLE)))
|
|
acts = self.d(70, F1, IDLE)
|
|
self.assertEqual(kinds(acts), ["log", "restart"])
|
|
self.assertEqual(acts[1][1], F1)
|
|
# the same failure again (the new XRService hasn't logged yet, or failed the same way)
|
|
acts = self.d(72, F1, IDLE)
|
|
self.assertNotIn("restart", kinds(acts))
|
|
self.assertIn("restart the headset", self.mem["why"])
|
|
for t in range(74, 10000, 500):
|
|
self.assertNotIn("restart", kinds(self.d(t, F1, IDLE)))
|
|
|
|
def test_put_on_resets_the_wait(self):
|
|
self.d(0, F1, IDLE)
|
|
self.d(50, F1, WORN) # back on at 50 s
|
|
self.assertNotIn("restart", kinds(self.d(70, F1, IDLE))) # off again: the 60 s start over
|
|
self.assertNotIn("restart", kinds(self.d(129, F1, IDLE)))
|
|
self.assertIn("restart", kinds(self.d(131, F1, IDLE)))
|
|
|
|
def test_cooldown(self):
|
|
self.d(0, F1, IDLE)
|
|
self.assertIn("restart", kinds(self.d(60, F1, IDLE)))
|
|
# The restart's new XRService fails too: a new failure, but within the cooldown.
|
|
self.d(90, "", IDLE)
|
|
acts = self.d(120, F2, IDLE)
|
|
self.assertIn("notify", kinds(acts))
|
|
self.assertNotIn("restart", kinds(acts))
|
|
self.assertNotIn("restart", kinds(self.d(60 + 30 * 60 - 2, F2, IDLE)))
|
|
self.assertIn("cooldown", self.mem["why"])
|
|
self.assertIn("restart", kinds(self.d(60 + 30 * 60, F2, IDLE)))
|
|
|
|
def test_vr_app_remote_and_no_steamvr(self):
|
|
for env, word in ((dict(IDLE, vr_apps=["AppId=450390"]), "VR app"),
|
|
(dict(IDLE, remote=["0A00000B:D2F0"]), "remote desktop"),
|
|
(dict(IDLE, steamvr=False), "SteamVR isn't running")):
|
|
with self.subTest(word=word):
|
|
self.mem = cw.new_memory()
|
|
for t in range(0, 600, 2):
|
|
self.assertNotIn("restart", kinds(self.d(t, F1, env)))
|
|
self.assertIn(word, self.mem["why"])
|
|
# once it's gone (and the headset still off), the restart happens
|
|
self.assertIn("restart", kinds(self.d(600, F1, IDLE)))
|
|
|
|
def test_resolved(self):
|
|
self.d(0, F1, WORN)
|
|
acts = self.d(10, "", {})
|
|
self.assertEqual(kinds(acts), ["log"])
|
|
self.assertIn("no longer current", acts[0][1])
|
|
self.assertEqual(self.mem["pending"], "")
|
|
self.assertEqual(self.d(12, "", {}), [])
|
|
|
|
def test_memory_round_trip(self):
|
|
tmp = tempfile.mkdtemp(prefix="camwatch-test-")
|
|
try:
|
|
path = os.path.join(tmp, "state", "camwatch.json")
|
|
self.d(0, F1, IDLE)
|
|
self.d(60, F1, IDLE)
|
|
cw.save_memory(self.mem, path)
|
|
mem = cw.load_memory(path)
|
|
self.assertEqual(mem["restarted"], {F1: 60})
|
|
self.assertEqual(mem["notified"], [F1])
|
|
# the watcher restarted: the same failure neither notifies nor restarts again
|
|
acts = cw.decide(200, F1, IDLE, mem, ON)
|
|
self.assertNotIn("notify", kinds(acts))
|
|
self.assertNotIn("restart", kinds(acts))
|
|
self.assertEqual(cw.load_memory(os.path.join(tmp, "missing.json"))["restarted"], {})
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
class Follow(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp(prefix="camwatch-follow-")
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
|
|
def test_growing_log(self):
|
|
path = os.path.join(self.tmp, "x.log")
|
|
f = cw.Follower(path)
|
|
text = "\n".join(START + GOOD_OPEN) + "\n"
|
|
with open(path, "w") as out:
|
|
out.write(text)
|
|
self.assertEqual(cw.current_failure(f.poll()), "")
|
|
self.assertEqual(f.state.verdict()[0], "ok")
|
|
# half a line, then the rest: read as one line
|
|
more = "\n".join(CLOSE + RESUME + FAIL) + "\n"
|
|
cut = more.index("Failed to load VCINT") + 5
|
|
with open(path, "a") as out:
|
|
out.write(more[:cut])
|
|
self.assertEqual(cw.current_failure(f.poll()), "")
|
|
with open(path, "a") as out:
|
|
out.write(more[cut:])
|
|
self.assertTrue(cw.current_failure(f.poll()).endswith("@11:30:01"))
|
|
self.assertEqual(cw.current_failure(f.state, steamvr_running=False), "")
|
|
with open(path, "a") as out:
|
|
out.write("\n".join(CLOSE + RESUME_OK) + "\n")
|
|
self.assertEqual(cw.current_failure(f.poll()), "")
|
|
|
|
def test_new_log_starts_afresh(self):
|
|
a, b = os.path.join(self.tmp, "a.log"), os.path.join(self.tmp, "b.log")
|
|
with open(a, "w") as out:
|
|
out.write("\n".join(START + GOOD_OPEN + CLOSE + RESUME + FAIL) + "\n")
|
|
f = cw.Follower(a)
|
|
self.assertTrue(cw.current_failure(f.poll()))
|
|
with open(b, "w") as out:
|
|
out.write("\n".join(START + GOOD_OPEN) + "\n")
|
|
f.fixed = b # as when xrservice.txt points at a new XRService's log
|
|
self.assertEqual(cw.current_failure(f.poll()), "")
|
|
self.assertEqual(f.path, b)
|
|
|
|
|
|
class Settings(unittest.TestCase):
|
|
def test_read_conf(self):
|
|
tmp = tempfile.mkdtemp(prefix="camwatch-conf-")
|
|
try:
|
|
path = os.path.join(tmp, "frametop.conf")
|
|
with open(path, "w") as f:
|
|
f.write("# Frametop\nREMOTE=1 # comment\nCAMWATCH_AUTO_RESTART=1\nCAMWATCH_IDLE_S='90'\n"
|
|
"CAMWATCH_IGNORE_APPIDS=\"1, 2\"\nnot a setting\n")
|
|
conf = cw.read_conf(path)
|
|
self.assertEqual(conf["CAMWATCH_AUTO_RESTART"], "1")
|
|
self.assertEqual(cw.conf_int(conf, "CAMWATCH_IDLE_S"), 90)
|
|
self.assertEqual(conf["CAMWATCH_IGNORE_APPIDS"], "1, 2")
|
|
self.assertEqual(cw.conf_int(conf, "CAMWATCH_COOLDOWN_MIN"), 30) # the default
|
|
self.assertEqual(cw.read_conf(os.path.join(tmp, "none"))["CAMWATCH_AUTO_RESTART"], "0")
|
|
self.assertEqual(cw.conf_int({"CAMWATCH_IDLE_S": "soon"}, "CAMWATCH_IDLE_S"), 60)
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_headset_worn(self):
|
|
tmp = tempfile.mkdtemp(prefix="camwatch-bl-")
|
|
orig = cw.process_running
|
|
try:
|
|
os.makedirs(os.path.join(tmp, "panel0"))
|
|
bright = os.path.join(tmp, "panel0", "brightness")
|
|
for value, compositor, worn in (("95", True, True), ("0", True, False), ("95", False, False)):
|
|
with open(bright, "w") as f:
|
|
f.write(value + "\n")
|
|
cw.process_running = lambda name, c=compositor: c and name == "vrcompositor"
|
|
self.assertEqual(cw.headset_worn(tmp), worn, (value, compositor))
|
|
self.assertFalse(cw.headset_worn(os.path.join(tmp, "none")))
|
|
finally:
|
|
cw.process_running = orig
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_remote_viewers(self):
|
|
tmp = tempfile.mkdtemp(prefix="camwatch-tcp-")
|
|
try:
|
|
path = os.path.join(tmp, "tcp")
|
|
with open(path, "w") as f:
|
|
f.write(" sl local_address rem_address st tx_queue rx_queue\n"
|
|
" 0: 0100007F:170C 00000000:0000 0A 00000000:00000000\n" # listening
|
|
" 1: 0B00640A:170C 0C00640A:D2F0 01 00000000:00000000\n" # a viewer
|
|
" 2: 0B00640A:0016 0C00640A:D2F1 01 00000000:00000000\n") # ssh
|
|
self.assertEqual(cw.remote_viewers(5900, (path,)), ["0C00640A:D2F0"])
|
|
self.assertEqual(cw.remote_viewers(5901, (path,)), [])
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|