ATEM: audio, camera settings and MIDI control watched and fixed against the restore file

atem_watch.js (run with Companion's node + its atem-connection) keeps one
connection to the switcher and compares, every second, every Fairlight
source's mix option and fader and every camera's settings with
atem-reference.xml (the 2026-10-07 14:59:55 restore file). ATEM shows
ATEM AUDIO WRONG / ATEM CAMERA WRONG; the hold and Sunday auto-fix send
back only what differs. ATEM SW MIDI ON is fixed by quitting ATEM Software
Control, switching midiEnabled off and reopening it. Live check: switcher
matches the file (0 differences, 1440 camera values).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-11 10:54:59 -04:00
co-authored by Claude Opus 5.5
parent b468b42889
commit 98070f31c4
7 changed files with 27211 additions and 5 deletions
+47 -1
View File
@@ -283,6 +283,51 @@ class TestChecks(unittest.TestCase):
self.snap.prefs[key] = "1"
self.assertEqual(run_steps(engine, self.snap, self.now + 5, 4)[0]["atem"]["x"], "ATEM\nATEM SW MIDI ON")
def test_atem_reference_from_the_restore_file(self):
d = tempfile.mkdtemp()
xml = os.path.join(d, "r.xml")
open(xml, "w").write("""<Profile product="ATEM 4 M/E Constellation 4K Plus"><FairlightAudioMixer>
<AudioInputs><AudioInput id="16"><AudioSources><AudioSource id="-65280" faderGain="0" mixOption="On"/></AudioSources></AudioInput>
<AudioInput id="1301"><AudioSources><AudioSource id="-65280" faderGain="-inf" mixOption="Off"/></AudioSources></AudioInput>
<AudioInput id="7"><AudioSources><AudioSource id="-65280" faderGain="-2.5" mixOption="AudioFollowVideo"/></AudioSources></AudioInput>
</AudioInputs></FairlightAudioMixer><CameraControl>
<Parameter device="2" category="Video" parameter="ManualWhiteBalance" temperature="4500" tint="0"/>
<Parameter device="6" category="ColorCorrection" parameter="LiftAdjust" red="0" green="0" blue="0" luma="0.18994140625"/>
<Parameter device="1" category="Lens" parameter="ApertureNormalised" value="0"/>
<Parameter device="1" property="ApertureCoarse" value="1"/></CameraControl></Profile>""")
ref = sp.atem_reference(xml)
self.assertEqual(ref["audio"], {"16/-65280": {"mixOption": 2, "faderGain": 0},
"1301/-65280": {"mixOption": 1}, # -inf: fader not compared
"7/-65280": {"mixOption": 4, "faderGain": -250}})
self.assertEqual(ref["camera"], {"2/1/2": [4500.0, 0.0], "6/8/0": [0.0, 0.0, 0.0, 0.18994140625]})
def test_atem_audio_and_camera_checks(self):
engine = sp.Engine(CONFIG)
self.snap.atem = {"connected": True, "audio": [], "camera": [], "time": self.now}
self.assertEqual(run_steps(engine, self.snap, self.now, 4)[0]["atem"]["x"], "ATEM\nOK")
self.snap.atem = {"connected": True, "audio": ["input 16: Off, should be On"], "camera": ["camera 2 ..."],
"time": self.now + 5}
tiles, _ = run_steps(engine, self.snap, self.now + 5, 4)
self.assertEqual(tiles["atem"]["x"], "ATEM\nATEM AUDIO WRONG\n+1")
self.assertEqual(sorted(w for _, k, w, _ in engine.fix_plan("atem", self.snap)), ["atem_audio", "atem_camera"])
self.snap.atem = None # watcher not reporting
self.assertEqual(run_steps(engine, self.snap, self.now + 10, 2)[0]["atem"]["x"], "ATEM\nOK")
for f in ("atem_audio", "atem_camera", "atem_midi"):
self.assertIn(f, CONFIG["auto_fix"]["fixes"])
self.assertIn("midiEnabled -bool false", CONFIG["fixes"]["atem_midi"]["run"])
def test_atem_fix_goes_through_the_watcher(self):
class W:
def __init__(self): self.parts = []
def fix(self, part): self.parts.append(part); return True
w, old = W(), sp.ATEM_WATCHER
sp.ATEM_WATCHER = w
try:
self.assertTrue(sp.execute_fix(CONFIG, "atem_camera", {}))
finally:
sp.ATEM_WATCHER = old
self.assertEqual(w.parts, ["camera"])
def test_propresenter_yellow_warning_is_ignored(self):
engine = sp.Engine(CONFIG)
self.snap.conns["Propresenter-API"].update(category="warning", message="Socket closed")
@@ -1284,7 +1329,7 @@ class TestFixes(unittest.TestCase):
def test_every_fix_is_valid(self):
for f in CONFIG["fixes"].values():
self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck", "midi_connect", "reconnect"))
self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck", "midi_connect", "reconnect", "atem"))
class FakeDeck:
@@ -1511,6 +1556,7 @@ class TestAgainstFakeCompanion(unittest.TestCase):
def test_end_to_end_loop(self):
"""The real run() loop: it writes state and heartbeat, and consumes button commands."""
cfg = json.loads(json.dumps(CONFIG))
cfg.pop("atem", None) # never connect to the real switcher from a test
cfg["companion"]["port"] = self.port
# keep it offline: no pings, no web, no launchd, no bonjour lookups
cfg["network"]["gateway"] = None