Private
Public Access
547 lines
28 KiB
Python
547 lines
28 KiB
Python
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_similar_coloured_paper_does_not_hide_saturated_marker(self):
|
|
image = Image.new('RGB', (640,480), '#181818')
|
|
draw = ImageDraw.Draw(image)
|
|
draw.rectangle((200,0,560,330), fill=(125,158,204)) # bluish paper, hue ~152, S ~95
|
|
draw.rectangle((250,250,520,330), fill=(120,153,198)) # slightly stronger paper shade
|
|
draw.ellipse((300,30,350,80), fill=(68,73,177))
|
|
data = BytesIO(); image.save(data, 'JPEG'); data = data.getvalue()
|
|
for tolerance in (10, 18, 35):
|
|
target = locate(data, dict(DEFAULT, hue=165, tolerance=tolerance))
|
|
self.assertIsNotNone(target)
|
|
self.assertLess(target['y'], .2)
|
|
|
|
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,1524)
|
|
self.assertEqual(self.follow.tilt,1500)
|
|
self.assertLessEqual(abs(driver.moves[-1][0]-driver.moves[-2][0]),40)
|
|
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_legacy_speed_migration_preserves_endpoints(self):
|
|
import json
|
|
legacy = dict(DEFAULT, pan_min=1234, pan_max=1789)
|
|
del legacy['follow_speed']
|
|
Path(self.folder.name, 'follow.json').write_text(json.dumps(legacy))
|
|
restored = FollowService(self.camera, Path(self.folder.name))
|
|
self.assertEqual(restored.config, dict(legacy, follow_speed=80))
|
|
ancient = {k:v for k,v in legacy.items() if '_setup_' not in k}
|
|
self.assertEqual(validate(ancient), restored.config)
|
|
for speed in (39, 801, float('nan'), True):
|
|
with self.assertRaises(ValueError): validate(dict(DEFAULT, follow_speed=speed))
|
|
|
|
def test_speed_is_time_based_and_delayed_frame_is_capped(self):
|
|
config = dict(DEFAULT, pan_min=500, pan_max=2500,
|
|
pan_setup_min=500, pan_setup_max=2500)
|
|
for speed in (80, 400, 800):
|
|
for fps in (15, 20):
|
|
self.follow.halt()
|
|
self.follow.configure(dict(config, follow_speed=speed))
|
|
self.start()
|
|
for number in range(fps):
|
|
self.now += 1/fps; self.camera.captured = self.now
|
|
self.follow.heartbeat(self.follow.session)
|
|
self.follow.process(number, self.camera.jpeg, self.now,
|
|
self.follow.revision, {'x':1., 'y':.5}, 3.2)
|
|
self.assertAlmostEqual(self.follow.pan, 1500+min(speed, 300))
|
|
self.assertEqual(self.follow.status()['fps'], fps)
|
|
self.assertEqual(self.follow.status()['analysis_ms'], 3.2)
|
|
before = self.follow.pan
|
|
self.now += .5; self.camera.captured = self.now
|
|
self.follow.process(99, self.camera.jpeg, self.now,
|
|
self.follow.revision, {'x':1., 'y':.5})
|
|
self.assertAlmostEqual(self.follow.pan-before, min(speed, 300)*.1)
|
|
self.follow.halt()
|
|
self.assertEqual(self.follow.status()['fps'], 0)
|
|
|
|
def test_search_is_opt_in_and_requires_prior_target_and_real_session(self):
|
|
legacy = dict(DEFAULT); del legacy['search_enabled']
|
|
self.assertFalse(validate(legacy)['search_enabled'])
|
|
with self.assertRaises(ValueError): validate(dict(DEFAULT, search_enabled=1))
|
|
self.follow.configure(dict(DEFAULT, search_enabled=True))
|
|
self.assertTrue(FollowService(self.camera,Path(self.folder.name)).config['search_enabled'])
|
|
self.start()
|
|
for _ in range(6): self.step(None)
|
|
self.assertTrue(self.follow.active) # no blind search on start
|
|
self.assertTrue(self.follow.waiting_for_target)
|
|
self.follow.halt()
|
|
self.follow.begin(False)
|
|
self.step({'x':.7,'y':.7})
|
|
for _ in range(10): self.step(None)
|
|
self.assertFalse(self.follow.status()['searching'])
|
|
self.assertEqual(len(self.drivers),1) # preview creates no driver
|
|
|
|
def test_search_sweep_limits_speed_and_finite_completion(self):
|
|
for reverse in (False, True):
|
|
self.follow.halt()
|
|
self.follow.configure(dict(DEFAULT,search_enabled=True,pan_reverse=reverse,
|
|
tilt_reverse=reverse,pan_min=1420,pan_max=1580,tilt_min=1430,tilt_max=1570,
|
|
follow_speed=800))
|
|
self.start(); self.step({'x':.8,'y':.8})
|
|
driver=self.drivers[-1]; before=self.follow.pan
|
|
self.step(None); self.step(None)
|
|
self.assertEqual(self.follow.pan,before) # brief occlusion: hold
|
|
for _ in range(6): self.step(None)
|
|
self.assertTrue(self.follow.status()['searching'])
|
|
self.assertGreater((self.follow.pan-before)*(-1 if reverse else 1),0)
|
|
start=len(driver.moves)-1
|
|
for _ in range(200):
|
|
if not self.follow.active: break
|
|
self.step(None)
|
|
self.assertTrue(self.follow.active)
|
|
self.assertFalse(driver.closed)
|
|
self.assertTrue(self.follow.waiting_for_target)
|
|
for pan,tilt in driver.moves:
|
|
self.assertTrue(1420 <= pan <= 1580)
|
|
self.assertTrue(1430 <= tilt <= 1570)
|
|
for a,b in zip(driver.moves[start:],driver.moves[start+1:]):
|
|
self.assertLessEqual(abs(a[0]-b[0]),20.0001)
|
|
self.assertLessEqual(abs(a[1]-b[1]),20.0001)
|
|
self.assertEqual({p for p,t in driver.moves} & {1420,1580},{1420,1580})
|
|
self.assertEqual({t for p,t in driver.moves} & {1430,1570},{1430,1570})
|
|
|
|
def test_search_reacquisition_holds_then_resumes_and_new_loss_searches_again(self):
|
|
self.follow.configure(dict(DEFAULT,search_enabled=True))
|
|
self.start(); self.step({'x':.7,'y':.5})
|
|
for _ in range(8): self.step(None)
|
|
before=(self.follow.pan,self.follow.tilt)
|
|
self.step({'x':.3,'y':.5})
|
|
self.assertEqual((self.follow.pan,self.follow.tilt),before)
|
|
self.assertTrue(self.follow.status()['reacquiring'])
|
|
self.step(None) # isolated detection must not end search
|
|
self.assertFalse(self.follow.status()['reacquiring'])
|
|
self.step({'x':.5,'y':.5}); self.step({'x':.5,'y':.5})
|
|
self.assertFalse(self.follow.status()['searching'])
|
|
self.assertTrue(self.follow.active)
|
|
for _ in range(8): self.step(None)
|
|
self.assertTrue(self.follow.status()['searching'])
|
|
self.follow.halt()
|
|
self.assertFalse(self.follow.status()['searching'])
|
|
self.start()
|
|
self.assertIsNone(self.follow.last_target)
|
|
|
|
def test_search_timeout_locked_axes_and_failure_watchdogs(self):
|
|
self.follow.configure(dict(DEFAULT,search_enabled=True,follow_speed=40,
|
|
pan_min=500,pan_max=2500,pan_setup_min=500,pan_setup_max=2500,
|
|
tilt_min=1500,tilt_max=1500))
|
|
self.start(); self.step({'x':.9,'y':.8})
|
|
for _ in range(310):
|
|
if not self.follow.active: break
|
|
self.step(None)
|
|
self.assertTrue(self.follow.active)
|
|
self.assertTrue(self.follow.waiting_for_target)
|
|
self.assertFalse(self.follow.status()['searching'])
|
|
self.follow.halt()
|
|
self.assertTrue(all(t==1500 for p,t in self.drivers[-1].moves))
|
|
for reason in ('camera','heartbeat','inference'):
|
|
self.camera.captured=self.now
|
|
self.start(); self.step({'x':.9,'y':.5})
|
|
for _ in range(8): self.step(None)
|
|
self.assertTrue(self.follow.status()['searching'])
|
|
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(self.drivers[-1].closed,reason)
|
|
|
|
def test_missing_target_waits_indefinitely_and_reacquires_without_restarting(self):
|
|
self.follow.configure(dict(DEFAULT,hue=179))
|
|
session=self.start()
|
|
self.step({'x':.7,'y':.5})
|
|
driver=self.drivers[-1]; moves=len(driver.moves)
|
|
for _ in range(900): self.step(None) # three minutes, healthy connection
|
|
self.assertTrue(self.follow.active)
|
|
self.assertEqual(self.follow.session,session)
|
|
self.assertEqual(len(driver.moves),moves)
|
|
self.assertTrue(self.follow.status()['waiting_for_target'])
|
|
self.step({'x':.3,'y':.5})
|
|
self.assertEqual(len(driver.moves),moves)
|
|
self.step({'x':.3,'y':.5})
|
|
self.assertFalse(self.follow.waiting_for_target)
|
|
self.assertEqual(len(driver.moves),moves+1)
|
|
self.assertEqual(self.follow.config['hue'],179)
|
|
self.follow.halt()
|
|
restored=FollowService(self.camera,Path(self.folder.name))
|
|
self.assertFalse(restored.active)
|
|
self.assertEqual(restored.config['hue'],179)
|
|
self.follow.configure(dict(self.follow.config,hue=0))
|
|
self.assertEqual(FollowService(self.camera,Path(self.folder.name)).config['hue'],0)
|
|
|
|
def test_finished_search_waits_without_repeating_and_reacquires(self):
|
|
self.follow.configure(dict(DEFAULT,search_enabled=True))
|
|
self.start(); self.step({'x':.7,'y':.5})
|
|
for _ in range(500): self.step(None)
|
|
self.assertTrue(self.follow.waiting_for_target)
|
|
count=len(self.drivers[-1].moves)
|
|
for _ in range(100): self.step(None)
|
|
self.assertEqual(len(self.drivers[-1].moves),count)
|
|
self.step({'x':.3,'y':.5}); self.step({'x':.3,'y':.5})
|
|
self.assertFalse(self.follow.waiting_for_target)
|
|
self.assertTrue(self.follow.active)
|
|
self.assertGreater(len(self.drivers[-1].moves),count)
|
|
self.now+=2; self.camera.captured=self.now
|
|
self.follow.watchdog()
|
|
self.assertFalse(self.follow.active)
|
|
self.assertTrue(self.drivers[-1].closed)
|
|
|
|
def test_start_uses_persisted_independent_endpoint_midpoints(self):
|
|
config = dict(DEFAULT, pan_min=1000, pan_max=1800,
|
|
pan_setup_min=500, pan_setup_max=2500,
|
|
tilt_min=1700, tilt_max=2001, tilt_setup_max=2300,
|
|
pan_reverse=True, tilt_reverse=True)
|
|
self.follow.configure(config)
|
|
for operation in ('setup', 'follow'):
|
|
driver = Driver()
|
|
service = FollowService(self.camera, Path(self.folder.name),
|
|
lambda: driver, lambda: self.now)
|
|
self.assertFalse(service.active)
|
|
self.assertEqual(driver.moves, []) # restart does not move anything
|
|
service.begin(True, True, operation)
|
|
self.assertEqual(driver.moves, [(1400,1851)])
|
|
self.assertEqual(service.status()['setup_origin'], {'pan':1400,'tilt':1851})
|
|
service.halt()
|
|
self.follow.configure(dict(config,pan_min=1200,pan_max=1200))
|
|
self.start()
|
|
self.assertEqual(self.drivers[-1].moves, [(1200,1851)])
|
|
|
|
def test_remember_uses_fixed_session_origin_and_next_start_recalculates(self):
|
|
self.follow.configure(dict(DEFAULT,pan_min=1700,pan_max=1800))
|
|
session = self.follow.begin(True,True,'setup')
|
|
self.assertEqual(self.follow.pan,1750)
|
|
for seq in range(1,5):
|
|
self.follow.jog(session,'left',seq); self.step(None)
|
|
self.follow.jog(session,None,5)
|
|
count = len(self.drivers[-1].moves)
|
|
with self.assertRaises(ValueError): self.follow.remember(session,'right')
|
|
saved = self.follow.remember(session,'left')
|
|
self.assertEqual(saved['pan_min'],1728)
|
|
self.assertEqual(self.follow.status()['setup_origin']['pan'],1750)
|
|
self.assertEqual(len(self.drivers[-1].moves),count)
|
|
self.follow.halt(); self.start()
|
|
self.assertEqual(self.drivers[-1].moves,[(1764,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 ('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=1601), 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_custom_setup_envelope_persistence_migration_and_follow_limits(self):
|
|
old = {k:v for k,v in DEFAULT.items() if '_setup_' not in k}
|
|
self.assertEqual(validate(old), DEFAULT)
|
|
config = dict(DEFAULT, pan_setup_min=700, pan_setup_max=2300,
|
|
tilt_setup_min=1000, tilt_setup_max=2000)
|
|
self.follow.configure(config)
|
|
self.assertEqual(FollowService(self.camera,Path(self.folder.name)).config, config)
|
|
session = self.follow.begin(True,True,'setup')
|
|
with self.assertRaises(ValueError): self.follow.configure(DEFAULT)
|
|
for seq in range(1,130):
|
|
self.follow.jog(session,'right',seq); self.step(None)
|
|
self.assertEqual(self.follow.pan,2300)
|
|
self.assertEqual(self.follow.tilt,1500)
|
|
self.follow.jog(session,None,130)
|
|
self.assertEqual(self.follow.remember(session,'right')['pan_max'],2290)
|
|
for seq in range(131,350):
|
|
self.follow.jog(session,'left',seq); self.step(None)
|
|
self.assertEqual(self.follow.pan,700)
|
|
self.follow.jog(session,None,350)
|
|
self.assertEqual(self.follow.remember(session,'left')['pan_min'],710)
|
|
self.follow.halt(); self.start()
|
|
for _ in range(100): self.step({'x':1,'y':1})
|
|
self.assertEqual(self.follow.pan,2290)
|
|
self.assertEqual(self.follow.tilt,1600)
|
|
for bad in (dict(config,pan_setup_min=499), dict(config,tilt_setup_max=2501),
|
|
dict(config,pan_min=600), dict(config,tilt_max=2100)):
|
|
with self.assertRaises(ValueError): validate(bad)
|
|
|
|
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_center_moves_slowly_to_current_saved_limits_without_saving(self):
|
|
self.follow.configure(dict(DEFAULT,pan_min=1200,pan_max=1700,
|
|
tilt_min=1400,tilt_max=1800))
|
|
session = self.follow.begin(True,True,'setup')
|
|
for seq in range(1,11):
|
|
self.follow.jog(session,'right',seq); self.step(None)
|
|
self.follow.jog(session,None,11)
|
|
self.follow.remember(session,'right')
|
|
for seq in range(12,22):
|
|
self.follow.jog(session,'up',seq); self.step(None)
|
|
self.follow.jog(session,None,22)
|
|
saved = self.follow.path.read_bytes()
|
|
before = (self.follow.pan,self.follow.tilt)
|
|
self.assertTrue(self.follow.center(session,23)['accepted'])
|
|
self.assertEqual((self.follow.pan,self.follow.tilt),before) # no sudden jump
|
|
with self.assertRaises(ValueError): self.follow.remember(session,'left')
|
|
self.step(None)
|
|
self.assertAlmostEqual(self.follow.pan,before[0]-8)
|
|
self.assertAlmostEqual(self.follow.tilt,before[1]+8)
|
|
for _ in range(40): self.step(None)
|
|
self.assertEqual((self.follow.pan,self.follow.tilt),(1360,1600))
|
|
self.assertFalse(self.follow.status()['centering'])
|
|
self.assertEqual(self.follow.path.read_bytes(),saved)
|
|
self.assertEqual(self.follow.setup_origin,{'pan':1450,'tilt':1600})
|
|
|
|
def test_center_cancellation_stale_commands_watchdog_and_session(self):
|
|
with self.assertRaises(ValueError): self.follow.center('none',1)
|
|
session = self.start()
|
|
with self.assertRaises(ValueError): self.follow.center(session,1)
|
|
self.follow.halt(); session = self.follow.begin(True,True,'setup')
|
|
with self.assertRaises(ValueError): self.follow.center('wrong',1)
|
|
self.follow.jog(session,'right',1); self.step(None)
|
|
self.follow.center(session,2)
|
|
self.follow.jog(session,None,3)
|
|
stopped = self.follow.pan
|
|
self.assertFalse(self.follow.center(session,2)['accepted'])
|
|
self.step(None); self.assertEqual(self.follow.pan,stopped)
|
|
self.follow.center(session,4)
|
|
self.follow.jog(session,'right',5); self.step(None)
|
|
self.assertGreater(self.follow.pan,stopped)
|
|
self.assertFalse(self.follow.status()['centering'])
|
|
self.follow.center(session,6)
|
|
self.now+=2; self.camera.captured=self.now
|
|
self.follow.watchdog()
|
|
self.assertFalse(self.follow.active)
|
|
self.assertFalse(self.follow.status()['centering'])
|
|
self.assertTrue(self.drivers[-1].closed)
|
|
with self.assertRaises(ValueError): self.follow.center(session,7)
|
|
|
|
def test_center_api_requires_header_and_setup_session(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
|
|
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/center',json={})).status_code,403)
|
|
self.assertEqual((await c.post('/api/follow/center',headers=h,json={})).status_code,422)
|
|
start=await c.post('/api/follow/start',headers=h,json={'operation':'setup'})
|
|
session=start.json()['session']
|
|
result=await c.post('/api/follow/center',headers=h,json={'session':session,'sequence':1})
|
|
self.assertEqual(result.status_code,200)
|
|
self.assertTrue(result.json()['accepted'])
|
|
await c.post('/api/follow/stop',headers=h,json={})
|
|
self.assertFalse(app.state.follow.status()['centering'])
|
|
asyncio.run(scenario())
|
|
|
|
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':True,'acknowledged':True})).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,200)
|
|
self.assertFalse(app.state.controller.stopped)
|
|
asyncio.run(scenario())
|
|
|
|
if __name__=='__main__': unittest.main()
|