Aktoren nur simuliert. Kein Motor, Licht oder Ton ist mit dieser Version verbunden. Die automatische Erkennung kann ausschließlich diesen simulierten Ablauf auslösen.
+Grabablauf nur simuliert. Hand, Atemmotor, Licht und Ton bleiben simuliert. Kameraservos können im separaten Farbtest wirklich bewegt werden. Die automatische Erkennung kann ausschließlich diesen simulierten Ablauf auslösen.
Die folgenden Knöpfe sind für manuelle Tests. Die Automatik gibt sich selbst frei.
Halte einen kräftig roten oder grünen Aufkleber vor die Kamera. Ein Punkt am Finger geht auch. Kein Training und kein Internet nötig. Zuerst die automatische Auslösung ausschalten und speichern.
+Eigene Farbe: Beispielfoto aufnehmen und dort auf die Mitte des Farbpunkts tippen. Das Foto wird nicht gespeichert.
+ +Farbtest lädt …
+Einheit µs = Pulsbreite, keine Winkel. Der Start setzt beide Servos auf 1500 µs; dabei kann die Halterung springen. Beim ersten Test Servohörner/Gestänge lösen, Mittelstellung prüfen und danach spannungsfrei wieder montieren. Grenzen nur schrittweise erweitern.
+Langsam annähern, vor dem Anschlag loslassen und die Grenze speichern. Auch links/rechts nicht gegen den Anschlag drücken. Keine automatische Endpunkterkennung.
+ +Noch keine Grenze in dieser Sitzung gespeichert.
+ +Sollposition: PAN 1500 / TILT 1500 µs
+Taste gedrückt halten: langsame Bewegung. Loslassen: Zielposition bleibt stehen. Bei fehlendem Tastensignal endet die Bewegung spätestens nach 0,35 s, sofern Pi und Treiber reagieren.
+Speichert den angeforderten Pulswert mit 10 µs Abstand nach innen, keine gemessene Position. Beim Einrichten gelten 1100–1900 µs, auch außerhalb der bisherigen Farbtest-Grenzen. Falls das nicht reicht: stoppen, nichts erzwingen. Zum Beenden „Kameraservos stoppen“ drücken. Neue Sitzung startet wieder bei 1500 µs.
+Nur Kameraservos CH0/CH1. Kein automatischer Start nach Neustart. Im Farbtest: Bei Punktverlust keine Suchbewegung; nach einer Sekunde wird die Servositzung beendet. Bei verlorener Bedienverbindung oder Kamerafehler werden die PWM-Signale abgeschaltet, soweit die Verbindung zum Treiber funktioniert. Die Stromversorgung bleibt eingeschaltet. Bei einem Hardwarefehler Servoversorgung abziehen.
+Erfolgreiche manuelle und automatische Auslösungen bleiben nach einem Neustart erhalten. Eine Profilzuordnung kann hier zum Testen oder Korrigieren gesetzt werden.
@@ -403,6 +460,170 @@ detectionConfig = c; $('detection-enabled').checked = c.enabled; $('detection-method').value = c.method; updateDetectionHelp(); drawDetection(); }).catch(error => { $('detection-status').textContent = error.message; }); + // Camera marker test has its own frame/overlay, never the trigger geometry. + let followConfig = null, followSession = null, followSample = null, followBusy = false; + let followEpoch = 0; + const followCanvas = $('follow-canvas'), followCtx = followCanvas.getContext('2d'); + function fillFollow(c) { + followConfig = c; + for (const axis of ['pan','tilt']) { + $( `follow-${axis}-min`).value = c[axis+'_min']; + $( `follow-${axis}-max`).value = c[axis+'_max']; + $( `follow-${axis}-reverse`).checked = c[axis+'_reverse']; + } + $('follow-tolerance').value = c.tolerance; + } + async function showFollowImage(b64, target, epoch) { + const image = new Image(); image.src = 'data:image/jpeg;base64,'+b64; + await image.decode(); + if (epoch !== followEpoch) return; + followCtx.drawImage(image,0,0,640,480); + followCtx.strokeStyle = '#fff'; followCtx.lineWidth = 2; + followCtx.beginPath(); followCtx.moveTo(305,240); followCtx.lineTo(335,240); + followCtx.moveTo(320,225); followCtx.lineTo(320,255); followCtx.stroke(); + if (target) { + followCtx.strokeStyle = '#ffca78'; followCtx.lineWidth = 5; + followCtx.beginPath(); followCtx.arc(target.x*640,target.y*480,Math.max(15,target.radius*640),0,Math.PI*2); followCtx.stroke(); + } + } + async function followAction(fn) { + if (followBusy) return; + followBusy = true; + try { await fn(); } + catch (e) { $('follow-status').textContent = e.message; } + finally { followBusy = false; } + } + async function saveFollow() { + const c = {...followConfig}; + for (const axis of ['pan','tilt']) { + c[axis+'_min'] = Number($(`follow-${axis}-min`).value); + c[axis+'_max'] = Number($(`follow-${axis}-max`).value); + c[axis+'_reverse'] = $(`follow-${axis}-reverse`).checked; + } + c.tolerance = Number($('follow-tolerance').value); + fillFollow(await api('/api/follow/config',c)); + } + for (const [id,hue] of [['follow-green',85],['follow-red',0]]) { + $(id).onclick = () => followAction(async () => { + if (!followConfig) throw new Error('Einstellungen noch nicht geladen'); + fillFollow(await api('/api/follow/config',{...followConfig,hue})); + $('follow-status').textContent = 'Farbe gespeichert. Vorschau starten.'; + }); + } + $('follow-save').onclick = () => followAction(async () => { + await saveFollow(); $('follow-status').textContent = 'Grenzen und Richtung gespeichert'; + }); + $('follow-capture').onclick = () => followAction(async () => { + const photo = await api('/api/follow/capture'); + followSample = photo.token; followEpoch++; + await showFollowImage(photo.image_base64,null,followEpoch); + $('follow-status').textContent = 'Standbild: Jetzt auf die Mitte des Farbpunkts tippen.'; + }); + followCanvas.onclick = e => followAction(async () => { + if (!followSample) return; + const rect = followCanvas.getBoundingClientRect(); + const config = await api('/api/follow/select',{token:followSample, + x:(e.clientX-rect.left)/rect.width,y:(e.clientY-rect.top)/rect.height}); + fillFollow(config); $('follow-status').textContent = 'Eigene Farbe übernommen. Vorschau starten.'; + }); + async function beginFollow(hardware, operation = 'follow') { + if (hardware && !$('follow-ack').checked) throw new Error('Bitte zuerst Mittelstellung und freien Bewegungsraum prüfen und bestätigen.'); + await saveFollow(); + const result = await api('/api/follow/start',{hardware,acknowledged:$('follow-ack').checked,operation}); + releaseJog(); jogSequence = 0; followSession = result.session; followSample = null; followEpoch++; + $('follow-status').textContent = hardware ? 'Echte Servos aktiv' : 'Farbvorschau ohne Servobewegung'; + } + $('follow-preview').onclick = () => followAction(() => beginFollow(false)); + $('follow-hardware').onclick = () => followAction(() => beginFollow(true)); + $('follow-stop').onclick = async () => { + releaseJog(); setupReady = false; followSession = null; followEpoch++; + try { await api('/api/follow/stop'); $('follow-status').textContent = 'Kameraservos gestoppt'; } + catch(e) { $('follow-status').textContent = 'Stopp nicht bestätigt – Servoversorgung abziehen. '+e.message; } + }; + document.addEventListener('visibilitychange', () => { + if (document.hidden && followSession) $('follow-stop').click(); + }); + // Hold-to-jog: separate short motion lease; sequence numbers reject late requests. + let jogTimer = null, jogSequence = 0, setupReady = false, followDemo = false; + async function sendJog(direction) { + if (!followSession) return; + return api('/api/follow/jog',{session:followSession,direction,sequence:++jogSequence}); + } + function releaseJog() { + if (jogTimer === null) return; + clearInterval(jogTimer); jogTimer = null; + sendJog(null).catch(e => { $('follow-status').textContent = 'Tastenstopp nicht bestätigt: '+e.message; }); + } + function pressJog(direction) { + if (!setupReady || !followSession || jogTimer !== null) return; + const send = () => sendJog(direction).catch(e => { + releaseJog(); $('follow-status').textContent = e.message; + }); + jogTimer = setInterval(send,100); send(); + } + for (const direction of ['left','right','up','down']) { + const button = $('setup-'+direction); + button.disabled = true; + button.onpointerdown = e => { + if (e.button !== 0) return; + e.preventDefault(); button.setPointerCapture(e.pointerId); pressJog(direction); + }; + button.onpointerup = button.onpointercancel = button.onlostpointercapture = releaseJog; + button.onkeydown = e => { + if ([' ','Enter'].includes(e.key)) { e.preventDefault(); if (!e.repeat) pressJog(direction); } + }; + button.onkeyup = e => { if ([' ','Enter'].includes(e.key)) { e.preventDefault(); releaseJog(); } }; + button.onblur = releaseJog; + $('setup-save-'+direction).disabled = true; + $('setup-save-'+direction).onclick = () => followAction(async () => { + releaseJog(); + await sendJog(null); + fillFollow(await api('/api/follow/remember',{session:followSession,direction})); + $('setup-note').textContent = 'Grenze gespeichert – mit 10 µs Abstand nach innen.'; + }); + } + $('setup-release').onclick = releaseJog; + document.addEventListener('pointerup',releaseJog); + document.addEventListener('pointercancel',releaseJog); + window.addEventListener('blur',releaseJog); + $('setup-start').onclick = () => followAction(() => beginFollow(!followDemo,'setup')); + + async function pollFollow() { + try { + if (followSession) await api('/api/follow/heartbeat',{session:followSession}); + const r = await fetch('/api/follow',{cache:'no-store'}); + if (!r.ok) throw new Error('Farbtest nicht erreichbar'); + const f = await r.json(); + followDemo = f.demo; + $('setup-start').textContent = followDemo ? 'Einrichtmodus als Demo starten' : 'Einrichtmodus starten'; + setupReady = f.active && f.operation === 'setup' && !!followSession; + if (!setupReady) releaseJog(); + for (const direction of ['left','right','up','down']) { + $('setup-'+direction).disabled = !setupReady; + $('setup-save-'+direction).disabled = !setupReady; + } + $('setup-limits').textContent = 'Gespeicherte Grenzen: PAN '+f.config.pan_min+'–'+f.config.pan_max+' / TILT '+f.config.tilt_min+'–'+f.config.tilt_max+' µs'; + $('setup-position').textContent = 'Sollposition: PAN '+f.pan_us+' / TILT '+f.tilt_us+' µs'; + if (!followConfig) { + fillFollow(f.config); + $('follow-status').textContent = 'Bereit. Farbe wählen oder Beispielfoto aufnehmen.'; + } + if (!f.active) followSession = null; + if (f.active || f.error) $('follow-status').textContent = f.error || + (f.operation === 'setup' ? (f.hardware ? 'Einrichtmodus · echte Servos' : 'Einrichtmodus · nur Demo') + : (f.hardware ? 'Echte Servos aktiv' : 'Vorschau ohne Bewegung'))+' · '+ + (f.operation === 'setup' ? 'Richtungstaste gedrückt halten' : (f.target ? 'Punkt erkannt' : 'Kein eindeutiger Farbpunkt'))+' · PAN '+f.pan_us+' / TILT '+f.tilt_us+' µs'; + if (f.active && f.image_base64 && !followSample) await showFollowImage(f.image_base64,f.target,followEpoch); + } catch(e) { + if (followSession) { + releaseJog(); setupReady = false; followSession = null; + $('follow-status').textContent = 'Verbindung verloren; Nachführung wird beendet. '+e.message; + } + } + setTimeout(pollFollow,300); + } + pollFollow(); + refreshDetection(); setInterval(drawDetection, 250); refresh(); diff --git a/halloween/instructions.md b/halloween/instructions.md index fe5bcb3..2d12467 100644 --- a/halloween/instructions.md +++ b/halloween/instructions.md @@ -31,7 +31,7 @@ Hier liegt die erste Pi-WebUI. `run.py` im Projektstamm startet sie. Internetfreigabe ist nicht beauftragt. Zugangsschutz bleibt offen. - Laufzeitdaten liegen standardmäßig unter `data/` und werden weder hochgeladen noch in Git aufgenommen. -- Echte GPIO-/Aktorsteuerung erst nach geklärter Mechanik, Versorgung, +- Echte GPIO-/Grabaktorsteuerung erst nach geklärter Mechanik, Versorgung, sicherem Stopp und den Hardwaretests aus `PLAN.md` ergänzen. Bei Änderungen an Zustandsübergängen oder Endpunkten die zugehörigen Tests @@ -93,3 +93,38 @@ unter `tests/` und die Bedienhinweise in `SOFTWARE.md` prüfen. bestätigtes Bildlöschen; Ereignis/Profilzuordnung bleiben bestehen. Löschen betrifft den aktiven Bestand, keine Upload-Sicherungen. Keine automatische Alterslöschung von Bildern und keine Änderung der Auto-Profil-Frist. + + +## Kamera folgt einem Farbpunkt + +- `follow.py` bietet lokale Pillow-HSV-Erkennung ohne OpenCV/Training und einen + optional bewusst gestarteten PCA9685-Kameratest. Vorschau und Demo greifen + niemals auf I²C zu; Demo lehnt Hardwarestart ab. Echte Ausgabe nur CH0/CH1, + Bus 1 / 0x40, nominal 50 Hz. Platine für diesen Test ausschließlich dafür nutzen. +- Farbwert, Toleranz, Achsrichtungen und Pulsgrenzen atomar in data/follow.json; + Beispielfoto und Analysebilder nur flüchtig. Bildauswahl nutzt Token und exakt + das fotografierte Bild; keine neuen Ereignisse, Profile oder Trainingsbilder. +- Automatische Auslösung und Follow-Sitzung gegenseitig ausschließen. Follow + hält Controller an; allgemeiner Stopp und Reset müssen zuerst Follow stoppen. + Danach bleibt bewusste Rücksetzung des Grabablaufs nötig. +- Start braucht frische Kamera und explizite Hardware-/Mechanikbestätigung. + Beide Achsen starten bei 1500 µs (möglicher Sprung, vorab entkoppelt prüfen). + Defaults 1400–1600, absolute Eingabegrenzen 1100–1900 inklusive 1500, + maximal 80 µs/s, keine Suchfahrt; Richtungswechsel nur gestoppt. +- Sitzung startet nie automatisch. Startender Browser erneuert geheimen + Sitzungsschlüssel; 1,5 s ohne Heartbeat, 1 s ohne Bild/Auswertung oder 1 s + ohne eindeutigen Marker im Hardwarebetrieb schaltet PWM aus. Stale Resultate + und alte Heartbeats dürfen keine neu gestartete Sitzung steuern. +- PWM-Abschaltfehler sichtbar melden; keine sichere Hardwareabschaltung aus + Softwaretests ableiten. I²C-Ausfall/harter Prozessabbruch können PWM stehen + lassen. Keine unabhängige OE-Sperre; nur beaufsichtigter Test. + +- `operation=setup` ist ein eigener manueller Einrichtmodus derselben exklusiven + Kamerasitzung. Kein Farbzielsuchen/-verluststopp; Kamera und Heartbeat bleiben + erforderlich. Richtungstaste halten: maximal 40 µs/s, Lease 0,35 s und steigende + Sequenznummer gegen verspätete Befehle. Loslassen hält die Sollposition per + PWM; globaler/Kamerastopp beendet PWM. +- Einrichtung darf bisherige Follow-Grenzen überschreiten, niemals 1100–1900 µs. + Grenzspeicherung braucht passende Sitzung und beendete Bewegung; Richtungs- + umkehr beachten, 10 µs nach innen speichern, atomar in follow.json. Keine + Anschlagerkennung behaupten; nur Sollwerte, beide Achsen vor Anschlag stoppen. diff --git a/halloween/web.py b/halloween/web.py index 48b64e4..e0f7c35 100644 --- a/halloween/web.py +++ b/halloween/web.py @@ -1,4 +1,4 @@ -"""Lokale WebUI und MJPEG-Livebild; alle Aktoren bleiben simuliert.""" +"""Lokale WebUI und MJPEG-Livebild; Grabaktoren simuliert, Kameraservos separat freigebbar.""" from contextlib import asynccontextmanager, suppress from pathlib import Path @@ -15,6 +15,7 @@ from .camera import CameraStream from .controller import Controller, MODES from .event_store import EventStore from .detection import DetectionService +from .follow import FollowService PAGE = (Path(__file__).parent / "index.html").read_text(encoding="utf-8") @@ -46,6 +47,9 @@ def create_app(demo: bool = False, state_dir: Path | None = None) -> FastAPI: events = EventStore(directory) detection = DetectionService(camera, controller, events, directory) + follow = FollowService(camera, directory) + follow_api_lock = asyncio.Lock() + @asynccontextmanager async def lifespan(app: FastAPI): async def periodic_profile_cleanup(): @@ -59,16 +63,19 @@ def create_app(demo: bool = False, state_dir: Path | None = None) -> FastAPI: async def detection_watchdog(): while True: detection.watchdog() + follow.watchdog() await asyncio.sleep(0.1) camera.start() detection.start() + follow.start() watchdog_task = asyncio.create_task(detection_watchdog()) cleanup_task = asyncio.create_task(periodic_profile_cleanup()) try: yield finally: controller.stop() + follow.stop() watchdog_task.cancel() with suppress(asyncio.CancelledError): await watchdog_task @@ -83,6 +90,7 @@ def create_app(demo: bool = False, state_dir: Path | None = None) -> FastAPI: app.state.controller = controller app.state.events = events app.state.detection = detection + app.state.follow = follow @app.middleware("http") async def security_headers(request: Request, call_next): @@ -112,6 +120,8 @@ def create_app(demo: bool = False, state_dir: Path | None = None) -> FastAPI: @app.post("/api/detection/config", dependencies=[Depends(require_ui)]) async def configure_detection(payload: dict): + if follow.active and payload.get("enabled"): + raise HTTPException(status_code=409, detail="Zuerst Kameranachführung stoppen") try: return detection.configure(payload) except (ValueError, TypeError, KeyError, IndexError) as exc: @@ -119,6 +129,46 @@ def create_app(demo: bool = False, state_dir: Path | None = None) -> FastAPI: except OSError as exc: raise HTTPException(status_code=500, detail=f"Einstellung nicht gespeichert: {exc}") from exc + @app.get("/api/follow") + async def follow_status(): + return follow.status() + + @app.post("/api/follow/{action}", dependencies=[Depends(require_ui)]) + async def follow_action(action: str, payload: dict): + async with follow_api_lock: + try: + if action == "config": + return follow.configure(payload) + if action == "capture": + return follow.capture() + if action == "select": + return follow.select(payload.get("token"), payload.get("x"), payload.get("y")) + if action == "start": + if detection.get_config()["enabled"]: + raise ValueError("Zuerst automatische Auslösung ausschalten und speichern") + if type(payload.get("hardware", False)) is not bool: + raise ValueError("Ungültige Servofreigabe") + controller.stop() + session = follow.begin(payload.get("hardware", False), + payload.get("acknowledged") is True, + payload.get("operation", "follow")) + return {"session": session} + if action == "jog": + return follow.jog(payload.get("session"), payload.get("direction"), payload.get("sequence")) + if action == "remember": + return follow.remember(payload.get("session"), payload.get("direction")) + if action == "heartbeat": + follow.heartbeat(payload.get("session")) + return {"ok": True} + if action == "stop": + follow.halt() + return follow.status() + raise HTTPException(status_code=404, detail="Unbekannte Aktion") + except (ValueError, TypeError) as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + except OSError as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + @app.get("/api/events") async def recent_events(limit: int = Query(20, ge=1, le=100)): try: @@ -213,11 +263,13 @@ def create_app(demo: bool = False, state_dir: Path | None = None) -> FastAPI: @app.post("/api/stop", dependencies=[Depends(require_ui)]) async def stop(): + follow.halt() controller.stop() return controller.status() @app.post("/api/reset", dependencies=[Depends(require_ui)]) async def reset(): + follow.halt() controller.reset() detection.reset() return controller.status() diff --git a/requirements.txt b/requirements.txt index 5d9ffe5..a3b7558 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ fastapi>=0.115,<1 uvicorn>=0.30,<1 Pillow>=10,<13 +smbus2>=0.4.3,<1 diff --git a/tests/instructions.md b/tests/instructions.md index 29a5607..36d5843 100644 --- a/tests/instructions.md +++ b/tests/instructions.md @@ -64,3 +64,13 @@ Automatische Tests prüfen die aktuelle Software ohne Pi-Hardware. privaten Socket für alle Schritte, BatchMode auf Slaves, TTY für sudo sowie Aufräumen bei Erfolg/Verbindungs-/Transfer-/Aktivierungsfehlern und pure Vorschau. Dies bestätigt keine reale SSH-/Passwortanmeldung am Pi. + + +- `test_follow.py` prüft synthetische Farbpunkte (auch Rot um die HSV-Grenze), + Mehrdeutigkeit, Bildauswahl, Persistenz ohne Fotos, Pulsgrenzen/Geschwindigkeit, + Richtungswechsel, alte Resultate/Sitzungen, Fehler-/Abbruchpfade, API-Interlock + und I²C-Protokoll mit künstlichem Bus. Kein echter Pi oder Servo wird angesteuert. + +- Follow-Einrichttests prüfen Hold/Release, ablaufende Bewegungslease, verspätete + Tastenbefehle, falsche/alte Sitzung, Grenzen jenseits gespeicherter Followwerte, + absolute Pulsgrenzen, Richtungsumkehr und persistente Grenze mit Innenabstand. diff --git a/tests/test_follow.py b/tests/test_follow.py new file mode 100644 index 0000000..a219f8d --- /dev/null +++ b/tests/test_follow.py @@ -0,0 +1,235 @@ +import asyncio +from io import BytesIO +from pathlib import Path +import tempfile +import unittest + +import httpx +from PIL import Image, ImageDraw + +from halloween.follow import DEFAULT, FollowService, PCA9685, locate, validate +from halloween.web import create_app + + +def frame(color='lime', double=False): + image = Image.new('RGB', (640,480), '#181818') + draw = ImageDraw.Draw(image) + draw.ellipse((440,180,510,250), fill=color) + if double: draw.ellipse((100,180,170,250), fill=color) + data = BytesIO(); image.save(data, 'JPEG'); return data.getvalue() + + +class Camera: + demo = False + error = None + def __init__(self, clock): self.clock, self.jpeg, self.captured = clock, frame(), clock() + def latest_timed(self): return 1, self.jpeg, self.captured + + +class Driver: + def __init__(self): self.moves, self.closed = [], False + def move(self, pan, tilt): self.moves.append((pan,tilt)) + def close(self): self.closed = True + + +class FollowTest(unittest.TestCase): + def setUp(self): + self.folder = tempfile.TemporaryDirectory(); self.addCleanup(self.folder.cleanup) + self.now = 10. + self.camera = Camera(lambda:self.now) + self.drivers = [] + def factory(): + driver = Driver(); self.drivers.append(driver); return driver + self.follow = FollowService(self.camera, Path(self.folder.name), factory, lambda:self.now) + + def start(self): return self.follow.begin(True, True) + def step(self, target): + self.now += .2; self.camera.captured = self.now + self.follow.heartbeat(self.follow.session) + self.follow.process(2,self.camera.jpeg,self.now,self.follow.revision,target) + + def test_marker_and_ambiguity(self): + target = locate(frame(), DEFAULT) + self.assertGreater(target['x'], .7) + self.assertIsNone(locate(frame(double=True), DEFAULT)) + self.assertIsNone(locate(frame('gray'), DEFAULT)) + self.assertIsNone(locate(frame('red'), DEFAULT)) + self.assertIsNotNone(locate(frame('red'), dict(DEFAULT,hue=0))) + + def test_photo_selection_and_no_persistent_image(self): + shot = self.follow.capture() + config = self.follow.select(shot['token'],.74,.45) + self.assertLess(abs(config['hue']-85),3) + self.assertEqual([p.name for p in Path(self.folder.name).iterdir()], ['follow.json']) + with self.assertRaises(ValueError): self.follow.select('old',.74,.45) + with self.assertRaises(ValueError): self.follow.select(shot['token'],float('nan'),.5) + with self.assertRaises(ValueError): self.follow.select(shot['token'],.2,.2) + + def test_no_hardware_without_explicit_start_and_no_restart_resume(self): + self.follow.begin(False) + self.step({'x':.9,'y':.9}) + self.assertEqual(self.drivers, []) + self.follow.halt() + with self.assertRaises(ValueError): self.follow.begin(True,False) + self.assertFalse(FollowService(self.camera,Path(self.folder.name)).active) + self.assertEqual(self.drivers, []) + + def test_rate_limits_reverse_deadband_and_stop(self): + self.start(); driver=self.drivers[0] + self.step({'x':.9,'y':.5}) + self.assertAlmostEqual(self.follow.pan,1512.8) + self.assertEqual(self.follow.tilt,1500) + self.assertLessEqual(abs(driver.moves[-1][0]-driver.moves[-2][0]),16) + for _ in range(30): self.step({'x':1.,'y':1.}) + self.assertEqual(self.follow.pan,1600) + self.follow.halt(); self.assertTrue(driver.closed) + count=len(driver.moves) + self.follow.process(99,self.camera.jpeg,self.now,self.follow.revision,{'x':0,'y':0}) + self.assertEqual(len(driver.moves),count) + self.follow.configure(dict(DEFAULT,pan_reverse=True)) + self.start(); self.step({'x':.9,'y':.5}) + self.assertLess(self.follow.pan,1500) + + def test_stale_results_cannot_move_new_session(self): + self.start(); revision=self.follow.revision + self.follow.halt(); self.start(); driver=self.drivers[-1] + self.follow.process(2,self.camera.jpeg,self.now,revision,{'x':1,'y':1}) + self.assertEqual(driver.moves,[(1500,1500)]) + + def test_lost_marker_heartbeat_and_camera_latch_off(self): + for reason in ('marker','heartbeat','camera','inference'): + self.follow.halt(); self.camera.captured=self.now + self.start(); driver=self.drivers[-1] + if reason=='marker': + for _ in range(6): self.step(None) + else: + self.now+=2 + if reason!='camera': self.camera.captured=self.now + if reason=='inference': self.follow.heartbeat(self.follow.session) + self.follow.watchdog() + self.assertFalse(self.follow.active, reason) + self.assertTrue(driver.closed, reason) + self.assertIsNotNone(self.follow.error) + + def test_invalid_limits_and_corrupt_saved_config(self): + for config in (dict(DEFAULT,pan_min=1600), dict(DEFAULT,pan_max=2000), + dict(DEFAULT,hue=float('inf')), dict(DEFAULT,pan_reverse='false')): + with self.assertRaises(ValueError): validate(config) + Path(self.folder.name,'follow.json').write_text('bad') + follow=FollowService(self.camera,Path(self.folder.name)) + with self.assertRaises(ValueError): follow.begin() + + def test_close_error_is_visible_and_restart_is_explicit(self): + self.start() + def bad_close(): raise OSError('I2C ausgefallen') + self.drivers[0].close=bad_close + self.follow.halt() + self.assertFalse(self.follow.active) + self.assertIn('Servoversorgung ausschalten',self.follow.error) + + def test_failed_start_disables_channels_and_demo_rejects_hardware(self): + driver=Driver() + def bad_move(*args): raise OSError('I2C fehlgeschlagen') + driver.move=bad_move + self.follow.factory=lambda:driver + with self.assertRaises(ValueError): self.start() + self.assertTrue(driver.closed) + self.assertFalse(self.follow.active) + self.camera.demo=True + with self.assertRaises(ValueError): self.start() + + def test_old_heartbeat_cannot_renew_new_session(self): + old=self.start(); self.follow.halt(); current=self.start() + self.assertNotEqual(old,current) + with self.assertRaises(ValueError): self.follow.heartbeat(old) + self.now+=2; self.camera.captured=self.now + self.follow.watchdog() + self.assertFalse(self.follow.active) + + def test_setup_hold_release_expiry_and_out_of_order_requests(self): + session=self.follow.begin(True,True,'setup') + driver=self.drivers[-1] + self.follow.jog(session,'right',1) + self.step(None) + self.assertAlmostEqual(self.follow.pan,1508) + self.step(None) # no renewal: 0.35 s motion lease has expired + self.assertAlmostEqual(self.follow.pan,1508) + self.follow.jog(session,'right',2) + self.follow.jog(session,None,3) + self.assertFalse(self.follow.jog(session,'right',2)['accepted']) + self.step(None) + self.assertAlmostEqual(self.follow.pan,1508) + for _ in range(8): self.step(None) + self.assertTrue(self.follow.active) # setup needs no colour marker + self.follow.halt() + with self.assertRaises(ValueError): self.follow.jog(session,'up',4) + self.assertTrue(driver.closed) + + def test_setup_limits_reverse_and_remember_are_persistent(self): + self.follow.configure(dict(DEFAULT,pan_min=1500,pan_max=1500,tilt_reverse=True)) + session=self.follow.begin(True,True,'setup') + for seq in range(1,5): + self.follow.jog(session,'right',seq); self.step(None) + self.assertGreater(self.follow.pan,1500) # setup intentionally exceeds saved limits + with self.assertRaises(ValueError): self.follow.remember(session,'right') + self.follow.jog(session,None,5) + c=self.follow.remember(session,'right') + self.assertEqual(c['pan_max'],1522) + self.assertEqual(FollowService(self.camera,Path(self.folder.name)).config['pan_max'],1522) + self.follow.jog(session,'up',6); self.step(None) + self.assertGreater(self.follow.tilt,1500) + for seq in range(7,80): + self.follow.jog(session,'up',seq); self.step(None) + self.assertEqual(self.follow.tilt,1900) + self.follow.jog(session,None,80) + self.assertEqual(self.follow.remember(session,'up')['tilt_max'],1890) + + def test_setup_cannot_be_controlled_by_wrong_session_or_follow_mode(self): + session=self.start() + with self.assertRaises(ValueError): self.follow.jog(session,'up',1) + with self.assertRaises(ValueError): self.follow.remember(session,'up') + self.follow.halt(); session=self.follow.begin(True,True,'setup') + for sequence in (True,-1,1.5): + with self.assertRaises(ValueError): self.follow.jog(session,'up',sequence) + with self.assertRaises(ValueError): self.follow.jog('wrong','up',1) + with self.assertRaises(ValueError): self.follow.remember('wrong','up') + self.now+=2; self.camera.captured=self.now + with self.assertRaises(ValueError): self.follow.jog(session,'up',1) + self.assertFalse(self.follow.active) + + def test_register_protocol(self): + class Bus: + def __init__(self,n): self.writes=[]; self.closed=False + def write_byte_data(self,a,r,v): self.writes.append((a,r,v)) + def write_i2c_block_data(self,a,r,v): self.writes.append((a,r,v)) + def close(self): self.closed=True + driver=PCA9685(Bus) + self.assertEqual(driver.bus.writes[:2],[(0x40,0x09,0x10),(0x40,0x0d,0x10)]) + self.assertIn((0x40,0xfe,121),driver.bus.writes) + driver.move(1500,1500) + self.assertEqual(driver.bus.writes[-2:],[(0x40,6,[0,0,51,1]),(0x40,10,[0,0,51,1])]) + driver.close(); self.assertTrue(driver.bus.closed) + + def test_api_guard_interlock_global_stop_and_reset(self): + async def scenario(): + app=create_app(demo=True,state_dir=Path(self.folder.name)) + app.state.follow.camera=self.camera + app.state.follow.clock=lambda:self.now + app.state.follow.factory=lambda:Driver() + app.state.camera.demo=True + h={'X-Halloween-UI':'1'} + async with httpx.AsyncClient(transport=httpx.ASGITransport(app),base_url='http://local') as c: + self.assertEqual((await c.post('/api/follow/start',json={})).status_code,403) + self.assertEqual((await c.post('/api/follow/start',headers=h,json={'hardware':False})).status_code,200) + cfg=app.state.detection.get_config(); cfg['enabled']=True + self.assertEqual((await c.post('/api/detection/config',headers=h,json=cfg)).status_code,409) + await c.post('/api/stop',headers=h) + self.assertFalse(app.state.follow.active) + await c.post('/api/follow/start',headers=h,json={'hardware':False}) + await c.post('/api/reset',headers=h) + self.assertFalse(app.state.follow.active) + await c.post('/api/detection/config',headers=h,json=cfg) + self.assertEqual((await c.post('/api/follow/start',headers=h,json={})).status_code,422) + asyncio.run(scenario()) + +if __name__=='__main__': unittest.main() diff --git a/tests/test_web.py b/tests/test_web.py index 3325e11..c84da40 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -20,7 +20,7 @@ class WebTest(unittest.TestCase): page = await client.get("/") self.assertEqual(page.status_code, 200) self.assertIn("/stream.mjpg", page.text) - self.assertIn("Aktoren nur simuliert", page.text) + self.assertIn("Grabablauf nur simuliert", page.text) for _ in range(30): if (await client.get("/api/status")).json()["camera"]["ready"]: break