cheng
61 lines
3.4 KiB
Python
61 lines
3.4 KiB
Python
import tempfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import TestCase
|
|
import numpy as np
|
|
from arm_application.tactile_flow import tactile_devices,select_source,flow_preview,read_sample
|
|
|
|
|
|
class TactileFlowTest(TestCase):
|
|
def test_discovery_filters_realsense_and_duplicate_interfaces(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root=Path(folder);sys=root/'class';sys.mkdir()
|
|
for name,vendor,index in [('video0','8086','0'),('video1','1bcf','0'),('video2','1bcf','1'),('video3','0bda','0')]:
|
|
entry=sys/name;entry.mkdir();(entry/'index').write_text(index)
|
|
usb=root/('usb-'+name);usb.mkdir();(usb/'idVendor').write_text(vendor)
|
|
(entry/'device').symlink_to(usb,target_is_directory=True)
|
|
found=tactile_devices(sys,Path('/dev'))
|
|
self.assertEqual(found,['/dev/video1'])
|
|
self.assertEqual(select_source('left',{},found),'/dev/video1')
|
|
self.assertIsNone(select_source('right',{},found))
|
|
self.assertEqual(select_source('right',{'right_video':'/dev/v4l/by-path/sensor'},found),'/dev/v4l/by-path/sensor')
|
|
|
|
def test_actual_vector_field_changes_preview_and_rejects_bad_shapes(self):
|
|
flow=np.zeros((64,64,2),np.float32)
|
|
rest=flow_preview(flow);flow[...,0]=12
|
|
motion=flow_preview(flow)
|
|
self.assertEqual(motion.shape,(360,360,3));self.assertFalse(np.array_equal(rest,motion))
|
|
with self.assertRaises(ValueError):flow_preview(np.zeros((8,8)))
|
|
|
|
def test_uses_raw_force_channels_and_reads_flow_only_once(self):
|
|
info=SimpleNamespace(FPS=1,FNORMAL=2,FSHEARX=3,FSHEARY=4,FNORMAL_RAW=5,FSHEARX_RAW=6,FSHEARY_RAW=7,VRAW=8)
|
|
calls=[]
|
|
def read(*keys):
|
|
calls.append(keys);return 30.,200.,3.,-4.,np.zeros((8,8,2))
|
|
row=read_sample(SimpleNamespace(info=info,read_info=read))
|
|
self.assertEqual(calls,[(1,5,6,7,8)])
|
|
self.assertEqual(row['fnormal'],200.);self.assertEqual(row['fsheary'],-4.)
|
|
|
|
def test_preview_polling_is_independent_of_slow_state_poll(self):
|
|
import base64,threading,time
|
|
from unittest.mock import patch
|
|
from station.runtime_client import RuntimeClient
|
|
client=RuntimeClient.__new__(RuntimeClient)
|
|
client._lock=threading.RLock();client._stop=threading.Event()
|
|
from collections import deque
|
|
client._force_history={'left':deque(maxlen=1800),'right':deque(maxlen=1800)}
|
|
client._force_cursors={};client._force_epochs={}
|
|
client._snapshot={'view':{'on':True}};client._frames={};client.socket=Path('/unused')
|
|
calls=[]
|
|
def rpc_call(*args,**kwargs):
|
|
calls.append(args[1]);return {'frames':{'left_shear':base64.b64encode(b'jpeg').decode()},
|
|
'samples':{'left':[{'seq':1,'t':time.time(),'fnormal':.5,'fshearx':.2,'fsheary':-.3}]},'epochs':{'left':0}}
|
|
with patch('station.runtime_client.rpc.call',side_effect=rpc_call):
|
|
thread=threading.Thread(target=client._poll_frames);thread.start()
|
|
time.sleep(.16);client._stop.set();thread.join(1)
|
|
self.assertGreaterEqual(len(calls),3)
|
|
self.assertEqual(set(calls),{'tactile_preview'})
|
|
self.assertEqual(client._frames['left_shear'],b'jpeg')
|
|
self.assertEqual(len(client._force_history['left']),1)
|
|
self.assertEqual(client._force_cursors['left'],1)
|