Skip to content

Instantly share code, notes, and snippets.

@frank-engel

frank-engel/full-motion-video.ipynb

Created July 31, 2025 12:33
Show Gist options
  • Save frank-engel/979eaa6f89bae87ac192397e13932977 to your computer and use it in GitHub Desktop.
Save frank-engel/979eaa6f89bae87ac192397e13932977 to your computer and use it in GitHub Desktop.
Download ZIP
Display the source blob
Display the rendered blob
Raw
full-motion-video.ipynb
{
"cells": [
{
"cell_type": "markdown",
"id": "5d7e31ad-eeed-4250-8f8d-7be2e2a0abe4",
"metadata": {},
"source": [
"# Parsing Full Motion Video from a TS media container\n",
"The SkyDio X10 drone can save Full Motion Video (FMV). FMV adopts the [MISP-2023.2](https://nsgreg.nga.mil/misb.jsp) standard. See the \"DOC\" associated with the standards for the full specifications. \n",
"\n",
"## Extract the FML data\n",
"The FMV data is located in an embedded KLV stream. It needs to be extracted from the TS container before we can look at it."
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "35296e48-add9-4bd0-85df-f0e71cf986c5",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"STDOUT:\n",
" \n",
"STDERR:\n",
" ffmpeg version 2025-05-12-git-8ce32a7cbb-essentials_build-www.gyan.dev Copyright (c) 2000-2025 the FFmpeg developers\n",
" built with gcc 15.1.0 (Rev2, Built by MSYS2 project)\n",
" configuration: --enable-gpl --enable-version3 --enable-static --disable-w32threads --disable-autodetect --enable-fontconfig --enable-iconv --enable-gnutls --enable-libxml2 --enable-gmp --enable-bzlib --enable-lzma --enable-zlib --enable-libsrt --enable-libssh --enable-libzmq --enable-avisynth --enable-sdl2 --enable-libwebp --enable-libx264 --enable-libx265 --enable-libxvid --enable-libaom --enable-libopenjpeg --enable-libvpx --enable-mediafoundation --enable-libass --enable-libfreetype --enable-libfribidi --enable-libharfbuzz --enable-libvidstab --enable-libvmaf --enable-libzimg --enable-amf --enable-cuda-llvm --enable-cuvid --enable-dxva2 --enable-d3d11va --enable-d3d12va --enable-ffnvcodec --enable-libvpl --enable-nvdec --enable-nvenc --enable-vaapi --enable-libgme --enable-libopenmpt --enable-libopencore-amrwb --enable-libmp3lame --enable-libtheora --enable-libvo-amrwbenc --enable-libgsm --enable-libopencore-amrnb --enable-libopus --enable-libspeex --enable-libvorbis --enable-librubberband\n",
" libavutil 60. 2.100 / 60. 2.100\n",
" libavcodec 62. 3.101 / 62. 3.101\n",
" libavformat 62. 0.102 / 62. 0.102\n",
" libavdevice 62. 0.100 / 62. 0.100\n",
" libavfilter 11. 0.100 / 11. 0.100\n",
" libswscale 9. 0.100 / 9. 0.100\n",
" libswresample 6. 0.100 / 6. 0.100\n",
"Input #0, mpegts, from 'S1005865.TS':\n",
" Duration: 00:01:09.97, start: 3600.000000, bitrate: 76909 kb/s\n",
" Program 1 \n",
" Stream #0:0[0x41]: Video: h264 (Main) (HDMV / 0x564D4448), yuv420p(progressive), 3840x2160, 30 tbr, 90k tbn, Start 3600.000000\n",
" Stream #0:1[0x42]: Data: klv (KLVA / 0x41564C4B), Start 3600.000000\n",
"Stream mapping:\n",
" Stream #0:1 -> #0:0 (copy)\n",
"Output #0, data, to 'klv_stream.bin':\n",
" Metadata:\n",
" encoder : Lavf62.0.102\n",
" Stream #0:0: Data: klv (KLVA / 0x41564C4B)\n",
"Press [q] to stop, [?] for help\n",
"size= 177KiB time=00:00:25.50 bitrate= 56.9kbits/s speed=48.3x elapsed=0:00:00.52 \n",
"size= 256KiB time=00:00:49.08 bitrate= 42.7kbits/s speed=47.1x elapsed=0:00:01.04 \n",
"[out#0/data @ 000002512767ab00] video:0KiB audio:0KiB subtitle:0KiB other streams:486KiB global headers:0KiB muxing overhead: 0.000000%\n",
"size= 486KiB time=00:01:09.93 bitrate= 57.0kbits/s speed=45.4x elapsed=0:00:01.53 \n",
"\n"
]
}
],
"source": [
"import subprocess\n",
"\n",
"def extract_klv(ts_path, output_bin):\n",
" result = subprocess.run([\n",
" \"C:/REPOS/openai/projects/full_motion_video/ffmpeg.exe\", \"-y\",\n",
" \"-i\", ts_path,\n",
" \"-map\", \"0:1\", # KLV stream\n",
" \"-codec\", \"copy\",\n",
" \"-f\", \"data\", # Specify muxer for raw stream\n",
" output_bin\n",
" ], capture_output=True, text=True)\n",
"\n",
" print(\"STDOUT:\\n\", result.stdout)\n",
" print(\"STDERR:\\n\", result.stderr)\n",
"\n",
" if result.returncode != 0:\n",
" raise RuntimeError(f\"ffmpeg failed with exit code {result.returncode}\")\n",
"\n",
"extract_klv(\"S1005865.TS\", \"klv_stream.bin\")"
]
},
{
"cell_type": "markdown",
"id": "0327ec72-1501-4781-ab1c-c666a4e128be",
"metadata": {},
"source": [
"## Override the `klvdata` linear map method\n",
"The `klvdata` model expects numeric data to look a certain way, and doesn't gracefully manage failures. This monkey patch overwrites the `safe_linear_map` function so that it doesn't crash the code. \n",
"\n",
"```python\n",
"klvdata.common.linear_map = safe_linear_map\n",
"```\n",
"sets the patch."
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "5a3d94d8-930d-40dd-b406-726745c77baf",
"metadata": {},
"outputs": [],
"source": [
"import klvdata\n",
"import klvdata.common\n",
"from collections.abc import Mapping\n",
"from collections import OrderedDict\n",
"\n",
"def safe_linear_map(src_value, src_domain, dst_range):\n",
" if None in src_domain or None in dst_range:\n",
" print(f\"⚠️ Skipped invalid mapping — domain: {src_domain}, range: {dst_range}\")\n",
" return src_value # return original value instead of None\n",
"\n",
" src_min, src_max = src_domain\n",
" dst_min, dst_max = dst_range\n",
"\n",
" if not isinstance(src_value, (int, float)):\n",
" print(f\"⚠️ Unexpected type: {type(src_value)} for value {src_value}\")\n",
" return src_value\n",
"\n",
" if not (src_min <= src_value <= src_max):\n",
" if src_value != -2147483648:\n",
" print(f\"⚠️ Warning: value {src_value} out of domain {src_domain}\")\n",
" return src_value\n",
"\n",
" slope = (dst_max - dst_min) / (src_max - src_min)\n",
" dst_value = slope * (src_value - src_min) + dst_min\n",
" return dst_value\n",
"\n",
"klvdata.common.linear_map = safe_linear_map"
]
},
{
"cell_type": "markdown",
"id": "4d1ef874-ba7b-4fe5-9dcd-1fb3a20f55b2",
"metadata": {},
"source": [
"## Parse the KLV data\n",
"Now, we can look into each variable contained in the KLV stream. This stream provides one object per video frame. \n",
"\n",
"Let's look at the first packet (i.e., the first frame):\n"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "34b42fa6-0447-41c4-9520-4f86b3de841c",
"metadata": {
"scrolled": true
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"<class 'klvdata.misb0601.UASLocalMetadataSet'>\n",
"b'\\x02': PrecisionTimeStamp: (b'\\x02', 8, 2025-06-14 14:43:28.124917+00:00)\n",
"b'\\r': SensorLatitude: (b'\\r', 4, 29.11049092147057)\n",
"b'\\x0e': SensorLongitude: (b'\\x0e', 4, -98.17406311546176)\n",
"b'\\x0f': SensorTrueAltitude: (b'\\x0f', 2, 173.72243839169892)\n",
"b'8': PlatformGroundSpeed: (b'8', 1, 0.0)\n",
"b'O': SensorNorthVelocity: (b'O', 2, -0.029938657795923973)\n",
"b'P': SensorEastVelocity: (b'P', 2, 0.0)\n",
"b'p': UnknownElement: (b'p', 2, b'-\\xf6')\n",
"b'{': UnknownElement: (b'{', 1, b'\\x1d')\n",
"b'\\x05': PlatformHeadingAngle: (b'\\x05', 2, 315.87273975738157)\n",
"b'Z': PlatformPitchAngleFull: (b'Z', 4, 7.084221368229123)\n",
"b'\\x06': PlatformPitchAngle: (b'\\x06', 2, 7.083956419568469)\n",
"b'[': PlatformRollAngleFull: (b'[', 4, -3.6307849751928813)\n",
"b'\\x07': PlatformRollAngle: (b'\\x07', 2, -3.6301767021698694)\n",
"b'\\x12': SensorRelativeAzimuthAngle: (b'\\x12', 4, 0.40979753257934876)\n",
"b'\\x13': SensorRelativeElevationAngle: (b'\\x13', 4, -95.2494663909308)\n",
"b'\\x14': SensorRelativeRollAngle: (b'\\x14', 4, 3.1756615459862307)\n",
"b'\\x17': FrameCenterLatitude: (b'\\x17', 4, -2147483648)\n",
"b'\\x18': FrameCenterLongitude: (b'\\x18', 4, -2147483648)\n",
"b'\\x03': MissionID: (b'\\x03', 8, 4068E416)\n",
"b'\\n': PlatformDesignation: (b'\\n', 10, Skydio X10)\n",
"b';': PlatformCallSign: (b';', 14, SkydioX10-b6es)\n",
"b'\\x0c': ImageCoordinateSystem: (b'\\x0c', 14, Geodetic WGS84)\n",
"b'R': CornerLatitudePoint1Full: (b'R', 4, -2147483648)\n",
"b'S': CornerLongitudePoint1Full: (b'S', 4, -2147483648)\n",
"b'T': CornerLatitudePoint2Full: (b'T', 4, -2147483648)\n",
"b'U': CornerLongitudePoint2Full: (b'U', 4, -2147483648)\n",
"b'V': CornerLatitudePoint3Full: (b'V', 4, -2147483648)\n",
"b'W': CornerLongitudePoint3Full: (b'W', 4, -2147483648)\n",
"b'X': CornerLatitudePoint4Full: (b'X', 4, -2147483648)\n",
"b'Y': CornerLongitudePoint4Full: (b'Y', 4, -2147483648)\n",
"b'\\x10': SensorHorizontalFieldOfView: (b'\\x10', 2, 77.4328221560998)\n",
"b'\\x11': SensorVerticalFieldOfView: (b'\\x11', 2, 48.82398718242161)\n",
"b'\\x0b': ImageSourceSensor: (b'\\x0b', 2, EO)\n",
"b'A': UASLSVersionNumber: (b'A', 1, 19.0)\n",
"b'\\x01': Checksum: (b'\\x01', 2, 0xE74F)\n"
]
}
],
"source": [
"packet = next(klvdata.StreamParser(open(\"klv_stream.bin\", \"rb\")))\n",
"print(type(packet))\n",
"\n",
"for key, val in packet.items.items():\n",
" print(f\"{key}: {val}\")"
]
},
{
"cell_type": "markdown",
"id": "a012117c-b192-4693-b606-e864144eea69",
"metadata": {},
"source": [
"## Compute the location of the frame\n",
"Now we can take the IMU and GPS data from the X10 embedded in the KLV stream to compute the location of the corners of the frame image. This is mostly correct, but I'm not correctly dealing with the rotation and the mapping from lat/lon decimal degrees to meters/feet. "
]
},
{
"cell_type": "code",
"execution_count": 59,
"id": "babb1259-a61a-488b-b2ce-2d0fd22054e8",
"metadata": {},
"outputs": [],
"source": [
"import numpy as np\n",
"\n",
"def estimate_fov_corners(packet):\n",
" # Helper to safely extract and convert MappedValue or int/float\n",
" def get_value(key):\n",
" return float(packet.items[key].value)\n",
"\n",
" try:\n",
" lat = get_value(b'\\r') # SensorLatitude\n",
" lon = get_value(b'\\x0e') # SensorLongitude\n",
" alt = get_value(b'\\x0f') # SensorTrueAltitude (meters)\n",
" hfov = get_value(b'\\x10') # Horizontal FOV (degrees)\n",
" vfov = get_value(b'\\x11') # Vertical FOV (degrees)\n",
" heading = get_value(b'\\x05') # PlatformHeadingAngle (degrees)\n",
" pitch = get_value(b'\\x06') # PlatformPitchAngle (degrees)\n",
" roll = get_value(b'\\x07') # PlatformRollAngle (degrees)\n",
" except Exception as e:\n",
" print(f\"Failed to extract telemetry: {e}\")\n",
" return []\n",
"\n",
" # Convert to radians\n",
" hfov_rad = np.radians(hfov)\n",
" vfov_rad = np.radians(vfov)\n",
" heading_rad = np.radians(heading)\n",
" pitch_rad = np.radians(pitch)\n",
" roll_rad = np.radians(roll)\n",
"\n",
" # Ground footprint under nadir assumption\n",
" ground_width = 2 * alt * np.tan(hfov_rad / 2)\n",
" ground_height = 2 * alt * np.tan(vfov_rad / 2)\n",
"\n",
" #### Fix orientation by swapping horizontal and vertical\n",
" ground_width = 2 * alt * np.tan(vfov_rad / 2)\n",
" ground_height = 2 * alt * np.tan(hfov_rad / 2)\n",
"\n",
" # Corner offsets in sensor/body frame\n",
" local_corners = np.array([\n",
" [-ground_width / 2, ground_height / 2, 0], # top-left\n",
" [ ground_width / 2, ground_height / 2, 0], # top-right\n",
" [ ground_width / 2, -ground_height / 2, 0], # bottom-right\n",
" [-ground_width / 2, -ground_height / 2, 0], # bottom-left\n",
" ])\n",
"\n",
" # Rotation matrices\n",
" R_roll = np.array([\n",
" [1, 0, 0],\n",
" [0, np.cos(roll_rad), -np.sin(roll_rad)],\n",
" [0, np.sin(roll_rad), np.cos(roll_rad)]\n",
" ])\n",
"\n",
" R_pitch = np.array([\n",
" [np.cos(pitch_rad), 0, np.sin(pitch_rad)],\n",
" [0, 1, 0],\n",
" [-np.sin(pitch_rad), 0, np.cos(pitch_rad)]\n",
" ])\n",
"\n",
" R_heading = np.array([\n",
" [np.cos(heading_rad), -np.sin(heading_rad), 0],\n",
" [np.sin(heading_rad), np.cos(heading_rad), 0],\n",
" [0, 0, 1]\n",
" ])\n",
"\n",
" # Combined rotation: heading * pitch * roll\n",
" R = R_heading @ R_pitch @ R_roll\n",
"\n",
" rotated_corners = (R @ local_corners.T).T\n",
"\n",
" # Project to lat/lon (assuming flat terrain at altitude)\n",
" corners = []\n",
" for corner in rotated_corners:\n",
" dx, dy = corner[0], corner[1]\n",
" dlat = dy / 111320\n",
" dlon = dx / (111320 * np.cos(np.radians(lat)))\n",
" corners.append((lat + dlat, lon + dlon))\n",
"\n",
" return corners"
]
},
{
"cell_type": "code",
"execution_count": 60,
"id": "ef4ae8d8-ef30-41e2-84bf-f5787d3750a5",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Corner 1: lat=29.111883, lon=-98.173654\n",
"Corner 2: lat=29.110904, lon=-98.172499\n",
"Corner 3: lat=29.109099, lon=-98.174473\n",
"Corner 4: lat=29.110077, lon=-98.175628\n"
]
}
],
"source": [
"packet = next(klvdata.StreamParser(open(\"klv_stream.bin\", \"rb\")))\n",
"corners = estimate_fov_corners(packet)\n",
"for i, (lat, lon) in enumerate(corners, 1):\n",
" print(f\"Corner {i}: lat={lat:.6f}, lon={lon:.6f}\")"
]
},
{
"cell_type": "markdown",
"id": "d3bda7c6-4abf-4d1d-b545-dc9b57effdac",
"metadata": {},
"source": [
"## Example batch processing\n",
"```python\n",
"def parse_klv(klv_file):\n",
" with open(klv_file, \"rb\") as f:\n",
" for i, packet in enumerate(klvdata.StreamParser(f), start=1): \n",
" try:\n",
" if isinstance(packet, klvdata.misb0601.UASLocalMetadataSet):\n",
" print(\"is Mapping\")\n",
" for key, val in packet.items.items():\n",
" print(f\" {key}: {val}\")\n",
" else:\n",
" print(f\"Skipped non-mapping packet of type: {type(packet).__name__}\")\n",
" except Exception as e:\n",
" print(f\"Failed to parse packet: {e}\")\n",
"\n",
"parse_klv(\"klv_stream.bin\")\n",
"```"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.7"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment