"""Generate the synthetic practice data for DRT 114 (Computer Programming). Everything produced here is SYNTHETIC: it comes from a simple flight model, not from a real aircraft. Run: python make_sample_data.py (needs: pip install pymavlink) Output: flight_log_sample.csv (onboard-style log, 2 Hz, with deliberate data-quality problems) telemetry_sample.tlog (ground-station MAVLink 2 telemetry, with an 8 s link outage) The random seed is fixed, so the files are identical on every run. Time is counted in whole milliseconds so that comparisons never suffer from floating-point drift. """ import csv import math import random import struct EARTH_R = 6_371_000.0 # metres, the same radius the lessons use for haversine HOME_LAT, HOME_LON = 14.1000, 100.6000 # an arbitrary synthetic launch point GROUND_AMSL_M = 4.0 # synthetic ground elevation above mean sea level CRUISE_MS = 9.0 CRUISE_ALT_M = 30.0 DT_MS = 50 # simulation step: 50 ms (20 Hz); CSV and telemetry are sampled from it DT = DT_MS / 1000 BATTERY_WH = 22.2 * 8.0 # 6S 8.0 Ah pack, 177.6 Wh (the pack used in the DRT 342 example) SEED = 114 # Deliberate problems for the exercises (milliseconds since the start of the log) ALT_SPIKE_MS = 120_000 # one barometer glitch: +40 m in a single 2 Hz sample BATT_MISSING_MS = (200_000, 200_500) # two samples with no battery voltage GPS_DIP_MS = (250_000, 255_000) # few satellites and a noisier position LINK_OUTAGE_MS = (150_000, 158_000) # nothing reaches the ground station def to_latlon(x_east, y_north): lat = HOME_LAT + math.degrees(y_north / EARTH_R) lon = HOME_LON + math.degrees(x_east / (EARTH_R * math.cos(math.radians(HOME_LAT)))) return lat, lon def build_legs(): """The flight as a list of legs: (mode, start_xyz, end_xyz), in metres east/north/up of the launch point.""" z = CRUISE_ALT_M legs = [("GUIDED", (0, 0, 0), (0, 0, z)), # take-off ("AUTO", (0, 0, z), (50, 50, z))] # transit to the survey block x, y = 50, 50 for k in range(8): # 8 lines of 300 m, 35 m apart x_end = 350 if k % 2 == 0 else 50 legs.append(("AUTO", (x, y, z), (x_end, y, z))) x = x_end if k < 7: legs.append(("AUTO", (x, y, z), (x, y + 35, z))) y += 35 legs.append(("RTL", (x, y, z), (0, 0, z))) # return at survey height legs.append(("LAND", (0, 0, z), (0, 0, 0))) # descend return legs def planned_horizontal_distance(legs): return sum(math.hypot(b[0] - a[0], b[1] - a[1]) for _, a, b in legs) def simulate(): """Step the aircraft along the legs. Returns the legs and a list of state dicts, one per DT_MS.""" legs = build_legs() states = [] def add(**kw): states.append(dict(t_ms=len(states) * DT_MS, **kw)) for _ in range(10_000 // DT_MS): # 10 s on the ground first add(x=0.0, y=0.0, z=0.0, v=0.0, climb=0.0, heading=90.0, mode="GUIDED", armed=False) for mode, a, b in legs: length = math.dist(a, b) s = 0.0 vertical = a[:2] == b[:2] while s < length - 1e-9: if vertical: # take-off 2.5 m/s, landing 2.0 m/s rate = 2.5 if b[2] > a[2] else 2.0 v_h, climb, step = 0.0, (rate if b[2] > a[2] else -rate), rate * DT else: # slow near waypoints, cruise in between v_h = max(CRUISE_MS - 5.5 * math.exp(-s / 8.0) - 5.5 * math.exp(-(length - s) / 8.0), 3.0) climb, step = 0.0, v_h * DT s = min(s + step, length) f = s / length dx, dy = b[0] - a[0], b[1] - a[1] heading = states[-1]["heading"] if vertical else math.degrees(math.atan2(dx, dy)) % 360.0 add(x=a[0] + dx * f, y=a[1] + dy * f, z=a[2] + (b[2] - a[2]) * f, v=v_h, climb=climb, heading=heading, mode=mode, armed=True) last = states[-1] for _ in range(2_000 // DT_MS): # 2 s disarmed at the end add(x=last["x"], y=last["y"], z=0.0, v=0.0, climb=0.0, heading=last["heading"], mode="LAND", armed=False) return legs, states def add_battery(states): """Integrate power to get the remaining charge, terminal voltage and current.""" used_wh = 0.0 for st in states: power = 260.0 + 8.0 * st["v"] + (120.0 if st["climb"] > 0 else 0.0) if st["armed"] else 25.0 used_wh += power * DT / 3600.0 soc = max(0.0, 1.0 - used_wh / BATTERY_WH) volts = 21.0 + 4.2 * soc - 0.0016 * power st.update(power=power, soc=soc, volts=volts, amps=power / volts) return states def write_csv(states, path): rng = random.Random(SEED) every = 500 // DT_MS # 2 Hz rows = [] for st in states[::every]: t_ms = st["t_ms"] dip = GPS_DIP_MS[0] <= t_ms < GPS_DIP_MS[1] sigma = 2.5 if dip else 0.3 lat, lon = to_latlon(st["x"] + rng.gauss(0, sigma), st["y"] + rng.gauss(0, sigma)) alt = st["z"] + rng.gauss(0, 0.15) + (40.0 if t_ms == ALT_SPIKE_MS else 0.0) speed = max(0.0, st["v"] + rng.gauss(0, 0.15)) if st["v"] > 0 else 0.0 rows.append([ f"{t_ms / 1000:.1f}", f"{lat:.6f}", f"{lon:.6f}", f"{alt:.2f}", f"{speed:.2f}", f"{st['heading']:.1f}", "" if t_ms in BATT_MISSING_MS else f"{st['volts'] + rng.gauss(0, 0.02):.2f}", f"{round(st['soc'] * 100)}", str(6 if dip else rng.choice([14, 15, 15, 16])), st["mode"], ]) with open(path, "w", newline="", encoding="utf-8") as f: w = csv.writer(f, lineterminator="\n") w.writerow(["t_s", "lat_deg", "lon_deg", "alt_m", "groundspeed_ms", "heading_deg", "batt_v", "batt_pct", "gps_sats", "mode"]) w.writerows(rows) return len(rows) class _TlogSink: """File-like object that writes an 8-byte big-endian timestamp (microseconds) before each packet.""" def __init__(self, f): self.f, self.ts_us = f, 0 def write(self, data): self.f.write(struct.pack(">Q", self.ts_us) + data) ARDUCOPTER_MODE = {"GUIDED": 4, "AUTO": 3, "RTL": 6, "LAND": 9} # ArduPilot Copter custom_mode numbers UTC_START_US = 1_773_455_400_000_000 # 2026-03-14 02:30:00 UTC (arbitrary, synthetic) BOOT_OFFSET_MS = 45_000 def write_tlog(states, path): from pymavlink.dialects.v20 import common as mav2 rng = random.Random(SEED + 1) counts, dropped = {}, [0] with open(path, "wb") as f: sink = _TlogSink(f) mav = mav2.MAVLink(sink, srcSystem=1, srcComponent=1) def send(st, msg): if LINK_OUTAGE_MS[0] <= st["t_ms"] < LINK_OUTAGE_MS[1]: msg.pack(mav) # the aircraft still sends, so the sequence number mav.seq = (mav.seq + 1) % 256 # advances, but the ground station records nothing dropped[0] += 1 return sink.ts_us = UTC_START_US + st["t_ms"] * 1000 mav.send(msg) counts[msg.get_type()] = counts.get(msg.get_type(), 0) + 1 for n, st in enumerate(states): boot_ms = BOOT_OFFSET_MS + st["t_ms"] if n % 20 == 0: # HEARTBEAT 1 Hz base = 1 | (128 if st["armed"] else 0) # custom-mode flag | armed flag state = mav2.MAV_STATE_ACTIVE if st["armed"] else mav2.MAV_STATE_STANDBY send(st, mav2.MAVLink_heartbeat_message(mav2.MAV_TYPE_QUADROTOR, mav2.MAV_AUTOPILOT_ARDUPILOTMEGA, base, ARDUCOPTER_MODE[st["mode"]], state, 3)) if n % 2 == 0: # ATTITUDE 10 Hz roll = 0.03 * math.sin(st["t_ms"] / 1000 * 0.7) + rng.gauss(0, 0.004) pitch = -0.13 * st["v"] / CRUISE_MS + rng.gauss(0, 0.004) yaw = math.radians(st["heading"] if st["heading"] <= 180 else st["heading"] - 360) send(st, mav2.MAVLink_attitude_message(boot_ms, roll, pitch, yaw, 0.0, 0.0, 0.0)) if n % 5 == 0: # GLOBAL_POSITION_INT 4 Hz lat, lon = to_latlon(st["x"] + rng.gauss(0, 0.3), st["y"] + rng.gauss(0, 0.3)) vx = int(st["v"] * math.cos(math.radians(90 - st["heading"])) * 100) vy = int(st["v"] * math.sin(math.radians(90 - st["heading"])) * 100) send(st, mav2.MAVLink_global_position_int_message( boot_ms, int(round(lat * 1e7)), int(round(lon * 1e7)), int((GROUND_AMSL_M + st["z"]) * 1000), int(st["z"] * 1000), vy, vx, int(-st["climb"] * 100), int(st["heading"] * 100) % 36000)) send(st, mav2.MAVLink_vfr_hud_message(st["v"], st["v"], int(st["heading"]) % 360, 60 if st["armed"] else 0, GROUND_AMSL_M + st["z"], st["climb"])) if n % 20 == 0: # SYS_STATUS 1 Hz send(st, mav2.MAVLink_sys_status_message(0, 0, 0, 350, int(st["volts"] * 1000), int(st["amps"] * 100), round(st["soc"] * 100), 0, 0, 0, 0, 0, 0)) return counts, dropped[0] def main(): legs, states = simulate() add_battery(states) rows = write_csv(states, "flight_log_sample.csv") counts, dropped = write_tlog(states, "telemetry_sample.tlog") print(f"planned horizontal distance: {planned_horizontal_distance(legs):.1f} m") print(f"duration: {states[-1]['t_ms'] / 1000:.1f} s, CSV rows: {rows}") print("tlog messages:", counts, "| sent but not recorded (outage):", dropped) if __name__ == "__main__": main()