mirror of
https://github.com/mytechnotalent/Embedded-Hacking.git
synced 2026-10-01 05:30:15 +02:00
Course update: lessons, CTF 0x0011a_cb, and documentation
- 0x0011a_cb (Operation Dark Vector): nation-state CTF redesign with an AES-128-ECB sealed target and a plaintext launch origin; RP2350 firmware with bearing-driven servo, tri-color LEDs, GSV stats, and a realistic no-fix path - docs: story-driven classified brief, GDB and Ghidra tutorials with deep step-throughs, regenerated artifacts and PDFs - scripts: docstring standard, AES per-student randomizer, telemetry monitor - week 3 to week 5 lessons: Ghidra patching tutorial, CMSIS-SVD hardware RE, double floating-point and GPIO architecture chapters, README structure
This commit is contained in:
1 parent
5201ee4b6b
commit
35eacd2c0e
162 files changed
+125658
-232
No files matched your search
@@ -0,0 +1,411 @@
|
||||
#!/usr/bin/env python3
|
||||
# MIT License
|
||||
#
|
||||
# Copyright (c) 2026 Kevin Thomas
|
||||
#
|
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
# of this software and associated documentation files (the "Software"), to deal
|
||||
# in the Software without restriction, including without limitation the rights
|
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
# copies of the Software, and to permit persons to whom the Software is
|
||||
# furnished to do so, subject to the following conditions:
|
||||
#
|
||||
# The above copyright notice and this permission notice shall be included in all
|
||||
# copies or substantial portions of the Software.
|
||||
#
|
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
# SOFTWARE.
|
||||
#
|
||||
# Author: Kevin Thomas
|
||||
# Email: kevin@mytechnotalent.com
|
||||
# GitHub: https://github.com/mytechnotalent
|
||||
# File: decode_coordinates.py
|
||||
# Desc: Decode and patch RP2350 micro-UAV navigation coordinates.
|
||||
# Created: 2026
|
||||
|
||||
"""
|
||||
Decode and patch RP2350 micro-UAV navigation coordinates.
|
||||
|
||||
Analyzes raw firmware images for IEEE 754 64-bit double-precision floating
|
||||
point coordinates. Scans target flight vectors and patches binaries with
|
||||
safe disposal coordinates in the Atlantic Ocean.
|
||||
"""
|
||||
|
||||
import math
|
||||
import struct
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
FLASH_BASE = 0x10000000
|
||||
ORIGIN_LAT = 38.840280
|
||||
ORIGIN_LON = -77.428890
|
||||
ATLANTIC_LAT = 37.000000
|
||||
ATLANTIC_LON = -74.000000
|
||||
|
||||
|
||||
def _haversine_calc(phi1: float, phi2: float,
|
||||
dphi: float, dlam: float) -> float:
|
||||
"""
|
||||
Compute central angle using haversine formula.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
phi1 : float
|
||||
Origin latitude in radians.
|
||||
phi2 : float
|
||||
Target latitude in radians.
|
||||
dphi : float
|
||||
Latitude difference in radians.
|
||||
dlam : float
|
||||
Longitude difference in radians.
|
||||
|
||||
Returns
|
||||
-------
|
||||
float
|
||||
Central angular distance in radians.
|
||||
"""
|
||||
s_phi = math.sin(dphi / 2.0) ** 2
|
||||
s_lam = math.sin(dlam / 2.0) ** 2
|
||||
a = s_phi + math.cos(phi1) * math.cos(phi2) * s_lam
|
||||
return 2.0 * math.atan2(math.sqrt(a), math.sqrt(1.0 - a))
|
||||
|
||||
|
||||
def haversine(lat1: float, lon1: float,
|
||||
lat2: float, lon2: float) -> tuple[float, float]:
|
||||
"""
|
||||
Compute Great-Circle distance and azimuth bearing.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
lat1 : float
|
||||
Origin latitude in degrees.
|
||||
lon1 : float
|
||||
Origin longitude in degrees.
|
||||
lat2 : float
|
||||
Destination latitude in degrees.
|
||||
lon2 : float
|
||||
Destination longitude in degrees.
|
||||
|
||||
Returns
|
||||
-------
|
||||
tuple[float, float]
|
||||
Distance in statute miles and bearing in degrees.
|
||||
"""
|
||||
p1, p2 = math.radians(lat1), math.radians(lat2)
|
||||
dl = math.radians(lon2 - lon1)
|
||||
dist = 6371.0 * _haversine_calc(p1, p2, math.radians(lat2 - lat1), dl)
|
||||
y = math.sin(dl) * math.cos(p2)
|
||||
term = math.sin(p1) * math.cos(p2) * math.cos(dl)
|
||||
x = math.cos(p1) * math.sin(p2) - term
|
||||
bearing = (math.degrees(math.atan2(y, x)) + 360.0) % 360.0
|
||||
return dist * 0.621371, bearing
|
||||
|
||||
|
||||
def _is_coord(val: float) -> bool:
|
||||
"""
|
||||
Verify if float falls within target geographic bounds.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
val : float
|
||||
Candidate double-precision value.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True if value is a valid latitude or longitude.
|
||||
"""
|
||||
if math.isnan(val) or math.isinf(val):
|
||||
return False
|
||||
in_lat = 35.0 <= val <= 41.0
|
||||
in_lon = -79.0 <= val <= -72.0
|
||||
return in_lat or in_lon
|
||||
|
||||
|
||||
def _parse_chunk(data: bytes, off: int) -> dict | None:
|
||||
"""
|
||||
Extract and validate one 8-byte candidate float.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
data : bytes
|
||||
Firmware image buffer.
|
||||
off : int
|
||||
Byte offset inside image buffer.
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict | None
|
||||
Parsed coordinate record or None.
|
||||
"""
|
||||
chunk = data[off:off + 8]
|
||||
val = struct.unpack("<d", chunk)[0]
|
||||
if not _is_coord(val):
|
||||
return None
|
||||
hex_str = " ".join(f"{b:02x}" for b in chunk)
|
||||
u64 = struct.unpack("<Q", chunk)[0]
|
||||
kind = "LATITUDE" if val > 0.0 else "LONGITUDE"
|
||||
return {"off": off, "addr": FLASH_BASE + off, "val": val,
|
||||
"hex": hex_str, "u64": u64, "kind": kind}
|
||||
|
||||
|
||||
def scan_coordinates(data: bytes) -> list[dict]:
|
||||
"""
|
||||
Scan firmware buffer for double-precision coordinates.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
data : bytes
|
||||
Firmware image buffer.
|
||||
|
||||
Returns
|
||||
-------
|
||||
list[dict]
|
||||
List of candidate coordinate records.
|
||||
"""
|
||||
found = []
|
||||
limit = len(data) - 8
|
||||
for off in range(0, limit, 4):
|
||||
item = _parse_chunk(data, off)
|
||||
if item is not None:
|
||||
found.append(item)
|
||||
return found
|
||||
|
||||
|
||||
def _print_banner(path: Path) -> None:
|
||||
"""
|
||||
Print operation heading and recovery origin.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
path : pathlib.Path
|
||||
Target firmware path.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
print("=" * 67)
|
||||
print(" OPERATION DARK VECTOR // FORENSIC COORDINATE TOOL")
|
||||
print(" GMU Rapid Hardware Exploitation Laboratory - Fairfax, VA")
|
||||
print("=" * 67)
|
||||
print(f"[*] Target Binary: {path.name} ({path.stat().st_size:,} bytes)")
|
||||
print(f"[*] Recovery Origin: Centreville, VA "
|
||||
f"({ORIGIN_LAT:.6f}, {ORIGIN_LON:.6f})")
|
||||
print("-" * 67)
|
||||
|
||||
|
||||
def _print_candidate(c: dict) -> None:
|
||||
"""
|
||||
Print formatted candidate coordinate entry.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
c : dict
|
||||
Candidate coordinate record.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
print(f" [{c['kind']:9s}] Value: {c['val']:12.6f} | "
|
||||
f"Addr: 0x{c['addr']:08x} (Offset: 0x{c['off']:04x})")
|
||||
print(f" Hex: {c['hex']} | uint64: 0x{c['u64']:016x}")
|
||||
|
||||
|
||||
def _cardinal_bearing(brg: float) -> str:
|
||||
"""Return compass direction string for given bearing."""
|
||||
dirs = ["N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE",
|
||||
"S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW"]
|
||||
idx = int((brg + 11.25) / 22.5) % 16
|
||||
return dirs[idx]
|
||||
|
||||
|
||||
def _print_summary(lat: float, lon: float, mi: float, brg: float) -> None:
|
||||
"""
|
||||
Print mission tactical assessment summary.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
lat : float
|
||||
Decoded target latitude.
|
||||
lon : float
|
||||
Decoded target longitude.
|
||||
mi : float
|
||||
Distance in statute miles.
|
||||
brg : float
|
||||
Initial bearing in degrees.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
card = _cardinal_bearing(brg)
|
||||
print("\n" + "=" * 67)
|
||||
print(" TACTICAL MISSION PROFILE DECODED")
|
||||
print("=" * 67)
|
||||
print(f" Target Latitude: {lat:.6f} deg N")
|
||||
print(f" Target Longitude: {lon:.6f} deg W")
|
||||
print(f" Distance from Origin: {mi:.2f} miles ({mi * 1.60934:.2f} km)")
|
||||
print(f" Flight Vector Bearing: {brg:.1f} deg ({card})")
|
||||
print("=" * 67)
|
||||
|
||||
|
||||
def _patch_bytes(data: bytes, old_lat: float, old_lon: float) -> bytes:
|
||||
"""
|
||||
Replace target coordinates with safe Atlantic Ocean coordinates.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
data : bytes
|
||||
Original firmware bytes.
|
||||
old_lat : float
|
||||
Original target latitude.
|
||||
old_lon : float
|
||||
Original target longitude.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bytes
|
||||
Patched firmware byte buffer.
|
||||
"""
|
||||
src_lat = struct.pack("<d", old_lat)
|
||||
src_lon = struct.pack("<d", old_lon)
|
||||
dst_lat = struct.pack("<d", ATLANTIC_LAT)
|
||||
dst_lon = struct.pack("<d", ATLANTIC_LON)
|
||||
buf = data.replace(src_lat, dst_lat)
|
||||
return buf.replace(src_lon, dst_lon)
|
||||
|
||||
|
||||
def _print_patch_info(out_name: str, mi: float, brg: float) -> None:
|
||||
"""
|
||||
Display confirmation of applied firmware patch.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
out_name : str
|
||||
Patched output file name.
|
||||
mi : float
|
||||
Distance to safe disposal zone.
|
||||
brg : float
|
||||
Azimuth bearing to disposal zone.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
print(f"\n[+] Patched firmware written to: {out_name}")
|
||||
print("[+] Overwrote waypoint -> Atlantic Ocean Disposal Zone:")
|
||||
print(f" Safe Latitude: {ATLANTIC_LAT:.6f} deg N")
|
||||
print(f" Safe Longitude: {ATLANTIC_LON:.6f} deg W")
|
||||
print(f" Offshore Distance: {mi:.2f} miles (Bearing: {brg:.1f} deg)")
|
||||
|
||||
|
||||
def patch_firmware(target: Path, out_path: Path,
|
||||
lat: float, lon: float) -> None:
|
||||
"""
|
||||
Patch firmware with safe Atlantic Ocean disposal waypoint.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
target : pathlib.Path
|
||||
Source binary path.
|
||||
out_path : pathlib.Path
|
||||
Destination patched binary path.
|
||||
lat : float
|
||||
Current target latitude.
|
||||
lon : float
|
||||
Current target longitude.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
raw = target.read_bytes()
|
||||
patched = _patch_bytes(raw, lat, lon)
|
||||
out_path.write_bytes(patched)
|
||||
mi, brg = haversine(ORIGIN_LAT, ORIGIN_LON, ATLANTIC_LAT, ATLANTIC_LON)
|
||||
_print_patch_info(out_path.name, mi, brg)
|
||||
|
||||
|
||||
def _evaluate(items: list[dict], target: Path, do_patch: bool) -> None:
|
||||
"""
|
||||
Display results and execute patch if requested.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
items : list[dict]
|
||||
Found coordinate records.
|
||||
target : pathlib.Path
|
||||
Target binary path.
|
||||
do_patch : bool
|
||||
Flag indicating if patch should be applied.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
lats = [c for c in items if c["kind"] == "LATITUDE"]
|
||||
lons = [c for c in items if c["kind"] == "LONGITUDE"]
|
||||
if not (lats and lons):
|
||||
return
|
||||
t_lat, t_lon = lats[-1]["val"], lons[-1]["val"]
|
||||
mi, brg = haversine(ORIGIN_LAT, ORIGIN_LON, t_lat, t_lon)
|
||||
_print_summary(t_lat, t_lon, mi, brg)
|
||||
if do_patch:
|
||||
out = target.parent / f"{target.stem}_patched.bin"
|
||||
patch_firmware(target, out, t_lat, t_lon)
|
||||
|
||||
|
||||
def _parse_args(args: list[str]) -> tuple[Path, bool]:
|
||||
"""
|
||||
Parse command line arguments for target path and patch flag.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
args : list[str]
|
||||
Command line arguments.
|
||||
|
||||
Returns
|
||||
-------
|
||||
tuple[pathlib.Path, bool]
|
||||
Target binary path and patch flag.
|
||||
"""
|
||||
do_patch = "--patch" in args
|
||||
paths = [p for p in args if not p.startswith("--")]
|
||||
target = Path(paths[0]) if paths else Path("0x0011a_cb.bin")
|
||||
return target, do_patch
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""
|
||||
Execute firmware coordinate extraction and optional patching.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
None
|
||||
|
||||
Returns
|
||||
-------
|
||||
int
|
||||
Zero on success, non-zero on failure.
|
||||
"""
|
||||
target, do_patch = _parse_args(sys.argv[1:])
|
||||
if not target.exists():
|
||||
print(f"[-] Error: '{target}' not found.")
|
||||
return 1
|
||||
_print_banner(target)
|
||||
items = scan_coordinates(target.read_bytes())
|
||||
for item in items:
|
||||
_print_candidate(item)
|
||||
_evaluate(items, target, do_patch)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,202 @@
|
||||
#!/usr/bin/env python3
|
||||
# MIT License
|
||||
#
|
||||
# Copyright (c) 2026 Kevin Thomas
|
||||
#
|
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
# of this software and associated documentation files (the "Software"), to deal
|
||||
# in the Software without restriction, including without limitation the rights
|
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
# copies of the Software, and to permit persons to whom the Software is
|
||||
# furnished to do so, subject to the following conditions:
|
||||
#
|
||||
# The above copyright notice and this permission notice shall be included in all
|
||||
# copies or substantial portions of the Software.
|
||||
#
|
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
# SOFTWARE.
|
||||
#
|
||||
# Author: Kevin Thomas
|
||||
# Email: kevin@mytechnotalent.com
|
||||
# GitHub: https://github.com/mytechnotalent
|
||||
# File: float_hex_converter.py
|
||||
# Desc: Convert and explain IEEE 754 float/hex operations step-by-step.
|
||||
# Created: 2026
|
||||
|
||||
"""Convert and explain IEEE 754 float/hex operations step-by-step."""
|
||||
|
||||
import argparse
|
||||
import struct
|
||||
import sys
|
||||
|
||||
|
||||
def _exp_str(e_val: int, bias: int, is_64: bool) -> str:
|
||||
"""
|
||||
Get accurate true exponent string accounting for IEEE 754 edge cases.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
e_val : int
|
||||
The stored exponent value.
|
||||
bias : int
|
||||
The exponent bias (127 or 1023).
|
||||
is_64 : bool
|
||||
True if 64-bit precision.
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
The formatted true exponent explanation string.
|
||||
"""
|
||||
if e_val == 0:
|
||||
return f"0 (Zero/Subnormal, True Exp: {1 - bias})"
|
||||
if e_val == (2047 if is_64 else 255):
|
||||
return f"{e_val} (Inf/NaN flag)"
|
||||
return f"{e_val} - {bias} = {e_val - bias}"
|
||||
|
||||
|
||||
def _print_encode_steps(val: float, b: int) -> None:
|
||||
"""
|
||||
Print the math steps for encoding a float to hex.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
val : float
|
||||
The float value to encode.
|
||||
b : int
|
||||
The bit size (32 or 64).
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
f1, f2, bias = ("<Q", "<d", 1023) if b == 64 else ("<I", "<f", 127)
|
||||
bstr = f"{struct.unpack(f1, struct.pack(f2, val))[0]:0{b}b}"
|
||||
s, e, m = (
|
||||
bstr[0],
|
||||
bstr[1 : 1 + (11 if b == 64 else 8)],
|
||||
bstr[1 + (11 if b == 64 else 8) :],
|
||||
)
|
||||
hex_val = f"0x{int(bstr, 2):0{b//4}X}"
|
||||
print(f"\n[ENCODE {val} to {b}-bit]\n1. Sign: {s} (0=Pos, 1=Neg)")
|
||||
print(f"2. Exp: {_exp_str(int(e, 2), bias, b == 64)}\n3. Mantissa: {m}")
|
||||
print(f"4. Full: {s} {e} {m}\n5. Hex: {hex_val}")
|
||||
if b == 64:
|
||||
raw_hex = f"{int(bstr, 2):016X}"
|
||||
r3 = f"0x{raw_hex[:8]}"
|
||||
r2 = f"0x{raw_hex[8:]}"
|
||||
print(f"6. ARM Regs: r3 (high) = {r3}, r2 (low) = {r2} (e.g. in printf)")
|
||||
else:
|
||||
print(f"6. ARM Reg: Single 32-bit register ({hex_val})")
|
||||
|
||||
|
||||
def _print_decode_steps(hex_str: str) -> None:
|
||||
"""
|
||||
Print the math steps for decoding a hex string to float.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
hex_str : str
|
||||
The hex string to decode.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
c = hex_str.lower()
|
||||
if c.startswith("0x"):
|
||||
c = c[2:]
|
||||
b, f1, f2, bias = (64, "<Q", "<d", 1023) if len(c) > 8 else (32, "<I", "<f", 127)
|
||||
bstr = f"{int(c, 16):0{b}b}"
|
||||
s, e, m = (
|
||||
bstr[0],
|
||||
bstr[1 : 1 + (11 if b == 64 else 8)],
|
||||
bstr[1 + (11 if b == 64 else 8) :],
|
||||
)
|
||||
print(f"\n[DECODE 0x{c.zfill(b//4).upper()} ({b}-bit)]\n1. Binary: {s} {e} {m}")
|
||||
print(f"2. Sign: {s}\n3. Exp: {_exp_str(int(e, 2), bias, b == 64)}")
|
||||
print(f"4. Value: {struct.unpack(f2, struct.pack(f1, int(c, 16)))[0]}")
|
||||
if b == 64:
|
||||
raw_hex = c.zfill(16).upper()
|
||||
r3 = f"0x{raw_hex[:8]}"
|
||||
r2 = f"0x{raw_hex[8:]}"
|
||||
print(f"5. ARM Regs: r3 (high) = {r3}, r2 (low) = {r2} (e.g. in printf)")
|
||||
else:
|
||||
print(f"5. ARM Reg: Single 32-bit register (0x{c.zfill(8).upper()})")
|
||||
|
||||
|
||||
def _is_hex(s: str) -> bool:
|
||||
"""
|
||||
Check if a string is a hexadecimal representation.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
s : str
|
||||
Candidate string, with or without a 0x prefix.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True if every character is a hexadecimal digit.
|
||||
"""
|
||||
cleaned = s.lower()
|
||||
if cleaned.startswith("0x"):
|
||||
cleaned = cleaned[2:]
|
||||
if not cleaned:
|
||||
return False
|
||||
return all(c in "0123456789abcdef" for c in cleaned)
|
||||
|
||||
|
||||
def _process_conversion(val_str: str) -> None:
|
||||
"""
|
||||
Execute the conversion and print the output.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
val_str : str
|
||||
The raw input string to process.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
cleaned = val_str.strip()
|
||||
if cleaned.lower().startswith("0x") or (len(cleaned) >= 8 and _is_hex(cleaned)):
|
||||
_print_decode_steps(cleaned)
|
||||
else:
|
||||
val = float(cleaned)
|
||||
_print_encode_steps(val, 32)
|
||||
_print_encode_steps(val, 64)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""
|
||||
Execute the conversion pipeline based on CLI arguments.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
None
|
||||
|
||||
Returns
|
||||
-------
|
||||
int
|
||||
Zero on successful conversion, otherwise non-zero.
|
||||
"""
|
||||
parser = argparse.ArgumentParser(description="Float/Hex step converter.")
|
||||
parser.add_argument("val", help="Hex (0x...) or Float value to convert.")
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
_process_conversion(args.val)
|
||||
return 0
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Executable
+135
@@ -0,0 +1,135 @@
|
||||
#!/usr/bin/env python3
|
||||
# MIT License
|
||||
#
|
||||
# Copyright (c) 2026 Kevin Thomas
|
||||
#
|
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
# of this software and associated documentation files (the "Software"), to deal
|
||||
# in the Software without restriction, including without limitation the rights
|
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
# copies of the Software, and to permit persons to whom the Software is
|
||||
# furnished to do so, subject to the following conditions:
|
||||
#
|
||||
# The above copyright notice and this permission notice shall be included in all
|
||||
# copies or substantial portions of the Software.
|
||||
#
|
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
# SOFTWARE.
|
||||
#
|
||||
# Author: Kevin Thomas
|
||||
# Email: kevin@mytechnotalent.com
|
||||
# GitHub: https://github.com/mytechnotalent
|
||||
# File: lora_console.py
|
||||
# Desc: Interactive REYAX RYLR998 terminal for the FT232RL ground station.
|
||||
# Created: 2026
|
||||
|
||||
"""
|
||||
Interactive LoRa terminal for the REYAX RYLR998 ground station.
|
||||
|
||||
Sends AT commands as complete bursts terminated with a carriage return and line
|
||||
feed to avoid the RYLR998 inter-character timeout error, and prints incoming
|
||||
packets from the airborne node as they arrive.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
try:
|
||||
import serial
|
||||
except ImportError:
|
||||
serial = None
|
||||
|
||||
|
||||
def _reader_thread(ser: "serial.Serial") -> None:
|
||||
"""
|
||||
Continuously read and display incoming packets from the radio.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
ser : serial.Serial
|
||||
Open serial connection to the ground RYLR998.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
while True:
|
||||
try:
|
||||
line = ser.readline().decode("utf-8", errors="ignore").strip()
|
||||
if line:
|
||||
print(f"\n[LORA RX] {line}\n> ", end="", flush=True)
|
||||
except Exception:
|
||||
break
|
||||
|
||||
|
||||
def _parse_args() -> argparse.Namespace:
|
||||
"""
|
||||
Parse command line arguments for the LoRa console.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
None
|
||||
|
||||
Returns
|
||||
-------
|
||||
argparse.Namespace
|
||||
Parsed arguments with the serial port and baud rate.
|
||||
"""
|
||||
parser = argparse.ArgumentParser(description="REYAX RYLR998 console")
|
||||
parser.add_argument("--port", default="/dev/cu.usbserial-A50285BI")
|
||||
parser.add_argument("--baud", type=int, default=115200)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""
|
||||
Open the ground station radio and forward operator input.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
None
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
if serial is None:
|
||||
sys.exit("pyserial required: pip install pyserial")
|
||||
args = _parse_args()
|
||||
|
||||
print(f"[*] Opening REYAX RYLR998 on {args.port} @ {args.baud}...")
|
||||
try:
|
||||
ser = serial.Serial(args.port, args.baud, timeout=0.5)
|
||||
except Exception as e:
|
||||
sys.exit(f"[-] Failed to open {args.port}: {e}")
|
||||
|
||||
thread = threading.Thread(target=_reader_thread, args=(ser,), daemon=True)
|
||||
thread.start()
|
||||
|
||||
time.sleep(0.1)
|
||||
ser.write(b"AT\r\n")
|
||||
|
||||
print("[+] Connected. Type AT commands (e.g. AT, AT+BAND?, AT+NETWORKID?).")
|
||||
print("[+] Incoming airborne packets print as [LORA RX] +RCV=...")
|
||||
print("[+] Press Ctrl-C or Ctrl-D to exit.\n")
|
||||
|
||||
try:
|
||||
while True:
|
||||
cmd = input("> ").strip()
|
||||
if not cmd:
|
||||
continue
|
||||
ser.write(cmd.encode("utf-8") + b"\r\n")
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
print("\n[*] Exiting LoRa console.")
|
||||
ser.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,234 @@
|
||||
#!/usr/bin/env python3
|
||||
# MIT License
|
||||
#
|
||||
# Copyright (c) 2026 Kevin Thomas
|
||||
#
|
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
# of this software and associated documentation files (the "Software"), to deal
|
||||
# in the Software without restriction, including without limitation the rights
|
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
# copies of the Software, and to permit persons to whom the Software is
|
||||
# furnished to do so, subject to the following conditions:
|
||||
#
|
||||
# The above copyright notice and this permission notice shall be included in all
|
||||
# copies or substantial portions of the Software.
|
||||
#
|
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
# SOFTWARE.
|
||||
#
|
||||
# Author: Kevin Thomas
|
||||
# Email: kevin@mytechnotalent.com
|
||||
# GitHub: https://github.com/mytechnotalent
|
||||
# File: randomize_build.py
|
||||
# Desc: Builds a per-student 0x0011a_cb image with an AES-encrypted target.
|
||||
# Created: 2026
|
||||
|
||||
"""
|
||||
Per-student randomized CTF build.
|
||||
|
||||
Jitters the target waypoint, AES-128-ECB encrypts it under a per-build key, and
|
||||
writes include/ctf_target.h so the firmware rebuilds it at boot. Every image
|
||||
carries a different ciphertext and the plaintext target exists nowhere in flash.
|
||||
An offline answer key or a memorized value is useless; the waypoint can only be
|
||||
recovered from the artifact and confirmed on hardware.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import pathlib
|
||||
import random
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
BASE_LAT = 38.881940
|
||||
BASE_LON = -77.450280
|
||||
JITTER_DEG = 0.01
|
||||
UF2_FAMILY = "0xe48bff59"
|
||||
FLASH_BASE = "0x10000000"
|
||||
|
||||
|
||||
def _parse_args() -> argparse.Namespace:
|
||||
"""
|
||||
Parse command line arguments for the randomized build.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
None
|
||||
|
||||
Returns
|
||||
-------
|
||||
argparse.Namespace
|
||||
Parsed arguments with seed, build dir, student id, and uf2 flag.
|
||||
"""
|
||||
here = pathlib.Path(__file__).resolve().parent
|
||||
parser = argparse.ArgumentParser(description="Randomized CTF build")
|
||||
parser.add_argument("--seed", type=int, default=None)
|
||||
parser.add_argument("--build-dir", default=str(here.parent / "build-ctf"))
|
||||
parser.add_argument("--student-id", default="student")
|
||||
parser.add_argument("--uf2", action="store_true")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def _make_target(seed: int) -> tuple[float, float]:
|
||||
"""
|
||||
Generate a jittered target waypoint around the base coordinates.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
seed : int
|
||||
Deterministic seed for the jitter.
|
||||
|
||||
Returns
|
||||
-------
|
||||
tuple[float, float]
|
||||
Jittered latitude and longitude, rounded to six decimals.
|
||||
"""
|
||||
rng = random.Random(seed)
|
||||
lat = round(BASE_LAT + rng.uniform(-JITTER_DEG, JITTER_DEG), 6)
|
||||
lon = round(BASE_LON + rng.uniform(-JITTER_DEG, JITTER_DEG), 6)
|
||||
return lat, lon
|
||||
|
||||
|
||||
def _make_key(seed: int) -> bytes:
|
||||
"""
|
||||
Derive a deterministic 16-byte AES key for the build.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
seed : int
|
||||
Build seed from which the key is derived.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bytes
|
||||
Sixteen byte AES-128 key.
|
||||
"""
|
||||
krng = random.Random(seed ^ 0x5EED)
|
||||
return bytes(krng.randrange(256) for _ in range(16))
|
||||
|
||||
|
||||
def _aes_ecb(plain: bytes, key: bytes) -> bytes:
|
||||
"""
|
||||
Encrypt one block with AES-128-ECB.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
plain : bytes
|
||||
Sixteen byte plaintext block.
|
||||
key : bytes
|
||||
Sixteen byte AES key.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bytes
|
||||
Sixteen byte ciphertext block.
|
||||
"""
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
enc = Cipher(algorithms.AES(key), modes.ECB(), backend=default_backend()).encryptor()
|
||||
return enc.update(plain) + enc.finalize()
|
||||
|
||||
|
||||
def _write_header(path: pathlib.Path, key: bytes, ct: bytes) -> None:
|
||||
"""
|
||||
Write the AES key and ciphertext header consumed by the firmware.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
path : pathlib.Path
|
||||
Destination header path.
|
||||
key : bytes
|
||||
Sixteen byte AES key.
|
||||
ct : bytes
|
||||
Sixteen byte ciphertext block.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
path.write_text(
|
||||
"#ifndef CTF_TARGET_H\n#define CTF_TARGET_H\n\n#include <stdint.h>\n\n"
|
||||
"#define CTF_AES_KEY { %s }\n"
|
||||
"#define CTF_TARGET_CT { %s }\n\n"
|
||||
"#endif // CTF_TARGET_H\n"
|
||||
% (", ".join(f"0x{b:02X}" for b in key), ", ".join(f"0x{b:02X}" for b in ct)))
|
||||
|
||||
|
||||
def _run(cmd: list[str], cwd: pathlib.Path) -> None:
|
||||
"""
|
||||
Run an external command and raise on failure.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
cmd : list[str]
|
||||
Command and arguments to execute.
|
||||
cwd : pathlib.Path
|
||||
Working directory for the command.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
subprocess.run(cmd, cwd=str(cwd), check=True)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""
|
||||
Build a per-student image with an AES-encrypted, randomized target.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
None
|
||||
|
||||
Returns
|
||||
-------
|
||||
int
|
||||
Zero on success.
|
||||
"""
|
||||
args = _parse_args()
|
||||
root = pathlib.Path(__file__).resolve().parent.parent
|
||||
seed = args.seed if args.seed is not None else random.randrange(2**31)
|
||||
lat, lon = _make_target(seed)
|
||||
key = _make_key(seed)
|
||||
pt = struct.pack("<d", lat) + struct.pack("<d", lon)
|
||||
ct = _aes_ecb(pt, key)
|
||||
|
||||
_write_header(root / "include" / "ctf_target.h", key, ct)
|
||||
|
||||
build = pathlib.Path(args.build_dir).resolve()
|
||||
_run(["cmake", "-S", str(root), "-B", str(build)], root)
|
||||
_run(["cmake", "--build", str(build)], root)
|
||||
|
||||
image = build / "0x0011a_cb.bin"
|
||||
if args.uf2:
|
||||
out = build / f"0x0011a_cb_{args.student_id}.uf2"
|
||||
_run([sys.executable, str(root / "uf2conv.py"), str(image),
|
||||
"-f", UF2_FAMILY, "-b", FLASH_BASE, "-c", "-o", str(out)], root)
|
||||
|
||||
key_out = {"student_id": args.student_id, "seed": seed,
|
||||
"target_lat": lat, "target_lon": lon,
|
||||
"aes_key_hex": key.hex(), "ciphertext_hex": ct.hex(),
|
||||
"image": str(image)}
|
||||
keydir = root / "scratch"
|
||||
keydir.mkdir(exist_ok=True)
|
||||
keyfile = keydir / f"answer_{args.student_id}.json"
|
||||
keyfile.write_text(json.dumps(key_out, indent=2) + "\n")
|
||||
|
||||
print(f"[+] student_id : {args.student_id}")
|
||||
print(f"[+] TARGET_LAT : {lat}")
|
||||
print(f"[+] TARGET_LON : {lon}")
|
||||
print(f"[+] AES key : {key.hex()}")
|
||||
print(f"[+] ciphertext : {ct.hex()}")
|
||||
print(f"[+] image : {image}")
|
||||
print(f"[+] answer key : {keyfile} (INSTRUCTOR ONLY, do not ship)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,302 @@
|
||||
#!/usr/bin/env python3
|
||||
# MIT License
|
||||
#
|
||||
# Copyright (c) 2026 Kevin Thomas
|
||||
#
|
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
# of this software and associated documentation files (the "Software"), to deal
|
||||
# in the Software without restriction, including without limitation the rights
|
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
# copies of the Software, and to permit persons to whom the Software is
|
||||
# furnished to do so, subject to the following conditions:
|
||||
#
|
||||
# The above copyright notice and this permission notice shall be included in all
|
||||
# copies or substantial portions of the Software.
|
||||
#
|
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
# SOFTWARE.
|
||||
#
|
||||
# Author: Kevin Thomas
|
||||
# Email: kevin@mytechnotalent.com
|
||||
# GitHub: https://github.com/mytechnotalent
|
||||
# File: telemetry_monitor.py
|
||||
# Desc: Real-time telemetry monitor for Operation Dark Vector.
|
||||
# Created: 2026
|
||||
|
||||
"""
|
||||
Real-time telemetry monitor for Operation Dark Vector.
|
||||
|
||||
Reads live avionics and GPS telemetry from the FT232RL USB-to-UART ground
|
||||
station bridge, parses coordinates and distance, and prints a mission status
|
||||
dashboard to the terminal.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
|
||||
try:
|
||||
import serial
|
||||
except ImportError:
|
||||
serial = None
|
||||
|
||||
|
||||
def _format_coord(val: float, pos_c: str, neg_c: str) -> str:
|
||||
"""
|
||||
Format a decimal coordinate with cardinal direction.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
val : float
|
||||
Coordinate value in degrees.
|
||||
pos_c : str
|
||||
Cardinal letter for positive values.
|
||||
neg_c : str
|
||||
Cardinal letter for negative values.
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
Formatted coordinate string.
|
||||
"""
|
||||
card = pos_c if val >= 0.0 else neg_c
|
||||
return f"{abs(val):.6f} deg {card}"
|
||||
|
||||
|
||||
def _format_pos(lat: float, lon: float) -> str:
|
||||
"""
|
||||
Format combined latitude and longitude string.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
lat : float
|
||||
Latitude coordinate.
|
||||
lon : float
|
||||
Longitude coordinate.
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
Formatted dual coordinate string.
|
||||
"""
|
||||
lat_s = _format_coord(lat, 'N', 'S')
|
||||
lon_s = _format_coord(lon, 'E', 'W')
|
||||
return f"{lat_s} {lon_s}"
|
||||
|
||||
|
||||
def _format_hud_rows(
|
||||
cur_lat: float,
|
||||
cur_lon: float,
|
||||
is_rel: bool,
|
||||
has_lock: bool = False
|
||||
) -> list[str]:
|
||||
"""
|
||||
Format HUD data rows.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
cur_lat : float
|
||||
Current UAV latitude.
|
||||
cur_lon : float
|
||||
Current UAV longitude.
|
||||
is_rel : bool
|
||||
Whether payload has released.
|
||||
has_lock : bool
|
||||
Whether active GNSS 3D lock has been acquired.
|
||||
|
||||
Returns
|
||||
-------
|
||||
list[str]
|
||||
List of formatted box rows.
|
||||
"""
|
||||
if cur_lat == 0.0 and cur_lon == 0.0:
|
||||
c_s = "0.000000 deg N 0.000000 deg E"
|
||||
g_s = "SEARCHING SATELLITES"
|
||||
m_s = "MOTOR STOPPED [WAITING FOR 3D LOCK]"
|
||||
else:
|
||||
c_s = _format_pos(cur_lat, cur_lon)
|
||||
g_s = "ACTIVE 3D LOCK" if has_lock else "SEARCHING SATELLITES"
|
||||
m_s = "ACTIVE PROPULSION [SERVO SPINNING]" if has_lock else "MOTOR STOPPED [WAITING FOR 3D LOCK]"
|
||||
|
||||
return [
|
||||
f"| CURRENT POSITION : {c_s:<44} |",
|
||||
f"| GNSS SUBSYSTEM : {g_s:<44} |",
|
||||
f"| PROPULSION MOTOR : {m_s:<44} |"
|
||||
]
|
||||
|
||||
|
||||
def _print_box(title: str, link: str, rows: list[str]) -> None:
|
||||
"""
|
||||
Print framed ASCII box.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
title : str
|
||||
Box title line.
|
||||
link : str
|
||||
Sub-header line.
|
||||
rows : list[str]
|
||||
Body content rows.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
hdr = f"+{'-' * 65}+"
|
||||
print(hdr)
|
||||
print(title)
|
||||
print(link)
|
||||
print(hdr)
|
||||
for r in rows:
|
||||
print(r)
|
||||
print(hdr + "\n", flush=True)
|
||||
|
||||
|
||||
def _print_hud(
|
||||
cur_lat: float,
|
||||
cur_lon: float,
|
||||
is_released: bool,
|
||||
has_lock: bool = False
|
||||
) -> None:
|
||||
"""
|
||||
Display the 67-character telemetry mission HUD.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
cur_lat : float
|
||||
Current UAV latitude.
|
||||
cur_lon : float
|
||||
Current UAV longitude.
|
||||
is_released : bool
|
||||
Whether payload solenoid has been energized.
|
||||
has_lock : bool
|
||||
Whether active GNSS 3D lock has been acquired.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
rows = _format_hud_rows(cur_lat, cur_lon, is_released, has_lock)
|
||||
title = f"|{'DARK VECTOR TELEMETRY CONSOLE':^65}|"
|
||||
status = f"{'STATUS: ONLINE':>20}"
|
||||
link = f"| LINK: FT232RL / RYLR998 LORA GROUND STATION{status} |"
|
||||
_print_box(title, link, rows)
|
||||
|
||||
|
||||
def _run_demo() -> None:
|
||||
"""
|
||||
Execute simulation of drone telemetry stream.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
None
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
print("[*] Running simulated ground station telemetry stream...\n")
|
||||
_print_hud(0.0, 0.0, False, False)
|
||||
time.sleep(1.0)
|
||||
_print_hud(38.840280, -77.428890, False, True)
|
||||
time.sleep(1.0)
|
||||
_print_hud(38.861110, -77.439585, False, True)
|
||||
time.sleep(1.0)
|
||||
_print_hud(38.881940, -77.450280, True, True)
|
||||
|
||||
|
||||
def _process_line(
|
||||
line: str,
|
||||
coords: dict[str, any]
|
||||
) -> bool:
|
||||
"""
|
||||
Parse a single line of serial telemetry using regex.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
line : str
|
||||
Raw serial string.
|
||||
coords : dict[str, any]
|
||||
State dictionary of coordinates.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True if telemetry data was updated, False otherwise.
|
||||
"""
|
||||
updated = False
|
||||
m_cur = re.search(r"CURRENT LAT:\s*([-+]?\d*\.?\d+).*?LON:\s*([-+]?\d*\.?\d+)", line)
|
||||
if m_cur:
|
||||
coords["cur_lat"] = float(m_cur.group(1))
|
||||
coords["cur_lon"] = float(m_cur.group(2))
|
||||
coords["has_lock"] = True
|
||||
updated = True
|
||||
if "PAYLOAD RELEASED" in line:
|
||||
coords["released"] = 1.0
|
||||
updated = True
|
||||
return updated
|
||||
|
||||
|
||||
def _monitor_serial(port: str, baud: int) -> None:
|
||||
"""
|
||||
Monitor serial stream from FT232RL ground station.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
port : str
|
||||
Serial device path.
|
||||
baud : int
|
||||
Baud rate.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
if serial is None:
|
||||
sys.exit("pyserial required: pip install pyserial")
|
||||
coords = {
|
||||
"cur_lat": 0.0,
|
||||
"cur_lon": 0.0,
|
||||
"released": 0.0,
|
||||
"has_lock": False
|
||||
}
|
||||
with serial.Serial(port, baud, timeout=1.0) as ser:
|
||||
while True:
|
||||
raw = ser.readline().decode("utf-8", errors="ignore").strip()
|
||||
if raw:
|
||||
_process_line(raw, coords)
|
||||
rel = coords["released"] > 0.5
|
||||
_print_hud(coords["cur_lat"], coords["cur_lon"], rel, coords["has_lock"])
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""
|
||||
Parse arguments and start telemetry monitor.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
None
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
"""
|
||||
parser = argparse.ArgumentParser(description="Dark Vector Telemetry")
|
||||
parser.add_argument("--port", default="/dev/tty.usbserial-0001")
|
||||
parser.add_argument("--baud", type=int, default=115200)
|
||||
parser.add_argument("--demo", action="store_true")
|
||||
args = parser.parse_args()
|
||||
if args.demo:
|
||||
_run_demo()
|
||||
return
|
||||
_monitor_serial(args.port, args.baud)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user