69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
# Minimal test harness to inject a fake RIS payload into DebugPayloadRouter
|
|
import logging
|
|
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s')
|
|
|
|
from target_simulator.gui.payload_router import DebugPayloadRouter
|
|
from target_simulator.analysis.simulation_state_hub import SimulationStateHub
|
|
import target_simulator.gui.payload_router as pr_mod
|
|
|
|
# Build fake ris target objects with expected attributes
|
|
class FakeRisTarget:
|
|
def __init__(self, flags, x, y, z, heading):
|
|
self.flags = flags
|
|
self.x = x
|
|
self.y = y
|
|
self.z = z
|
|
self.heading = heading
|
|
|
|
class FakeScenario:
|
|
def __init__(self, timetag=123456789):
|
|
self.timetag = timetag
|
|
|
|
class FakeParsed:
|
|
def __init__(self, tgt_list):
|
|
# parsed.tgt.tgt is iterated in payload_router
|
|
self.tgt = type('T', (), {'tgt': tgt_list})()
|
|
self.scenario = FakeScenario()
|
|
|
|
@staticmethod
|
|
def from_buffer_copy(payload):
|
|
# This will be monkeypatched per-call by assigning a closure; but keep for safety
|
|
raise RuntimeError('Should be monkeypatched')
|
|
|
|
# Prepare fake targets using your example headings (radians)
|
|
rads = [0.0, 1.5707964897155762, -3.141592502593994, -1.5707964897155762]
|
|
# Example positions (x,y,z) in meters approx; router expects M_TO_FT conversion inside
|
|
fake_targets = []
|
|
for i, h in enumerate(rads):
|
|
# Set flags non-zero
|
|
# Use x,y values that roughly match earlier logs (meters)
|
|
x_m = 9600 + i * 6000
|
|
y_m = 37000 - i * 5000
|
|
z_m = 3048.0 # 10000 ft ~= 3048 m
|
|
fake_targets.append(FakeRisTarget(flags=1, x=x_m, y=y_m, z=z_m, heading=h))
|
|
|
|
# Monkeypatch the SfpRisStatusPayload.from_buffer_copy in payload_router module
|
|
original_parser = pr_mod.SfpRisStatusPayload
|
|
|
|
def fake_from_buffer_copy(payload):
|
|
return FakeParsed(fake_targets)
|
|
|
|
pr_mod.SfpRisStatusPayload = type('P', (), {'from_buffer_copy': staticmethod(fake_from_buffer_copy)})
|
|
|
|
# Run the router handling
|
|
hub = SimulationStateHub()
|
|
router = DebugPayloadRouter(simulation_hub=hub, update_queue=None)
|
|
|
|
print('Invoking _handle_ris_status with fake payload...')
|
|
router._handle_ris_status(b'FAKE')
|
|
|
|
# Show hub contents and stored headings
|
|
for tid in range(len(rads)):
|
|
hist = hub.get_target_history(tid)
|
|
hdg = hub.get_real_heading(tid)
|
|
print(f'TID {tid} -> history entries: sim={len(hist["simulated"]) if hist else 0} real={len(hist["real"]) if hist else 0}, heading_in_hub={hdg}')
|
|
|
|
# Restore original parser symbol to avoid side effects
|
|
pr_mod.SfpRisStatusPayload = original_parser
|
|
print('Done')
|