PitresAli
8ccd7689f7
Add configurable Xbox motion and action bindings, frame-aware six-axis MAVLink controls, sensor logging, camera viewing and recording, and a faster startup path. Synchronize both app copies and document companion hardware setup and verification limits.
166 lines
7.5 KiB
Python
166 lines
7.5 KiB
Python
"""V2 protocol and persistence checks. Never opens a vehicle connection."""
|
|
import math
|
|
import queue
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, patch
|
|
|
|
from app import Axes, MavlinkWorker, normalize_endpoint
|
|
from rov_v2 import Telemetry, clean_station, clean_profile, decode_telemetry, save_json
|
|
from web_ui import ControlApi
|
|
|
|
|
|
class V2Tests(unittest.TestCase):
|
|
def worker(self, frame=0, version=(4, 5, 3)):
|
|
worker = MavlinkWorker(queue.Queue())
|
|
worker.connection = Mock()
|
|
worker.target_system = 1
|
|
worker.frame_type = frame
|
|
worker.firmware_version = version
|
|
worker.gain = .5
|
|
worker.last_control_at = time.monotonic() - 1
|
|
return worker
|
|
|
|
def test_six_axis_mapping_and_envelope(self):
|
|
worker = self.worker()
|
|
worker.set_axes(Axes(forward=1, lateral=1, vertical=1, yaw=1, pitch=1, roll=1))
|
|
worker._send_control()
|
|
args = worker.connection.mav.manual_control_send.call_args.args
|
|
self.assertEqual(args[6:8], (0, 3))
|
|
self.assertEqual(args[8:], (33, 33)) # S=pitch, T=roll; ArduSub 4.5 ±400.
|
|
self.assertEqual(args[1:5], (83, 83, 542, 83))
|
|
|
|
def test_modern_normalized_attitude(self):
|
|
worker = self.worker(version=(4, 8, 0))
|
|
worker.set_axes(Axes(pitch=.5, roll=-.5))
|
|
worker._send_control()
|
|
self.assertEqual(worker.connection.mav.manual_control_send.call_args.args[8:], (250, -250))
|
|
|
|
def test_unsupported_frame_and_old_firmware(self):
|
|
worker = self.worker(frame=1)
|
|
worker.set_axes(Axes(pitch=1, roll=.2))
|
|
worker._send_control()
|
|
self.assertEqual(worker.connection.mav.manual_control_send.call_args.args[8], 0)
|
|
worker.firmware_version = (4, 1, 0)
|
|
worker._send_control()
|
|
self.assertEqual(len(worker.connection.mav.manual_control_send.call_args.args), 6)
|
|
|
|
def test_watchdog_neutralizes_extended_axes(self):
|
|
worker = self.worker()
|
|
worker.require_input_lease = True
|
|
worker.armed = True
|
|
worker.last_heartbeat = time.monotonic()
|
|
worker.set_axes(Axes(pitch=1, roll=-1), 1)
|
|
worker.last_input_at -= 2
|
|
worker._send_control()
|
|
self.assertEqual(worker.connection.mav.manual_control_send.call_args.args, (1, 0, 0, 500, 0, 0, 0, 3, 0, 0))
|
|
self.assertEqual(worker.axes, Axes())
|
|
|
|
def test_endpoints_preserve_configured_network_host(self):
|
|
for endpoint in ('udpin:0.0.0.0:14550', 'udpout:192.168.2.2:14550', 'tcp:192.168.2.2:5760'):
|
|
self.assertEqual(normalize_endpoint(endpoint), endpoint)
|
|
|
|
def test_sensor_units_and_staleness(self):
|
|
msg = SimpleNamespace(get_type=lambda: 'ATTITUDE', roll=math.pi/2, pitch=0, yaw=-math.pi/2, rollspeed=1, pitchspeed=0, yawspeed=0)
|
|
group, values = decode_telemetry(msg)
|
|
self.assertEqual(values['roll'], 90)
|
|
self.assertEqual(values['heading'], 270)
|
|
clock = [10]
|
|
store = Telemetry(lambda: clock[0])
|
|
store.update(group, values)
|
|
clock[0] = 14
|
|
self.assertTrue(store.snapshot()[group]['stale'])
|
|
msg = SimpleNamespace(get_type=lambda: 'GLOBAL_POSITION_INT', relative_alt=-1200, vz=25)
|
|
self.assertEqual(decode_telemetry(msg)[1], {'depth_m': 1.2, 'vertical_ms': .25})
|
|
|
|
def test_config_validation_and_atomic_roundtrip(self):
|
|
with self.assertRaises(ValueError):
|
|
clean_station({'video_url': 'javascript:alert(1)'})
|
|
with self.assertRaises(ValueError):
|
|
clean_station({'lights_relay': 1, 'camera_relay': 1})
|
|
with self.assertRaises(ValueError):
|
|
clean_profile({'version': 2, 'expo': float('nan')})
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
target = Path(folder) / 'station.json'
|
|
save_json(target, clean_station({'frame': 2}))
|
|
import json
|
|
self.assertEqual(json.loads(target.read_text())['frame'], 2)
|
|
self.assertFalse(target.with_suffix('.json.tmp').exists())
|
|
|
|
def test_stale_current_is_not_refreshed_by_voltage(self):
|
|
api = ControlApi()
|
|
api.state['connected'] = True
|
|
with patch('web_ui.time.monotonic', return_value=10):
|
|
api._consume_battery({'current_a': 4, 'voltage': 16})
|
|
with patch('web_ui.time.monotonic', return_value=16):
|
|
api._consume_battery({'voltage': 15.9})
|
|
self.assertIsNone(api._snapshot()['battery']['current_a'])
|
|
with patch('web_ui.time.monotonic', return_value=17):
|
|
api._consume_battery({'current_a': None})
|
|
self.assertIsNone(api.state['battery']['current_a'])
|
|
|
|
def test_attitude_stream_is_never_disabled_when_requested(self):
|
|
worker = self.worker()
|
|
worker._configure_low_bandwidth_streams()
|
|
rates = [(name, interval) for name, _, interval in worker.stream_config]
|
|
self.assertIn(('ATTITUDE', 500000), rates)
|
|
self.assertNotIn(('ATTITUDE', -1), rates)
|
|
|
|
def test_network_control_omits_serial_priming_heartbeat(self):
|
|
worker = self.worker()
|
|
worker.network_link = True
|
|
worker._send_control()
|
|
worker.connection.mav.heartbeat_send.assert_not_called()
|
|
|
|
def test_profile_saves_atomically_and_cannot_change_while_armed(self):
|
|
api = ControlApi()
|
|
profile = {'version': 2, 'bindings': {}, 'actions': {}}
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
api.profiles_path = Path(folder) / 'profiles.json'
|
|
api.save_controller_profile('$xbox', profile)
|
|
self.assertTrue(api.profiles_path.is_file())
|
|
before = api.profiles_path.read_bytes()
|
|
api.state.update(connected=True, armed=True)
|
|
with self.assertRaises(RuntimeError):
|
|
api.save_controller_profile('$xbox', profile)
|
|
self.assertEqual(before, api.profiles_path.read_bytes())
|
|
|
|
def test_native_recording_chunk_roundtrip_and_expired_token(self):
|
|
import base64
|
|
api = ControlApi()
|
|
with tempfile.TemporaryDirectory() as folder, patch('web_ui.__file__', str(Path(folder) / 'web_ui.py')):
|
|
token = api.recording_begin()
|
|
api.recording_chunk(token, base64.b64encode(b'video chunk').decode())
|
|
path = api.recording_end(token)
|
|
self.assertEqual(Path(path).read_bytes(), b'video chunk')
|
|
with self.assertRaises(ValueError):
|
|
api.recording_chunk(token, 'AA==')
|
|
|
|
def test_relay_action_waits_for_ack(self):
|
|
api = ControlApi()
|
|
api.station_settings['lights_relay'] = 0
|
|
worker = api.worker
|
|
worker.connection = Mock()
|
|
worker.target_system = worker.target_component = 1
|
|
worker.last_heartbeat = time.monotonic()
|
|
api.set_accessory('lights', True)
|
|
self.assertIsNone(api.state['accessories']['lights']['accepted'])
|
|
api._consume(('accessory_ack', 'lights', True, 2))
|
|
self.assertFalse(api.state['accessories']['lights']['accepted'])
|
|
self.assertEqual(api.state['accessories']['lights']['error'], 'Rejected')
|
|
|
|
def test_motor_names_survive_app_restart(self):
|
|
with tempfile.TemporaryDirectory() as folder, patch('web_ui.__file__', str(Path(folder) / 'web_ui.py')):
|
|
first = ControlApi()
|
|
settings = first.motor_settings
|
|
settings['motors']['1']['name'] = 'Port'
|
|
first.save_motor_settings(settings)
|
|
self.assertEqual(ControlApi().motor_settings['motors']['1']['name'], 'Port')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|