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()