phantom-proof/demo_show.py at main · eldaeonuap/phantom-proof · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
#!/usr/bin/env python3
"""
demo_show.py — scripted Fantom Finder demo orchestrator.

Walks through a polished sequence of Remote ID broadcasts that map to all
three verdict kinds, paced for a 60-90 second on-stage demo:

    t+0    -> claim matching TRK-MAVIC-1   -> CONFIRMED
    t+8    -> claim with no track          -> PHANTOM #1
    t+10   -> claim with no track          -> PHANTOM #2
    t+12   -> claim with no track          -> PHANTOM #3
    t+22   -> claim "Aeroplane" matching TRK-MAVIC-1 (quadcopter) -> DECEPTION
    t+30   -> claim matching TRK-CESSNA-1  -> CONFIRMED
    t+38   -> hold, fade out

Use this when you want a hands-off rehearsal or the live RF spoofer isn't ready.
Runs against the same UDP listener as spoofer.py — both target fusion.py:7878.

Usage:
    python3 demo_show.py --target localhost:7878
    python3 demo_show.py --target localhost:7878 --speed 1.5    # 1.5x faster
"""
import argparse
import json
import socket
import time

# Known synthetic track positions matching fake_tracker.py
ANCHOR = (37.79504, -122.39349)


def msg(uas_id: str, ua_type: str, lat_off_m: float, lon_off_m: float, *,
        anchor=ANCHOR, alt=30, speed=4, heading=180):
    R = 6_371_000.0
    lat = anchor[0] + (lat_off_m / R) * 57.2958
    lon = anchor[1] + (lon_off_m / (R * 0.7)) * 57.2958
    return {
        "ts": time.time(),
        "basic_id": {
            "uas_id": uas_id,
            "ua_type": ua_type,
            "ua_class": "C0",
        },
        "location": {
            "latitude": lat,
            "longitude": lon,
            "altitude_m": alt,
            "speed_m_s": speed,
            "heading_deg": heading,
        },
        "operator_id": {"operator_id": f"OP-{uas_id[:6]}"},
        "_demo_show": True,
    }


# Target offsets matched to fake_tracker.py's TRK-MAVIC-1 (-50,+120) and
# TRK-CESSNA-1 (0,+150). For PHANTOMs we pick clearly disjoint offsets.
SCHEDULE = [
    # (delay_s, "label",          msg_kwargs)
    (0.0,  "CONFIRMED quadrotor",  ("VENDOR-MAVIC-2A", "Helicopter_Multirotor", -50, 120)),
    (8.0,  "PHANTOM #1",           ("PHANTOM-A1B2C3",  "Aeroplane",             400, -300)),
    (10.0, "PHANTOM #2",           ("PHANTOM-D4E5F6",  "Helicopter_Multirotor",-200, -400)),
    (12.0, "PHANTOM #3",           ("PHANTOM-G7H8I9",  "Helicopter_Multirotor", 50, -250)),
    (22.0, "DECEPTION (Aeroplane claim, quadrotor in radar)",
                                   ("FAKE-CESSNA-99",  "Aeroplane",             -50, 120)),
    (30.0, "CONFIRMED fixed-wing", ("REAL-N12345",     "Aeroplane",             0,   150)),
]


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--target", default="localhost:7878")
    p.add_argument("--speed", type=float, default=1.0,
                   help="time-compression factor (>1 plays faster)")
    p.add_argument("--once", action="store_true",
                   help="emit each event exactly once (default true)")
    args = p.parse_args()

    host, port = args.target.split(":")
    port = int(port)
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    print(f"# demo_show -> {host}:{port}, speed {args.speed}x")
    print(f"# (run alongside fusion.py + fake_tracker.py)\n")

    t0 = time.time()
    for delay, label, kwargs in SCHEDULE:
        target_t = t0 + delay / args.speed
        wait = max(0, target_t - time.time())
        if wait:
            time.sleep(wait)
        m = msg(*kwargs)
        sock.sendto(json.dumps(m).encode("utf-8"), (host, port))
        print(f"[{time.strftime('%H:%M:%S')}] {label:50s} -> {m['basic_id']['uas_id']}",
              flush=True)

    print("\n# done. Check dashboard at http://localhost:7879/")


if __name__ == "__main__":
    main()