mikhailov.aa/leaflet_svelte
1
Fork 0
Code Issues 10 Pull requests 2 Projects 1 Releases Activity

tracker fixes + test_client for tracker

Browse source
This commit is contained in:
Vasilisk9812 2026-05-24 16:46:00 +09:00
parent ea61e157ab
commit 0000ac9f03
5 changed files with 4223 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

361
test_client/client.py Normal file
View file

@ -0,0 +1,361 @@
"""
Telemetry test client for stratoflights.
Reads a trajectory file and sends one packet every N seconds to the station
WebSocket endpoint: ws://<host>/api/ws/station/<satellite_id>/telemetry/?token=<token>
Supported trajectory formats
-----------------------------
JSON array (default):
[
{"lat": 62.1, "lon": 129.4, "alt": 500.0, "timestamp": 1716000000},
...
]
CSV (first row = header):
lat,lon,alt,timestamp
62.1,129.4,500.0,1716000000
Balloon log (.log):
Supports both "beacon" and "tracking" log variants produced by the
on-board firmware. Each record has a bracketed position header, followed
by angle-bracket telemetry lines. Tracking logs may also embed a Python-
style dict with already-decoded coordinates (used as a cross-check).
Header: [1;<lat_raw>;<lon_raw>;<alt_m>;<frame_id>;]
Telemetry: <1;<seq>;<mode>;<alt>;<...more fields...>;>
Optional: {'lat': 62.1, 'lon': 129.4, 'alt': 500} (tracking only)
Coordinate decoding:
lat = lat_raw / 1_000_000
lon = lon_raw / 1_000_000
alt = alt_m (metres, integer)
Records with no GPS fix (lat_raw == 0 and lon_raw == 0) are skipped.
Exact duplicate positions (same lat / lon / alt triple) are removed.
Usage
-----
python client.py --server ws://localhost:8000 \\
--satellite <uuid> \\
--token <auth-token> \\
--file trajectory.json \\
--interval 10
# Balloon log files:
python client.py --satellite <uuid> --token <token> --file beacon-20250406.log
python client.py --satellite <uuid> --token <token> --file tracking-20250406-02.log
# Generate a sample trajectory file then run:
python client.py --generate sample.json
python client.py --satellite <uuid> --token <token> --file sample.json
"""
import argparse
import asyncio
import csv
import json
import math
import re
import sys
import time
from pathlib import Path
# ---------------------------------------------------------------------------
# Trajectory loaders
# ---------------------------------------------------------------------------
def load_json(path: Path) -> list[dict]:
data = json.loads(path.read_text())
if not isinstance(data, list):
# Support Tawhiri-style nested format
if "prediction" in data:
points = []
for stage in data["prediction"]:
for p in stage.get("trajectory", []):
points.append({
"lat": p["latitude"],
"lon": p["longitude"],
"alt": p["altitude"],
"timestamp": p.get("time", int(time.time())),
})
return points
raise ValueError("JSON file must be an array or a Tawhiri prediction object")
return data
def load_csv(path: Path) -> list[dict]:
points = []
with path.open(newline="") as f:
reader = csv.DictReader(f)
for row in reader:
points.append({
"lat": float(row["lat"]),
"lon": float(row["lon"]),
"alt": float(row["alt"]),
"timestamp": int(row.get("timestamp", time.time())),
})
return points
# Matches the position header line: [1;<lat_raw>;<lon_raw>;<alt_m>;<frame_id>;]
_HEADER_RE = re.compile(r'^\[1;(-?\d+);(-?\d+);(-?\d+);(\d+);]')
# Matches the start of a telemetry line to extract the sequence number
_TELEM_RE = re.compile(r'^<1;(\d+);')
def load_log(path: Path) -> list[dict]:
"""Parse beacon or tracking .log files into a trajectory point list.
Duplicate positions (identical lat/lon/alt) are silently dropped.
Points with no GPS fix (lat_raw == lon_raw == 0) are also dropped.
"""
base_ts = int(time.time())
points: list[dict] = []
seen: set[tuple] = set()
lines = path.read_text(errors="replace").splitlines()
i = 0
while i < len(lines):
m = _HEADER_RE.match(lines[i].strip())
if not m:
i += 1
continue
lat_raw = int(m.group(1))
lon_raw = int(m.group(2))
alt_raw = int(m.group(3))
# No GPS fix yet
if lat_raw == 0 and lon_raw == 0:
i += 1
continue
lat = round(lat_raw / 1_000_000, 6)
lon = round(lon_raw / 1_000_000, 6)
alt = float(alt_raw)
# Drop exact-position duplicates
key = (lat, lon, alt_raw)
if key in seen:
i += 1
continue
seen.add(key)
# Grab the sequence number from the telemetry line that follows
seq = len(points)
for j in range(i + 1, min(i + 5, len(lines))):
tm = _TELEM_RE.match(lines[j].strip())
if tm:
seq = int(tm.group(1))
break
points.append({
"lat": lat,
"lon": lon,
"alt": alt,
"timestamp": base_ts + seq,
})
i += 1
return points
def load_trajectory(path: Path) -> list[dict]:
suffix = path.suffix.lower()
if suffix == ".csv":
return load_csv(path)
if suffix == ".log":
return load_log(path)
return load_json(path)
def deduplicate(points: list[dict]) -> list[dict]:
"""Remove points with identical (lat, lon, alt), preserving order."""
seen: set[tuple] = set()
result = []
for p in points:
key = (round(float(p.get("lat", 0)), 6),
round(float(p.get("lon", 0)), 6),
round(float(p.get("alt", 0)), 1))
if key not in seen:
seen.add(key)
result.append(p)
return result
# ---------------------------------------------------------------------------
# Sample generator
# ---------------------------------------------------------------------------
def generate_sample(path: Path, n: int = 50) -> None:
"""Generate a simple ascending balloon trajectory."""
base_lat, base_lon = 62.0, 129.5
base_ts = int(time.time())
points = []
for i in range(n):
angle = i * 0.05
points.append({
"lat": round(base_lat + math.sin(angle) * 0.02, 6),
"lon": round(base_lon + i * 0.003, 6),
"alt": round(500.0 + i * 200.0, 1),
"timestamp": base_ts + i * 10,
})
path.write_text(json.dumps(points, indent=2))
print(f"Sample trajectory written to {path} ({n} points)")
# ---------------------------------------------------------------------------
# WebSocket sender
# ---------------------------------------------------------------------------
async def run_ws(server: str, satellite: str, token: str,
points: list[dict], interval: float, loop: bool) -> None:
try:
import websockets
except ImportError:
sys.exit("websockets package not found — run: pip install websockets")
url = f"{server.rstrip('/')}/api/ws/station/{satellite}/telemetry/?token={token}"
print(f"Connecting to {url}")
iteration = 0
while True:
try:
async with websockets.connect(url) as ws:
print("Connected.")
for i, point in enumerate(points):
packet = {
"lat": float(point.get("lat", 0)),
"lon": float(point.get("lon", 0)),
"alt": float(point.get("alt", 0)),
"timestamp": int(point.get("timestamp", time.time())),
"payload": point.get("payload", {}),
"raw_data": point.get("raw_data", {}),
}
await ws.send(json.dumps(packet))
print(
f"[{i+1}/{len(points)}] "
f"lat={packet['lat']:.5f} "
f"lon={packet['lon']:.5f} "
f"alt={packet['alt']:.1f} m"
)
try:
reply = await asyncio.wait_for(ws.recv(), timeout=2.0)
data = json.loads(reply)
if "error" in data:
print(f" Server error: {data['error']}")
except asyncio.TimeoutError:
pass # server only responds on errors
if i < len(points) - 1:
await asyncio.sleep(interval)
print("All points sent.")
iteration += 1
if not loop:
break
print(f"Looping (iteration {iteration+1})...")
await asyncio.sleep(interval)
except Exception as e:
print(f"Connection error: {e}. Retrying in 5s...")
await asyncio.sleep(5)
if not loop:
break
# ---------------------------------------------------------------------------
# REST sender (fallback, no auth needed per stratoflights AllowAny policy)
# ---------------------------------------------------------------------------
async def run_rest(server: str, satellite: str,
points: list[dict], interval: float, loop: bool) -> None:
try:
import aiohttp
except ImportError:
sys.exit("aiohttp package not found — run: pip install aiohttp")
url = f"{server.rstrip('/')}/api/{satellite}/telemetry/"
print(f"Sending {len(points)} points to {url} in order, no delay...")
async with aiohttp.ClientSession() as session:
while True:
for i, point in enumerate(points):
packet = {
"lat": float(point.get("lat", 0)),
"lon": float(point.get("lon", 0)),
"alt": float(point.get("alt", 0)),
"payload": point.get("payload", {}),
}
async with session.post(url, json=packet) as resp:
print(
f"[{i+1}/{len(points)}] "
f"lat={packet['lat']:.5f} lon={packet['lon']:.5f} "
f"alt={packet['alt']:.1f} m → HTTP {resp.status}"
)
print("All points sent.")
if not loop:
break
print("Looping...")
await asyncio.sleep(interval)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main() -> None:
parser = argparse.ArgumentParser(description="Stratoflights telemetry test client")
parser.add_argument("--server", default="ws://localhost:8000",
help="Base server URL (default: ws://localhost:8000)")
parser.add_argument("--satellite", help="Satellite UUID")
parser.add_argument("--token", help="Auth token (required for WebSocket mode)")
parser.add_argument("--file", help="Trajectory file (.json, .csv, or .log)")
parser.add_argument("--interval", type=float, default=5.0,
help="Seconds between packets (default: 10)")
parser.add_argument("--mode", choices=["ws", "rest"], default="ws",
help="Transport: ws (WebSocket, default) or rest (HTTP POST)")
parser.add_argument("--loop", action="store_true",
help="Replay the trajectory in a loop indefinitely")
parser.add_argument("--generate", metavar="FILE",
help="Generate a sample trajectory JSON and exit")
args = parser.parse_args()
if args.generate:
generate_sample(Path(args.generate))
return
if not args.satellite:
parser.error("--satellite is required")
if not args.file:
parser.error("--file is required")
if args.mode == "ws" and not args.token:
parser.error("--token is required for WebSocket mode")
path = Path(args.file)
if not path.exists():
sys.exit(f"File not found: {path}")
points = load_trajectory(path)
if not points:
sys.exit(f"No valid points found in {path}")
before = len(points)
points = deduplicate(points)
removed = before - len(points)
print(f"Loaded {before} points from {path}" +
(f", removed {removed} duplicates → {len(points)} unique" if removed else f" ({len(points)} points)"))
if args.mode == "ws":
# Swap http(s) → ws(s) if the user passed an HTTP URL
server = args.server.replace("http://", "ws://").replace("https://", "wss://")
asyncio.run(run_ws(server, args.satellite, args.token, points, args.interval, args.loop))
else:
server = args.server.replace("ws://", "http://").replace("wss://", "https://")
asyncio.run(run_rest(server, args.satellite, points, args.interval, args.loop))
if __name__ == "__main__":
main()