Private
Public Access
301 lines
12 KiB
Python
301 lines
12 KiB
Python
import asyncio
|
|
from io import BytesIO
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
from PIL import Image
|
|
import httpx
|
|
|
|
from halloween.controller import Controller
|
|
from halloween.detection import DEFAULT_CONFIG, DetectionService, LineTracker, NanoDet, validate_config
|
|
from halloween.event_store import EventStore
|
|
from halloween.web import create_app
|
|
|
|
|
|
def person(x=0.5, y=0.4):
|
|
return {"box": [x-0.03, y-0.1, x+0.03, y], "score": 0.9}
|
|
|
|
|
|
class TrackerTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tracker = LineTracker()
|
|
self.config = json.loads(json.dumps(DEFAULT_CONFIG))
|
|
self.time = 0
|
|
|
|
def step(self, people):
|
|
self.time += 0.2
|
|
return self.tracker.update(people, self.config, self.time)[1]
|
|
|
|
def test_forward_and_reverse(self):
|
|
for _ in range(3):
|
|
self.assertFalse(self.step([person()]))
|
|
self.assertTrue(self.step([person(y=0.65)]))
|
|
self.assertFalse(self.step([person(y=0.4)]))
|
|
self.assertFalse(self.step([person(y=0.4)]))
|
|
|
|
def test_reverse_direction_configuration(self):
|
|
self.config['direction'] = -1
|
|
for _ in range(3):
|
|
self.assertFalse(self.step([person(y=0.65)]))
|
|
self.assertTrue(self.step([person(y=0.4)]))
|
|
|
|
def test_appearance_behind_line(self):
|
|
for _ in range(5):
|
|
self.assertFalse(self.step([person(y=0.65)]))
|
|
|
|
def test_jitter_and_parallel_motion(self):
|
|
for y in [0.54, 0.56, 0.54, 0.56, 0.54, 0.56]:
|
|
self.assertFalse(self.step([person(y=y)]))
|
|
for x in [0.45, 0.5, 0.55, 0.6]:
|
|
self.assertFalse(self.step([person(x=x, y=0.4)]))
|
|
|
|
def test_flicker_and_outside_segment(self):
|
|
for _ in range(3):
|
|
self.step([person()])
|
|
self.step([])
|
|
self.assertFalse(self.step([person(y=0.65)]))
|
|
self.setUp()
|
|
for _ in range(3):
|
|
self.step([person(x=0.15)])
|
|
self.assertFalse(self.step([person(x=0.15, y=0.65)]))
|
|
|
|
|
|
class ServiceTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.folder = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.folder.cleanup)
|
|
self.directory = Path(self.folder.name)
|
|
self.time = 0
|
|
self.controller = Controller(self.directory, clock=lambda: self.time)
|
|
self.events = EventStore(self.directory)
|
|
self.camera = type('Camera', (), {'error': None})()
|
|
self.service = DetectionService(self.camera, self.controller, self.events,
|
|
self.directory, clock=lambda: self.time)
|
|
self.service.configure({**DEFAULT_CONFIG, 'enabled': True})
|
|
buffer = BytesIO()
|
|
Image.new('RGB', (640, 480)).save(buffer, format='JPEG')
|
|
self.jpeg = buffer.getvalue()
|
|
self.number = 0
|
|
|
|
def step(self, people=(), dt=0.2):
|
|
self.time += dt
|
|
self.number += 1
|
|
self.service.process(list(people), self.number, self.jpeg, self.time, self.service.revision)
|
|
|
|
def free(self):
|
|
for _ in range(12):
|
|
self.step()
|
|
|
|
def cross(self, x=0.5):
|
|
for _ in range(3):
|
|
self.step([person(x=x)])
|
|
self.step([person(x=x, y=0.65)])
|
|
|
|
def test_group_cooldown_and_new_group(self):
|
|
self.free()
|
|
self.assertTrue(self.service.status()['armed'])
|
|
self.cross()
|
|
self.assertEqual(len(self.events.recent()), 1)
|
|
event = self.events.recent()[0]
|
|
self.assertEqual(event['kind'], 'person_line_crossing')
|
|
self.assertIsNone(event['profile_id'])
|
|
self.cross(x=0.65)
|
|
self.assertEqual(len(self.events.recent()), 1)
|
|
for _ in range(60):
|
|
self.step([person(y=0.65)])
|
|
self.cross(x=0.65)
|
|
self.assertEqual(len(self.events.recent()), 1)
|
|
self.free()
|
|
self.cross()
|
|
self.assertEqual(len(self.events.recent()), 2)
|
|
|
|
def test_no_delayed_crossing(self):
|
|
self.free()
|
|
self.cross()
|
|
self.free()
|
|
self.cross(x=0.65)
|
|
self.assertEqual(len(self.events.recent()), 1)
|
|
for _ in range(60):
|
|
self.step([person(x=0.65, y=0.65)])
|
|
self.assertEqual(len(self.events.recent()), 1)
|
|
|
|
def test_start_stop_reset_and_configuration(self):
|
|
self.cross()
|
|
self.assertEqual(self.events.recent(), [])
|
|
self.free()
|
|
self.controller.stop()
|
|
self.cross()
|
|
self.assertEqual(self.events.recent(), [])
|
|
self.controller.reset()
|
|
self.service.reset()
|
|
self.cross()
|
|
self.assertEqual(self.events.recent(), [])
|
|
self.free()
|
|
old_revision = self.service.revision
|
|
self.service.configure(self.service.get_config())
|
|
self.service.process([], 99, self.jpeg, self.time, old_revision)
|
|
self.assertIsNone(self.service.last_valid)
|
|
self.cross()
|
|
self.assertEqual(self.events.recent(), [])
|
|
self.free()
|
|
self.cross()
|
|
self.assertEqual(len(self.events.recent()), 1)
|
|
|
|
def test_watchdog_and_manual_bypass(self):
|
|
self.free()
|
|
self.time += 2.01
|
|
self.service.watchdog()
|
|
self.assertTrue(self.controller.stopped)
|
|
self.assertTrue(self.service.status()['reset_required'])
|
|
self.controller.zone_clear()
|
|
self.assertFalse(self.controller.trigger()[0])
|
|
self.free()
|
|
self.assertTrue(self.service.status()['reset_required'])
|
|
self.controller.reset()
|
|
self.service.reset()
|
|
self.free()
|
|
self.cross()
|
|
self.assertEqual(len(self.events.recent()), 1)
|
|
|
|
def test_camera_error_and_stale_results(self):
|
|
self.free()
|
|
self.camera.error = 'camera disconnected'
|
|
self.service.watchdog()
|
|
self.assertTrue(self.service.fault)
|
|
self.assertTrue(self.controller.stopped)
|
|
self.time += 3
|
|
result = self.service.snapshot()
|
|
self.assertEqual(result['people'], [])
|
|
self.assertIsNone(result['image_base64'])
|
|
|
|
def test_disabled_detection_does_not_stop_manual_tests(self):
|
|
self.service.configure(DEFAULT_CONFIG)
|
|
self.service.fail('missing model')
|
|
self.assertFalse(self.controller.stopped)
|
|
self.controller.zone_clear()
|
|
self.assertTrue(self.controller.trigger()[0])
|
|
|
|
def test_inference_stall_is_stopped_by_independent_watchdog(self):
|
|
entered, release = threading.Event(), threading.Event()
|
|
class BlockingDetector:
|
|
def detect(inner, jpeg):
|
|
entered.set()
|
|
release.wait(3)
|
|
return []
|
|
self.camera.latest_timed = lambda: (1, self.jpeg, self.time)
|
|
self.service.factory = lambda path: BlockingDetector()
|
|
self.service.start()
|
|
try:
|
|
self.assertTrue(entered.wait(1))
|
|
self.time += 2.1
|
|
self.service.watchdog()
|
|
self.assertTrue(self.controller.stopped)
|
|
self.assertTrue(self.service.fault)
|
|
finally:
|
|
release.set()
|
|
self.service.stop()
|
|
|
|
def test_database_stall_does_not_block_watchdog(self):
|
|
entered, release = threading.Event(), threading.Event()
|
|
def blocked_record(mode, **kwargs):
|
|
entered.set()
|
|
release.wait(3)
|
|
self.events.record_detection_trigger = blocked_record
|
|
self.free()
|
|
for _ in range(3):
|
|
self.step([person()])
|
|
worker = threading.Thread(target=lambda: self.step([person(y=0.65)]))
|
|
worker.start()
|
|
try:
|
|
self.assertTrue(entered.wait(1))
|
|
self.time += 2.1
|
|
self.service.watchdog()
|
|
self.assertTrue(self.controller.stopped)
|
|
finally:
|
|
release.set()
|
|
worker.join(timeout=1)
|
|
self.assertFalse(worker.is_alive())
|
|
|
|
def test_database_failure_stops_simulation(self):
|
|
def broken_record(mode, **kwargs):
|
|
raise OSError('disk full')
|
|
self.events.record_detection_trigger = broken_record
|
|
self.free()
|
|
self.cross()
|
|
self.assertTrue(self.controller.stopped)
|
|
self.assertTrue(self.service.fault)
|
|
self.assertIn('disk full', self.service.status()['error'])
|
|
|
|
def test_shutdown_drops_inflight_frame(self):
|
|
self.free()
|
|
version = self.service.revision
|
|
self.service.stop()
|
|
self.service.process([], 99, self.jpeg, self.time, version)
|
|
self.assertTrue(self.controller.stopped)
|
|
self.assertEqual(self.events.recent(), [])
|
|
|
|
def test_persistence_and_corrupt_settings(self):
|
|
other = DetectionService(self.camera, self.controller, self.events, self.directory)
|
|
self.assertTrue(other.get_config()['enabled'])
|
|
self.assertFalse(other.status()['armed'])
|
|
self.service.path.write_text('{')
|
|
other = DetectionService(self.camera, self.controller, self.events, self.directory)
|
|
self.assertFalse(other.get_config()['enabled'])
|
|
self.assertIsNotNone(other.status()['error'])
|
|
|
|
def test_unconfirmed_person_prevents_clear(self):
|
|
for _ in range(30):
|
|
self.step([person()])
|
|
self.assertFalse(self.service.status()['armed'])
|
|
|
|
def test_snapshot_matches_number_and_image(self):
|
|
self.step([person()])
|
|
snapshot = self.service.snapshot()
|
|
self.assertEqual(snapshot['frame_number'], self.number)
|
|
self.assertTrue(snapshot['image_base64'])
|
|
self.assertEqual(len(snapshot['people']), 1)
|
|
|
|
|
|
class ApiTest(unittest.TestCase):
|
|
def test_config_validation_persistence_and_missing_model(self):
|
|
async def scenario():
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
app = create_app(demo=True, state_dir=Path(folder))
|
|
async with app.router.lifespan_context(app):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as client:
|
|
headers = {'X-Halloween-UI': '1'}
|
|
config = (await client.get('/api/detection/config')).json()
|
|
self.assertFalse(config['enabled'])
|
|
self.assertEqual((await client.post('/api/detection/config', json=config)).status_code, 403)
|
|
bad = {**config, 'region': [0.8,0.2,0.3,0.9]}
|
|
self.assertEqual((await client.post('/api/detection/config', json=bad, headers=headers)).status_code, 422)
|
|
config['enabled'] = True
|
|
self.assertEqual((await client.post('/api/detection/config', json=config, headers=headers)).status_code, 200)
|
|
for _ in range(30):
|
|
await asyncio.sleep(0.05)
|
|
status = (await client.get('/api/status')).json()
|
|
if status['detection']['reset_required']:
|
|
break
|
|
self.assertTrue(status['detection']['reset_required'])
|
|
self.assertTrue(status['controller']['stopped'])
|
|
self.assertIn('NanoDet fehlt', status['detection']['error'])
|
|
result = (await client.get('/api/detection')).json()
|
|
self.assertIsNone(result['image_base64'])
|
|
self.assertTrue(create_app(demo=True, state_dir=Path(folder)).state.detection.get_config()['enabled'])
|
|
asyncio.run(scenario())
|
|
|
|
def test_invalid_values(self):
|
|
for config in [ {**DEFAULT_CONFIG, 'direction': True},
|
|
{**DEFAULT_CONFIG, 'line': [[0.5,0.5],[0.5,0.5]]},
|
|
{**DEFAULT_CONFIG, 'region': [0,0,float('nan'),1]},
|
|
{**DEFAULT_CONFIG, 'enabled': 'true'}]:
|
|
with self.assertRaises(ValueError):
|
|
validate_config(config)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|