A me interessano i dati RTK del drone. Per estrarre questi dati si deve primare estraree la keychain per individuare la posizione
#!/usr/bin/env python3
"""
dji_rtk_extractor.py
Estrae i dati RTK epoca-per-epoca (record binario tipo 45, "RTKDifferenceDataType")
dai flight log DJI in formato TXT (versione 13+, criptati AES-256-CBC), e li esporta
in un CSV con posizione rover (drone), posizione base station RTK, quota ellissoidica,
flag di qualita' del fix, e tempo di volo (fly_time) correlato dai record OSD adiacenti.
Il crate Rust ufficiale "dji-log-parser" non implementa questo tipo di record (lo
cattura come bytes grezzi non decodificati). Questo script reimplementa in Python
pura la pipeline di decodifica (XOR + AES-256-CBC) usata dal crate per ottenere il
contenuto in chiaro, poi interpreta la struttura byte-per-byte del record 45
(dedotta empiricamente e validata contro fly_time / geografia nota).
REQUISITI:
- una keychain JSON gia' ottenuta (vedi sotto), NON serve una API key DJI qui
dentro: la decrittazione AES e' un'operazione locale una volta che si hanno
le chiavi. La keychain si ottiene una sola volta per log, con una chiamata
all'API DJI (serve connettivita' verso i server DJI + una API key valida),
es. tramite il crate dji-log-parser stesso (DJILog::fetch_keychains) o via
il tool CLI ufficiale "dji-log". Una volta salvata come JSON, riusabile
offline per quello stesso file di log.
USO:
python3 dji_rtk_extractor.py FLIGHT_LOG.txt keychain.json -o output.csv
Se il log NON e' criptato (versione < 13), la keychain non serve:
python3 dji_rtk_extractor.py FLIGHT_LOG.txt -o output.csv
STRUTTURA DEL RECORD TIPO 45 (83 byte, dedotta e validata su un volo di test):
offset 0-10 : non identificato (11 byte, probabile satelliti/flag)
offset 11 : double -> latitudine rover (drone), gradi
offset 19 : double -> longitudine rover, gradi
offset 27 : float32 -> quota ellissoidica rover, metri
offset 31 : double -> latitudine base station RTK, gradi (costante nel volo)
offset 39 : double -> longitudine base station RTK, gradi (costante)
offset 47 : float32 -> quota ellissoidica base station, metri (costante)
offset 51 : float32 -> ipotesi: heading rover, gradi (0-360) - NON VALIDATO
offset 57 : float32 -> flag qualita' fix (1/2/3/4, verosimilmente
Fixed/Float/DGPS/Standalone)
offset 61-82 : non identificato (22 byte)
Questa struttura e' stata dedotta per un M400 con log versione 14; se noti valori
palesemente incoerenti (es. lat/lon fuori range) con un log di versione diversa,
gli offset potrebbero non essere validi 1:1 e vanno riverificati.
"""
import argparse
import base64
import csv
import json
import struct
import sys
from pathlib import Path
try:
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
except ImportError:
sys.exit(
"Manca pycryptodome. Installa con:\n"
" pip install pycryptodome --break-system-packages"
)
# ----------------------------------------------------------------------------
# CRC64 (variante "Jones", usata da DJI per derivare la chiave XOR per-record)
# ----------------------------------------------------------------------------
_CRC64_JONES_POLY_REFLECTED = 0x95AC9329AC4BC9B5
def crc64(crc: int, data: bytes) -> int:
for byte in data:
crc ^= byte
for _ in range(8):
if crc & 1:
crc = (crc >> 1) ^ _CRC64_JONES_POLY_REFLECTED
else:
crc >>= 1
return crc
# ----------------------------------------------------------------------------
# XOR layer: ogni record e' prima offuscato con un keystream a 8 byte derivato
# dal primo byte del payload e dal record_type, poi (per i record non-plaintext)
# il risultato XOR e' il ciphertext AES-256-CBC vero e proprio.
# ----------------------------------------------------------------------------
_MAGIC = 0x123456789ABCDEF0
_MASK64 = (1 << 64) - 1
class XorKeystream:
def __init__(self, first_byte: int, record_type: int):
seed = (first_byte + record_type) & 0xFF
magic_mul = (_MAGIC * first_byte) & _MASK64
magic_bytes = magic_mul.to_bytes(8, "little")
key64 = crc64(seed, magic_bytes)
self.key = key64.to_bytes(8, "little")
def decode(self, data: bytes) -> bytes:
return bytes(b ^ self.key[i % 8] for i, b in enumerate(data))
# ----------------------------------------------------------------------------
# Mappatura record_type -> "feature point" (famiglia di chiavi nella keychain)
# ----------------------------------------------------------------------------
def feature_point(record_type: int, version: int) -> str:
def v13(base, other):
return base if version == 13 else other
version_dependent = {
3: ("BaseFeature", "GimbalFeature"),
4: ("BaseFeature", "RCFeature"),
7: ("BaseFeature", "BatteryFeature"),
8: ("BaseFeature", "BatteryFeature"),
11: ("BaseFeature", "RCFeature"),
22: ("AfterSalesFeature", "BatteryFeature"),
25: ("BaseFeature", "CameraFeature"),
28: ("AfterSalesFeature", "FlySafeFeature"),
29: ("BaseFeature", "RCFeature"),
33: ("BaseFeature", "RCFeature"),
51: ("AfterSalesFeature", "FlySafeFeature"),
52: ("AfterSalesFeature", "FlySafeFeature"),
53: ("AfterSalesFeature", "FlightHubFeature"),
}
fixed = {
1: "BaseFeature", 2: "BaseFeature", 5: "DJIFlyCustomFeature",
6: "BaseFeature", 9: "DJIFlyCustomFeature", 10: "DJIFlyCustomFeature",
12: "AfterSalesFeature", 13: "BaseFeature", 14: "BaseFeature",
15: "BaseFeature", 16: "AfterSalesFeature", 17: "VisionFeature",
18: "VisionFeature", 19: "AfterSalesFeature", 20: "DJIFlyCustomFeature",
21: "AgricultureFeature", 24: "DJIFlyCustomFeature", 26: "AfterSalesFeature",
27: "AfterSalesFeature", 30: "DJIFlyCustomFeature", 31: "WaypointFeature",
32: "WaypointFeature", 34: "WaypointFeature", 35: "WaypointFeature",
36: "WaypointFeature", 38: "WaypointFeature", 39: "WaypointFeature",
40: "BaseFeature", 41: "AgricultureFeature", 43: "AgricultureFeature",
44: "AgricultureFeature", 45: "AgricultureFeature", 46: "AgricultureFeature",
47: "AgricultureFeature", 48: "AgricultureFeature", 49: "AirLinkFeature",
50: "PlaintextFeature", 54: "DJIFlyCustomFeature", 55: "SecurityFeature",
56: "PlaintextFeature", 58: "BaseFeature", 59: "BaseFeature",
62: "RCFeature", 63: "BaseFeature",
}
if record_type in version_dependent:
base, other = version_dependent[record_type]
return v13(base, other)
return fixed.get(record_type, "PlaintextFeature")
_FEATURE_JSON_NAME = {
"BaseFeature": "FR_Standardization_Feature_Base_1",
"VisionFeature": "FR_Standardization_Feature_Vision_2",
"WaypointFeature": "FR_Standardization_Feature_Waypoint_3",
"AgricultureFeature": "FR_Standardization_Feature_Agriculture_4",
"AirLinkFeature": "FR_Standardization_Feature_AirLink_5",
"AfterSalesFeature": "FR_Standardization_Feature_AfterSales_6",
"DJIFlyCustomFeature": "FR_Standardization_Feature_DJIFlyCustom_7",
"PlaintextFeature": "FR_Standardization_Feature_Plaintext_8",
"FlightHubFeature": "FR_Standardization_Feature_FlightHub_9",
"GimbalFeature": "FR_Standardization_Feature_Gimbal_10",
"RCFeature": "FR_Standardization_Feature_RC_11",
"CameraFeature": "FR_Standardization_Feature_Camera_12",
"BatteryFeature": "FR_Standardization_Feature_Battery_13",
"FlySafeFeature": "FR_Standardization_Feature_FlySafe_14",
"SecurityFeature": "FR_Standardization_Feature_Security_15",
}
DJI_KEYCHAIN_ENDPOINT = "https://dev.dji.com/openapi/v1/flight-records/keychains"
_FEATURE_ID_TO_NAME = {
1: "BaseFeature", 2: "VisionFeature", 3: "WaypointFeature", 4: "AgricultureFeature",
5: "AirLinkFeature", 6: "AfterSalesFeature", 7: "DJIFlyCustomFeature", 8: "PlaintextFeature",
9: "FlightHubFeature", 10: "GimbalFeature", 11: "RCFeature", 12: "CameraFeature",
13: "BatteryFeature", 14: "FlySafeFeature", 15: "SecurityFeature",
}
def parse_auxiliary_metadata(raw: bytes):
"""Legge i blocchi Auxiliary::Info e Auxiliary::Version (a partire da offset 100,
valido per log versione >=12) per ottenere version/department richiesti dall'API DJI.
Ritorna (version_meta, department)."""
pos = 100 # PREFIX_SIZE
# Info block: magic(1) + length(2) + contenuto (length bytes, XOR'd, non ci serve)
magic0 = raw[pos]; pos += 1
if magic0 != 0:
raise ValueError(f"Atteso Auxiliary::Info (magic=0) a offset 100, trovato {magic0}")
len0 = struct.unpack_from("<H", raw, pos)[0]; pos += 2
pos += len0
# Version block: magic(1) + length(2) + version:u16 + department:u8 (+ padding)
magic1 = raw[pos]; pos += 1
if magic1 != 1:
raise ValueError(f"Atteso Auxiliary::Version (magic=1) a offset {pos-1}, trovato {magic1}")
len1 = struct.unpack_from("<H", raw, pos)[0]; pos += 2
version_meta = struct.unpack_from("<H", raw, pos)[0]
department = raw[pos + 2]
pos += len1
return version_meta, department, pos # pos = inizio atteso dei record (records_offset)
def extract_keystorage_entries(raw: bytes, records_offset: int, version: int):
"""Scansiona i record e ritorna [(feature_point_id, ciphertext_bytes), ...] dai
record tipo 56 (KeyStorage). Questi non sono AES-criptati (solo XOR), quindi non
serve nessuna keychain per leggerli: il loro contenuto E' l'input da mandare
all'API DJI per ottenere la keychain vera e propria."""
entries = []
for record_type, length_field, payload_start, payload in iter_records(raw, records_offset):
if record_type != 56:
continue
first_byte = payload[0]
xor = XorKeystream(first_byte, record_type)
content = xor.decode(payload[1:])
if len(content) < 4:
continue
feature_point_id = struct.unpack_from("<H", content, 0)[0]
data_length = struct.unpack_from("<H", content, 2)[0]
data = content[4:4 + data_length]
entries.append((feature_point_id, data))
return entries
def fetch_keychain_via_api(raw: bytes, records_offset: int, version: int, api_key: str,
endpoint: str = DJI_KEYCHAIN_ENDPOINT) -> dict:
"""Ricava automaticamente la keychain chiamando l'API DJI. Richiede connettivita'
di rete verso dev.dji.com e una API key valida. Ritorna un dict nello stesso
formato di load_keychain() (feature_name -> [iv_bytes, key_bytes]), e salva anche
la risposta raw in formato compatibile con keychains_cache.json per riuso offline."""
import urllib.request
import urllib.error
version_meta, department, _ = parse_auxiliary_metadata(raw)
entries = extract_keystorage_entries(raw, records_offset, version)
if not entries:
sys.exit("Nessun record KeyStorage trovato nel log: impossibile costruire la richiesta API.")
keychain_array = [
{
"featurePoint": _FEATURE_ID_TO_NAME.get(fid, "PlaintextFeature"),
"aesCiphertext": base64.b64encode(data).decode("ascii"),
}
for fid, data in entries
]
body = json.dumps({
"version": version_meta,
"department": department,
"keychainsArray": [keychain_array],
}).encode("utf-8")
req = urllib.request.Request(
endpoint,
data=body,
headers={"Content-Type": "application/json", "Api-Key": api_key},
method="POST",
)
print(f"Chiamata API DJI: {endpoint} (version={version_meta}, department={department}, "
f"{len(keychain_array)} feature points)")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
response_json = json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
body_text = e.read().decode("utf-8", errors="replace")
sys.exit(f"Errore HTTP {e.code} dall'API DJI: {body_text}")
except urllib.error.URLError as e:
sys.exit(
f"Impossibile raggiungere l'API DJI ({e}). Verifica la connettivita' di rete "
"verso dev.dji.com da questa macchina."
)
result = response_json.get("result", {})
if result.get("code", -1) != 0:
sys.exit(f"L'API DJI ha risposto con errore: {result.get('msg')}")
data = response_json.get("data")
if not data:
sys.exit("Risposta API DJI senza dati di keychain.")
# data ha la stessa forma di keychains_cache.json: List[List[{featurePoint,aesKey,aesIv}]]
cache_path = str(Path(sys.argv[1]).with_suffix("")) + "_keychain.json"
with open(cache_path, "w") as f:
json.dump(data, f, indent=2)
print(f"Keychain ottenuta e salvata in: {cache_path} (riusabile offline con l'argomento keychain_file)")
keychain = {}
for entry in data[0]:
fname = entry["featurePoint"]
key_bytes = base64.b64decode(entry["aesKey"])
iv_bytes = base64.b64decode(entry["aesIv"])
for internal, jsonname in _FEATURE_JSON_NAME.items():
if jsonname == fname:
keychain[internal] = [iv_bytes, key_bytes]
break
return keychain
def load_keychain(path: str) -> dict:
"""Carica keychains_cache.json -> dict feature_name -> [iv_bytes, key_bytes] (mutabile)."""
with open(path) as f:
keychain_json = json.load(f)[0] # usa il primo set di chiavi del file
keychain = {}
for entry in keychain_json:
fname = entry["featurePoint"]
key_bytes = base64.b64decode(entry["aesKey"])
iv_bytes = base64.b64decode(entry["aesIv"])
for internal, jsonname in _FEATURE_JSON_NAME.items():
if jsonname == fname:
keychain[internal] = [iv_bytes, key_bytes]
break
return keychain
# ----------------------------------------------------------------------------
# Parsing del file di log
# ----------------------------------------------------------------------------
def parse_header(raw: bytes):
"""Ritorna (detail_offset, detail_length, version)."""
detail_offset, detail_length, version, _unknown = struct.unpack_from("<QHBB", raw, 0)
return detail_offset, detail_length, version
def iter_records(raw: bytes, records_offset: int):
"""Itera (record_type, length_field, payload_start, payload_bytes) senza decrittare."""
pos = records_offset
end = len(raw)
while pos < end:
if pos + 1 > end:
return
record_type = raw[pos]
pos += 1
if pos + 2 > end:
return
length_field = struct.unpack_from("<H", raw, pos)[0]
pos += 2
if length_field <= 2:
return
payload_start = pos
if pos + length_field > end:
return
payload = raw[payload_start: payload_start + length_field]
pos += length_field
if pos >= end:
return
end_byte = raw[pos]
pos += 1
if end_byte != 0xFF:
# framing rotto: interrompe la scansione da qui in poi
return
yield record_type, length_field, payload_start, payload
def decrypt_record(record_type, length_field, payload, version, keychain):
"""Ritorna bytes in chiaro, oppure None se non decodificabile (plaintext o manca chiave)."""
fname = feature_point(record_type, version)
if fname == "PlaintextFeature" or fname not in keychain:
return None
first_byte = payload[0]
xor = XorKeystream(first_byte, record_type)
ciphertext_xored_input = payload[1:]
cipher_len = length_field - 2
if cipher_len <= 0 or cipher_len > len(ciphertext_xored_input):
return None
ciphertext = xor.decode(ciphertext_xored_input[:cipher_len])
if len(ciphertext) == 0 or len(ciphertext) % 16 != 0:
return None
iv, key = keychain[fname]
next_iv = ciphertext[-16:]
try:
cipher = AES.new(key, AES.MODE_CBC, iv)
plain_padded = cipher.decrypt(ciphertext)
try:
plain = unpad(plain_padded, 16)
except ValueError:
plain = plain_padded
except Exception:
keychain[fname][0] = next_iv
return None
keychain[fname][0] = next_iv # aggiorna IV "a catena" per il prossimo record della stessa feature
return plain
# ----------------------------------------------------------------------------
# Interpretazione del contenuto dei record
# ----------------------------------------------------------------------------
def parse_rtk_record(plain: bytes):
"""Interpreta un record tipo 45 (RTKDifferenceDataType) gia' decrittato."""
if len(plain) < 61:
return None
return {
"rover_lat": struct.unpack_from("<d", plain, 11)[0],
"rover_lon": struct.unpack_from("<d", plain, 19)[0],
"rover_ellip_height_m": struct.unpack_from("<f", plain, 27)[0],
"base_lat": struct.unpack_from("<d", plain, 31)[0],
"base_lon": struct.unpack_from("<d", plain, 39)[0],
"base_ellip_height_m": struct.unpack_from("<f", plain, 47)[0],
"heading_or_quality_deg": struct.unpack_from("<f", plain, 51)[0],
"fix_flag": struct.unpack_from("<f", plain, 57)[0],
}
def parse_osd_fly_time(plain: bytes):
"""Estrae fly_time (secondi) da un record OSD (tipo 1) gia' decrittato."""
if len(plain) < 44:
return None
fly_time_raw = struct.unpack_from("<H", plain, 42)[0]
return fly_time_raw / 10.0
# ----------------------------------------------------------------------------
# Pipeline principale
# ----------------------------------------------------------------------------
def extract_rtk(log_path: str, keychain_path: str | None, output_csv: str, api_key: str | None = None):
raw = Path(log_path).read_bytes()
detail_offset, detail_length, version = parse_header(raw)
print(f"Versione log: {version} | detail_offset: {detail_offset} | detail_length: {detail_length}")
records_offset = detail_offset if version > 12 else 100 # 100 = PREFIX_SIZE per log non criptati
if version >= 13 and not keychain_path and not api_key:
sys.exit(
f"Il log e' versione {version} (criptato). Serve una keychain JSON gia' salvata "
"(secondo argomento) oppure una API key DJI (--api-key) per recuperarla automaticamente."
)
if version >= 13 and api_key and not keychain_path:
keychain = fetch_keychain_via_api(raw, records_offset, version, api_key)
elif keychain_path:
keychain = load_keychain(keychain_path)
else:
keychain = {}
rtk_records = [] # (file_offset, plain_bytes)
osd_records = [] # (file_offset, fly_time_s)
n_total = 0
for record_type, length_field, payload_start, payload in iter_records(raw, records_offset):
n_total += 1
plain = decrypt_record(record_type, length_field, payload, version, keychain)
if plain is None:
continue
if record_type == 45:
rtk_records.append((payload_start, plain))
elif record_type == 1:
fly_time = parse_osd_fly_time(plain)
if fly_time is not None:
osd_records.append((payload_start, fly_time))
print(f"Record totali scansionati: {n_total}")
print(f"Record RTK (tipo 45) decodificati: {len(rtk_records)}")
print(f"Record OSD (tipo 1) decodificati: {len(osd_records)}")
if not rtk_records:
print("Nessun record RTK trovato: controlla che il log contenga davvero dati RTK, "
"o che la keychain sia quella corretta per questo file.")
osd_records.sort()
def nearest_fly_time(target_offset):
if not osd_records:
return None
best_off, best_ft, best_diff = None, None, None
for off, ft in osd_records:
diff = abs(off - target_offset)
if best_diff is None or diff < best_diff:
best_off, best_ft, best_diff = off, ft, diff
return best_ft
rows = []
for idx, (off, plain) in enumerate(rtk_records):
parsed = parse_rtk_record(plain)
if parsed is None:
continue
fly_time = nearest_fly_time(off)
rows.append({
"record_idx": idx,
"file_offset": off,
"fly_time_s": fly_time,
**parsed,
})
if rows:
fieldnames = list(rows[0].keys())
with open(output_csv, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=fieldnames)
w.writeheader()
w.writerows(rows)
print(f"\nScritte {len(rows)} epoche RTK in: {output_csv}")
else:
print("Nessuna epoca RTK esportata (0 righe).")
def main():
parser = argparse.ArgumentParser(
description="Estrae dati RTK epoca-per-epoca da un flight log DJI (TXT) in CSV.",
)
parser.add_argument("log_file", help="Percorso al file DJIFlightRecord_....txt")
parser.add_argument(
"keychain_file", nargs="?", default=None,
help="Percorso a keychains_cache.json gia' salvata (opzionale se si usa --api-key)",
)
parser.add_argument(
"-o", "--output", default="rtk_epochs.csv",
help="File CSV di output (default: rtk_epochs.csv)",
)
parser.add_argument(
"--api-key", default=None,
help="API key DJI: se fornita (e non si passa keychain_file), la keychain viene "
"recuperata automaticamente da dev.dji.com e salvata in <log>_keychain.json "
"per riuso futuro offline.",
)
args = parser.parse_args()
extract_rtk(args.log_file, args.keychain_file, args.output, api_key=args.api_key)
if __name__ == "__main__":
main()
