777 lines
28 KiB
Python
777 lines
28 KiB
Python
import os
|
|
import math
|
|
import requests
|
|
import dateutil.parser
|
|
from datetime import datetime, timedelta, timezone
|
|
import xml.etree.ElementTree as ET
|
|
from PIL import Image
|
|
from fractions import Fraction
|
|
from django.conf import settings
|
|
import time
|
|
|
|
def parse_gpx(gpx_file_path):
|
|
"""
|
|
Parses a GPX file and returns a list of dictionaries with trackpoint coordinates,
|
|
elevation, and timestamp.
|
|
"""
|
|
if not os.path.exists(gpx_file_path):
|
|
return []
|
|
|
|
try:
|
|
tree = ET.parse(gpx_file_path)
|
|
root = tree.getroot()
|
|
except Exception as e:
|
|
print(f"Error parsing GPX file XML: {e}")
|
|
return []
|
|
|
|
# Strip namespace if any
|
|
ns = ""
|
|
if root.tag.startswith("{"):
|
|
ns = root.tag.split("}")[0] + "}"
|
|
|
|
points = []
|
|
for trkpt in root.findall(f'.//{ns}trkpt'):
|
|
try:
|
|
lat = float(trkpt.attrib['lat'])
|
|
lon = float(trkpt.attrib['lon'])
|
|
except (KeyError, ValueError):
|
|
continue
|
|
|
|
time_node = trkpt.find(f'{ns}time')
|
|
time_val = None
|
|
if time_node is not None and time_node.text:
|
|
try:
|
|
time_val = dateutil.parser.parse(time_node.text)
|
|
except Exception:
|
|
pass
|
|
|
|
ele_node = trkpt.find(f'{ns}ele')
|
|
ele_val = 0.0
|
|
if ele_node is not None and ele_node.text:
|
|
try:
|
|
ele_val = float(ele_node.text)
|
|
except Exception:
|
|
pass
|
|
|
|
points.append({
|
|
'lat': lat,
|
|
'lon': lon,
|
|
'time': time_val,
|
|
'ele': ele_val
|
|
})
|
|
|
|
# Sort points by time if timestamps exist
|
|
if all(p['time'] is not None for p in points):
|
|
points.sort(key=lambda x: x['time'])
|
|
|
|
return points
|
|
|
|
def haversine_distance(lat1, lon1, lat2, lon2):
|
|
"""
|
|
Computes the great-circle distance between two points in meters.
|
|
"""
|
|
R = 6371000 # Radius of earth in meters
|
|
phi1 = math.radians(lat1)
|
|
phi2 = math.radians(lat2)
|
|
delta_phi = math.radians(lat2 - lat1)
|
|
delta_lambda = math.radians(lon2 - lon1)
|
|
|
|
a = math.sin(delta_phi / 2) ** 2 + \
|
|
math.cos(phi1) * math.cos(phi2) * \
|
|
math.sin(delta_lambda / 2) ** 2
|
|
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
|
|
return R * c
|
|
|
|
def interpolate_gpx(gpx_points, timestamp):
|
|
"""
|
|
Linearly interpolates position (lat, lon, ele) on a GPX track at a given timestamp.
|
|
"""
|
|
if not gpx_points:
|
|
return None
|
|
|
|
# If single point or timestamp outside bounds, return boundary point
|
|
if len(gpx_points) == 1:
|
|
return gpx_points[0]
|
|
|
|
# Ensure timezone awareness match
|
|
tz = timestamp.tzinfo
|
|
t_start = gpx_points[0]['time']
|
|
t_end = gpx_points[-1]['time']
|
|
|
|
# Adjust timezone if needed
|
|
if t_start and t_start.tzinfo != tz:
|
|
if tz is None:
|
|
t_start = t_start.replace(tzinfo=None)
|
|
t_end = t_end.replace(tzinfo=None)
|
|
for p in gpx_points:
|
|
if p['time']:
|
|
p['time'] = p['time'].replace(tzinfo=None)
|
|
else:
|
|
timestamp = timestamp.astimezone(t_start.tzinfo)
|
|
|
|
if t_start and timestamp <= t_start:
|
|
return gpx_points[0]
|
|
if t_end and timestamp >= t_end:
|
|
return gpx_points[-1]
|
|
|
|
# Binary search to find the segment
|
|
low = 0
|
|
high = len(gpx_points) - 2
|
|
|
|
while low <= high:
|
|
mid = (low + high) // 2
|
|
p1 = gpx_points[mid]
|
|
p2 = gpx_points[mid + 1]
|
|
|
|
if p1['time'] <= timestamp <= p2['time']:
|
|
# Found segment, interpolate
|
|
total_sec = (p2['time'] - p1['time']).total_seconds()
|
|
if total_sec == 0:
|
|
return p1
|
|
fraction = (timestamp - p1['time']).total_seconds() / total_sec
|
|
|
|
lat = p1['lat'] + (p2['lat'] - p1['lat']) * fraction
|
|
lon = p1['lon'] + (p2['lon'] - p1['lon']) * fraction
|
|
ele = p1['ele'] + (p2['ele'] - p1['ele']) * fraction
|
|
|
|
return {'lat': lat, 'lon': lon, 'ele': ele, 'time': timestamp}
|
|
|
|
elif timestamp < p1['time']:
|
|
high = mid - 1
|
|
else:
|
|
low = mid + 1
|
|
|
|
return gpx_points[0]
|
|
|
|
def make_gpano_xmp(width, height):
|
|
xmp_xml = f"""<?xpacket begin="" id="W5M0MpCehiHzreSzNTczkc9d"?>
|
|
<x:xmpmeta xmlns:x="adobe:ns:meta/">
|
|
<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">
|
|
<rdf:Description rdf:about=""
|
|
xmlns:GPano="http://ns.google.com/photos/1.0/panorama/">
|
|
<GPano:ProjectionType>equirectangular</GPano:ProjectionType>
|
|
<GPano:UsePanoramaViewer>True</GPano:UsePanoramaViewer>
|
|
<GPano:CroppedAreaImageWidthPixels>{width}</GPano:CroppedAreaImageWidthPixels>
|
|
<GPano:CroppedAreaImageHeightPixels>{height}</GPano:CroppedAreaImageHeightPixels>
|
|
<GPano:FullPanoWidthPixels>{width}</GPano:FullPanoWidthPixels>
|
|
<GPano:FullPanoHeightPixels>{height}</GPano:FullPanoHeightPixels>
|
|
<GPano:CroppedAreaLeftPixels>0</GPano:CroppedAreaLeftPixels>
|
|
<GPano:CroppedAreaTopPixels>0</GPano:CroppedAreaTopPixels>
|
|
</rdf:Description>
|
|
</rdf:RDF>
|
|
</x:xmpmeta>
|
|
<?xpacket end="r"?>"""
|
|
return xmp_xml.encode('utf-8')
|
|
|
|
|
|
def calculate_bearing(lat1, lon1, lat2, lon2):
|
|
"""
|
|
Calculates the initial bearing (heading) between two coordinates in degrees (0-360).
|
|
"""
|
|
lat1_rad = math.radians(lat1)
|
|
lat2_rad = math.radians(lat2)
|
|
delta_lon = math.radians(lon2 - lon1)
|
|
|
|
y = math.sin(delta_lon) * math.cos(lat2_rad)
|
|
x = math.cos(lat1_rad) * math.sin(lat2_rad) - \
|
|
math.sin(lat1_rad) * math.cos(lat2_rad) * math.cos(delta_lon)
|
|
|
|
bearing = math.atan2(y, x)
|
|
bearing_deg = (math.degrees(bearing) + 360) % 360
|
|
return bearing_deg
|
|
|
|
|
|
def calculate_stable_heading(pos, gps_points, current_time, camera_offset=0, min_distance=10.0):
|
|
"""
|
|
Calculates the heading of travel by finding the first point forward on the GPX track
|
|
that is at least `min_distance` meters (default 10m, matching the photo interval) away from `pos`.
|
|
If no such point exists (at the end of the track), looks backward for a point `min_distance` meters behind.
|
|
Applies the `camera_offset` and returns the value modulo 360.
|
|
"""
|
|
total_offset = camera_offset
|
|
|
|
if not gps_points or not pos:
|
|
return (0.0 + total_offset) % 360
|
|
|
|
# 1. Search forward (future points)
|
|
forward_start_idx = None
|
|
for idx, p in enumerate(gps_points):
|
|
if p['time'] and p['time'] > current_time:
|
|
forward_start_idx = idx
|
|
break
|
|
|
|
if forward_start_idx is not None:
|
|
for idx in range(forward_start_idx, len(gps_points)):
|
|
p = gps_points[idx]
|
|
dist = haversine_distance(pos['lat'], pos['lon'], p['lat'], p['lon'])
|
|
if dist >= min_distance:
|
|
bearing = calculate_bearing(pos['lat'], pos['lon'], p['lat'], p['lon'])
|
|
return (bearing + total_offset) % 360
|
|
|
|
# 2. Search backward (past points)
|
|
backward_start_idx = None
|
|
for idx in range(len(gps_points) - 1, -1, -1):
|
|
p = gps_points[idx]
|
|
if p['time'] and p['time'] < current_time:
|
|
backward_start_idx = idx
|
|
break
|
|
|
|
if backward_start_idx is not None:
|
|
for idx in range(backward_start_idx, -1, -1):
|
|
p = gps_points[idx]
|
|
dist = haversine_distance(pos['lat'], pos['lon'], p['lat'], p['lon'])
|
|
if dist >= min_distance:
|
|
# Direction of travel is from past point p to current point pos
|
|
bearing = calculate_bearing(p['lat'], p['lon'], pos['lat'], pos['lon'])
|
|
return (bearing + total_offset) % 360
|
|
|
|
# 3. Fallback: overall track direction
|
|
if len(gps_points) > 1:
|
|
last_pt = gps_points[-1]
|
|
first_pt = gps_points[0]
|
|
if last_pt != first_pt:
|
|
bearing = calculate_bearing(first_pt['lat'], first_pt['lon'], last_pt['lat'], last_pt['lon'])
|
|
return (bearing + total_offset) % 360
|
|
|
|
return (0.0 + total_offset) % 360
|
|
|
|
|
|
def write_gps_exif(image_path, lat, lon, alt, timestamp, heading=None):
|
|
"""
|
|
Writes GPS location, timestamp, and optional heading metadata to a JPEG image.
|
|
If the image aspect ratio is close to 2:1, also injects GPano XMP
|
|
metadata to declare it as a 360° equirectangular panorama.
|
|
"""
|
|
try:
|
|
img = Image.open(image_path)
|
|
exif = img.getexif()
|
|
|
|
# DateTimeOriginal
|
|
if timestamp:
|
|
exif[36867] = timestamp.strftime('%Y:%m:%d %H:%M:%S')
|
|
|
|
gps_ifd = exif.get_ifd(34853)
|
|
|
|
# Latitude
|
|
lat_ref = 'N' if lat >= 0 else 'S'
|
|
lat_val = abs(lat)
|
|
lat_deg = int(lat_val)
|
|
lat_min = int((lat_val - lat_deg) * 60)
|
|
lat_sec = (lat_val - lat_deg - lat_min / 60) * 3600
|
|
gps_ifd[1] = lat_ref
|
|
gps_ifd[2] = (Fraction(lat_deg), Fraction(lat_min), Fraction(lat_sec).limit_denominator(1000))
|
|
|
|
# Longitude
|
|
lon_ref = 'E' if lon >= 0 else 'W'
|
|
lon_val = abs(lon)
|
|
lon_deg = int(lon_val)
|
|
lon_min = int((lon_val - lon_deg) * 60)
|
|
lon_sec = (lon_val - lon_deg - lon_min / 60) * 3600
|
|
gps_ifd[3] = lon_ref
|
|
gps_ifd[4] = (Fraction(lon_deg), Fraction(lon_min), Fraction(lon_sec).limit_denominator(1000))
|
|
|
|
# Altitude
|
|
alt_ref = 0 if alt >= 0 else 1
|
|
gps_ifd[5] = alt_ref
|
|
gps_ifd[6] = Fraction(abs(alt)).limit_denominator(1000)
|
|
|
|
# GPS Time Stamp and Date Stamp in UTC
|
|
if timestamp:
|
|
utc_ts = timestamp.astimezone(timezone.utc) if timestamp.tzinfo else timestamp
|
|
gps_ifd[7] = (Fraction(utc_ts.hour), Fraction(utc_ts.minute), Fraction(utc_ts.second))
|
|
gps_ifd[29] = utc_ts.strftime('%Y:%m:%d')
|
|
|
|
# GPS Image Direction (Heading)
|
|
if heading is not None:
|
|
gps_ifd[16] = 'T'
|
|
gps_ifd[17] = Fraction(float(heading)).limit_denominator(1000)
|
|
|
|
# Detect if equirectangular 360 (aspect ratio ~2:1)
|
|
width, height = img.size
|
|
xmp_bytes = None
|
|
if 1.9 <= (width / height) <= 2.1:
|
|
xmp_bytes = make_gpano_xmp(width, height)
|
|
|
|
quality = getattr(settings, 'PANORAMAX_IMPORT_JPEG_QUALITY', 98)
|
|
if xmp_bytes:
|
|
img.save(image_path, exif=exif, xmp=xmp_bytes, quality=quality, subsampling=0)
|
|
else:
|
|
img.save(image_path, exif=exif, quality=quality, subsampling=0)
|
|
|
|
except Exception as e:
|
|
print(f"Error writing GPS EXIF to {image_path}: {e}")
|
|
|
|
def extract_frames_from_video(video_path, gps_points, output_dir, camera_offset=0, progress_callback=None):
|
|
"""
|
|
Extracts frames from a video every 10 meters based on GPS coordinates.
|
|
Saves extracted frames as geotagged JPEG images in output_dir.
|
|
Calculates heading based on consecutive photo positions (current to next).
|
|
"""
|
|
if not gps_points:
|
|
raise ValueError("GPS track points are required for video positioning.")
|
|
|
|
try:
|
|
import cv2
|
|
except ImportError:
|
|
raise ImportError("OpenCV (cv2) is required for video frame extraction but is not installed.")
|
|
|
|
cap = cv2.VideoCapture(video_path)
|
|
if not cap.isOpened():
|
|
raise ValueError(f"Could not open video file: {video_path}")
|
|
|
|
fps = cap.get(cv2.CAP_PROP_FPS)
|
|
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
|
if fps <= 0 or total_frames <= 0:
|
|
cap.release()
|
|
raise ValueError("Invalid video file frame rate or length.")
|
|
|
|
duration = total_frames / fps
|
|
video_start_time = gps_points[0]['time']
|
|
if not video_start_time:
|
|
video_start_time = datetime.now(timezone.utc)
|
|
|
|
extracted_count = 0
|
|
last_pos = None
|
|
extracted_metadata = []
|
|
|
|
# Pass 1: Extract frames and save them as temporary files
|
|
step = 0.2
|
|
t = 0.0
|
|
|
|
while t < duration:
|
|
frame_time = video_start_time + timedelta(seconds=t)
|
|
pos = interpolate_gpx(gps_points, frame_time)
|
|
|
|
should_extract = False
|
|
if last_pos is None:
|
|
should_extract = True
|
|
else:
|
|
dist = haversine_distance(last_pos['lat'], last_pos['lon'], pos['lat'], pos['lon'])
|
|
if dist >= 10.0:
|
|
should_extract = True
|
|
|
|
if should_extract:
|
|
cap.set(cv2.CAP_PROP_POS_MSEC, t * 1000)
|
|
ret, frame = cap.read()
|
|
if ret:
|
|
extracted_count += 1
|
|
filename = f"frame_{extracted_count:05d}.jpg"
|
|
filepath = os.path.join(output_dir, filename)
|
|
|
|
# Use configured quality and support Unicode paths safely on Windows
|
|
quality = getattr(settings, 'PANORAMAX_IMPORT_JPEG_QUALITY', 98)
|
|
ext = os.path.splitext(filepath)[1]
|
|
params = [int(cv2.IMWRITE_JPEG_QUALITY), quality]
|
|
ret, buf = cv2.imencode(ext, frame, params)
|
|
if not ret:
|
|
raise ValueError(f"Could not encode video frame to JPEG format: {filepath}")
|
|
with open(filepath, "wb") as f:
|
|
f.write(buf.tobytes())
|
|
|
|
extracted_metadata.append({
|
|
'filepath': filepath,
|
|
'pos': pos,
|
|
'time': pos['time']
|
|
})
|
|
last_pos = pos
|
|
|
|
t += step
|
|
|
|
# Report progress up to 70% during frame extraction pass
|
|
if progress_callback:
|
|
progress_callback(int((t / duration) * 70))
|
|
|
|
time.sleep(0.01)
|
|
|
|
cap.release()
|
|
|
|
total_offset = camera_offset
|
|
num_photos = len(extracted_metadata)
|
|
image_paths = []
|
|
|
|
for idx, item in enumerate(extracted_metadata):
|
|
filepath = item['filepath']
|
|
pos = item['pos']
|
|
|
|
if num_photos > 1:
|
|
if idx < num_photos - 1:
|
|
next_pos = extracted_metadata[idx + 1]['pos']
|
|
bearing = calculate_bearing(pos['lat'], pos['lon'], next_pos['lat'], next_pos['lon'])
|
|
else:
|
|
prev_pos = extracted_metadata[idx - 1]['pos']
|
|
bearing = calculate_bearing(prev_pos['lat'], prev_pos['lon'], pos['lat'], pos['lon'])
|
|
heading = (bearing + total_offset) % 360
|
|
else:
|
|
heading = (0.0 + total_offset) % 360
|
|
|
|
write_gps_exif(filepath, pos['lat'], pos['lon'], pos['ele'], pos['time'], heading=heading)
|
|
image_paths.append(filepath)
|
|
|
|
# Report progress from 70% to 80% during EXIF pass
|
|
if progress_callback:
|
|
progress_callback(int(70 + (idx / num_photos) * 10))
|
|
|
|
time.sleep(0.01)
|
|
|
|
return image_paths
|
|
|
|
def call_panoramax_api_create_set(title, estimated_count, creator=None):
|
|
"""
|
|
Creates an UploadSet in Panoramax.
|
|
"""
|
|
url = f"{settings.PANORAMAX_API_URL.rstrip('/')}/upload_sets"
|
|
headers = {}
|
|
if settings.PANORAMAX_API_KEY:
|
|
headers['Authorization'] = f'Bearer {settings.PANORAMAX_API_KEY}'
|
|
|
|
data = {
|
|
'title': title,
|
|
'estimated_nb_files': estimated_count,
|
|
'visibility': 'anyone'
|
|
}
|
|
if creator:
|
|
data['semantics'] = [
|
|
{'key': 'creator', 'value': creator}
|
|
]
|
|
|
|
response = requests.post(url, headers=headers, json=data, timeout=30)
|
|
response.raise_for_status()
|
|
return response.json()['id']
|
|
|
|
def call_panoramax_api_upload_file(upload_set_id, filepath):
|
|
"""
|
|
Uploads a single image file to a Panoramax UploadSet.
|
|
"""
|
|
url = f"{settings.PANORAMAX_API_URL.rstrip('/')}/upload_sets/{upload_set_id}/files"
|
|
headers = {}
|
|
if settings.PANORAMAX_API_KEY:
|
|
headers['Authorization'] = f'Bearer {settings.PANORAMAX_API_KEY}'
|
|
|
|
filename = os.path.basename(filepath)
|
|
with open(filepath, 'rb') as f:
|
|
files = {
|
|
'file': (filename, f, 'image/jpeg')
|
|
}
|
|
response = requests.post(url, headers=headers, files=files, timeout=60)
|
|
response.raise_for_status()
|
|
|
|
def call_panoramax_api_complete_set(upload_set_id):
|
|
"""
|
|
Marks an UploadSet in Panoramax as complete.
|
|
"""
|
|
url = f"{settings.PANORAMAX_API_URL.rstrip('/')}/upload_sets/{upload_set_id}/complete"
|
|
headers = {}
|
|
if settings.PANORAMAX_API_KEY:
|
|
headers['Authorization'] = f'Bearer {settings.PANORAMAX_API_KEY}'
|
|
|
|
response = requests.post(url, headers=headers, timeout=30)
|
|
response.raise_for_status()
|
|
|
|
|
|
class MP4Parser:
|
|
def __init__(self, filepath):
|
|
self.filepath = filepath
|
|
self.gpmd_samples = []
|
|
|
|
def parse(self):
|
|
with open(self.filepath, 'rb') as f:
|
|
f.seek(0, 2)
|
|
file_size = f.tell()
|
|
boxes = self.read_boxes(f, 0, file_size)
|
|
|
|
moov = self.find_box_by_type(boxes, 'moov')
|
|
if not moov:
|
|
return []
|
|
|
|
moov_boxes = self.read_boxes(f, moov['data_start'], moov['start'] + moov['size'])
|
|
traks = [b for b in moov_boxes if b['type'] == 'trak']
|
|
|
|
for trak in traks:
|
|
trak_boxes = self.read_boxes(f, trak['data_start'], trak['start'] + trak['size'])
|
|
|
|
mdia = self.find_box_by_type(trak_boxes, 'mdia')
|
|
if not mdia: continue
|
|
mdia_boxes = self.read_boxes(f, mdia['data_start'], mdia['start'] + mdia['size'])
|
|
|
|
minf = self.find_box_by_type(mdia_boxes, 'minf')
|
|
if not minf: continue
|
|
minf_boxes = self.read_boxes(f, minf['data_start'], minf['start'] + minf['size'])
|
|
|
|
stbl = self.find_box_by_type(minf_boxes, 'stbl')
|
|
if not stbl: continue
|
|
stbl_boxes = self.read_boxes(f, stbl['data_start'], stbl['start'] + stbl['size'])
|
|
|
|
stsd = self.find_box_by_type(stbl_boxes, 'stsd')
|
|
if not stsd: continue
|
|
f.seek(stsd['data_start'] + 8)
|
|
|
|
entry_size = int.from_bytes(f.read(4), 'big')
|
|
entry_format = f.read(4).decode('latin1', errors='ignore')
|
|
|
|
if entry_format == 'gpmd':
|
|
self.extract_trak_samples(f, stbl_boxes)
|
|
return self.gpmd_samples
|
|
return []
|
|
|
|
def read_boxes(self, f, start, end):
|
|
f.seek(start)
|
|
boxes = []
|
|
while f.tell() < end:
|
|
box_start = f.tell()
|
|
size_bytes = f.read(4)
|
|
if len(size_bytes) < 4:
|
|
break
|
|
size = int.from_bytes(size_bytes, 'big')
|
|
box_type = f.read(4).decode('latin1', errors='ignore')
|
|
|
|
header_size = 8
|
|
if size == 1:
|
|
size = int.from_bytes(f.read(8), 'big')
|
|
header_size = 16
|
|
elif size == 0:
|
|
f.seek(0, 2)
|
|
size = f.tell() - box_start
|
|
|
|
boxes.append({
|
|
'type': box_type,
|
|
'start': box_start,
|
|
'header_size': header_size,
|
|
'size': size,
|
|
'data_start': box_start + header_size
|
|
})
|
|
f.seek(box_start + size)
|
|
return boxes
|
|
|
|
def find_box_by_type(self, boxes, box_type):
|
|
for b in boxes:
|
|
if b['type'] == box_type:
|
|
return b
|
|
return None
|
|
|
|
def extract_trak_samples(self, f, stbl_boxes):
|
|
stsz = self.find_box_by_type(stbl_boxes, 'stsz')
|
|
stco = self.find_box_by_type(stbl_boxes, 'stco')
|
|
co64 = self.find_box_by_type(stbl_boxes, 'co64')
|
|
stsc = self.find_box_by_type(stbl_boxes, 'stsc')
|
|
|
|
if not stsz or not (stco or co64) or not stsc:
|
|
return
|
|
|
|
f.seek(stsz['data_start'] + 4)
|
|
def_size = int.from_bytes(f.read(4), 'big')
|
|
num_samples = int.from_bytes(f.read(4), 'big')
|
|
|
|
sizes = []
|
|
if def_size > 0:
|
|
sizes = [def_size] * num_samples
|
|
else:
|
|
for _ in range(num_samples):
|
|
sizes.append(int.from_bytes(f.read(4), 'big'))
|
|
|
|
offsets = []
|
|
if stco:
|
|
f.seek(stco['data_start'] + 4)
|
|
num_chunks = int.from_bytes(f.read(4), 'big')
|
|
for _ in range(num_chunks):
|
|
offsets.append(int.from_bytes(f.read(4), 'big'))
|
|
elif co64:
|
|
f.seek(co64['data_start'] + 4)
|
|
num_chunks = int.from_bytes(f.read(4), 'big')
|
|
for _ in range(num_chunks):
|
|
offsets.append(int.from_bytes(f.read(8), 'big'))
|
|
|
|
f.seek(stsc['data_start'] + 4)
|
|
num_stsc = int.from_bytes(f.read(4), 'big')
|
|
stsc_entries = []
|
|
for _ in range(num_stsc):
|
|
stsc_entries.append({
|
|
'first_chunk': int.from_bytes(f.read(4), 'big'),
|
|
'samples_per_chunk': int.from_bytes(f.read(4), 'big'),
|
|
'desc_idx': int.from_bytes(f.read(4), 'big')
|
|
})
|
|
|
|
sample_idx = 0
|
|
samples_in_chunks = []
|
|
for i in range(len(stsc_entries)):
|
|
current = stsc_entries[i]
|
|
next_chunk = stsc_entries[i+1]['first_chunk'] if i + 1 < len(stsc_entries) else len(offsets) + 1
|
|
for c in range(current['first_chunk'], next_chunk):
|
|
samples_in_chunks.append((c - 1, current['samples_per_chunk']))
|
|
|
|
for chunk_i, num_samples_in_chunk in samples_in_chunks:
|
|
if chunk_i >= len(offsets):
|
|
break
|
|
chunk_offset = offsets[chunk_i]
|
|
for _ in range(num_samples_in_chunk):
|
|
if sample_idx >= len(sizes):
|
|
break
|
|
size = sizes[sample_idx]
|
|
self.gpmd_samples.append({
|
|
'offset': chunk_offset,
|
|
'size': size
|
|
})
|
|
chunk_offset += size
|
|
sample_idx += 1
|
|
|
|
|
|
def parse_gpmf_value(type_char, struct_size, repeat, data):
|
|
import struct
|
|
try:
|
|
if type_char == 'c':
|
|
return data.decode('latin1', errors='ignore').strip('\x00')
|
|
elif type_char == 'b':
|
|
return struct.unpack(f'>{repeat}b', data[:repeat]) if repeat > 1 else struct.unpack(f'>b', data[:1])[0]
|
|
elif type_char == 'B':
|
|
return struct.unpack(f'>{repeat}B', data[:repeat]) if repeat > 1 else struct.unpack(f'>B', data[:1])[0]
|
|
elif type_char == 's':
|
|
return struct.unpack(f'>{repeat}h', data[:repeat*2]) if repeat > 1 else struct.unpack(f'>h', data[:2])[0]
|
|
elif type_char == 'S':
|
|
return struct.unpack(f'>{repeat}H', data[:repeat*2]) if repeat > 1 else struct.unpack(f'>H', data[:2])[0]
|
|
elif type_char == 'l':
|
|
return struct.unpack(f'>{repeat}i', data[:repeat*4]) if repeat > 1 else struct.unpack(f'>i', data[:4])[0]
|
|
elif type_char == 'L':
|
|
return struct.unpack(f'>{repeat}I', data[:repeat*4]) if repeat > 1 else struct.unpack(f'>I', data[:4])[0]
|
|
elif type_char == 'f':
|
|
return struct.unpack(f'>{repeat}f', data[:repeat*4]) if repeat > 1 else struct.unpack(f'>f', data[:4])[0]
|
|
elif type_char == 'd':
|
|
return struct.unpack(f'>{repeat}d', data[:repeat*8]) if repeat > 1 else struct.unpack(f'>d', data[:8])[0]
|
|
elif type_char == 'F':
|
|
return data[:4].decode('latin1', errors='ignore')
|
|
elif type_char == 'U':
|
|
return data.decode('latin1', errors='ignore').strip('\x00')
|
|
except Exception:
|
|
pass
|
|
return data
|
|
|
|
|
|
def parse_klv(data, offset=0, end=None):
|
|
import struct
|
|
if end is None:
|
|
end = len(data)
|
|
|
|
results = {}
|
|
while offset < end:
|
|
if offset + 8 > end:
|
|
break
|
|
|
|
key = data[offset:offset+4].decode('latin1', errors='ignore')
|
|
type_char = chr(data[offset+4])
|
|
struct_size = data[offset+5]
|
|
repeat = struct.unpack('>H', data[offset+6:offset+8])[0]
|
|
|
|
header_size = 8
|
|
payload_size = struct_size * repeat
|
|
padded_size = ((payload_size + 3) // 4) * 4
|
|
|
|
if offset + header_size + payload_size > end:
|
|
break
|
|
|
|
value_data = data[offset+header_size : offset+header_size+payload_size]
|
|
|
|
if type_char == '\x00':
|
|
nested = parse_klv(value_data)
|
|
if key not in results:
|
|
results[key] = []
|
|
results[key].append(nested)
|
|
else:
|
|
val = parse_gpmf_value(type_char, struct_size, repeat, value_data)
|
|
results[key] = val
|
|
|
|
offset += header_size + padded_size
|
|
return results
|
|
|
|
|
|
def parse_gpsu_time(gpsu_str):
|
|
if not gpsu_str:
|
|
return None
|
|
try:
|
|
gpsu_str = gpsu_str.strip()
|
|
if '.' in gpsu_str:
|
|
date_part, ms_part = gpsu_str.split('.')
|
|
else:
|
|
date_part = gpsu_str
|
|
ms_part = '0'
|
|
|
|
if len(date_part) == 12:
|
|
year = 2000 + int(date_part[0:2])
|
|
month = int(date_part[2:4])
|
|
day = int(date_part[4:6])
|
|
hour = int(date_part[6:8])
|
|
minute = int(date_part[8:10])
|
|
second = int(date_part[10:12])
|
|
microsecond = int(float("0." + ms_part) * 1000000)
|
|
|
|
return datetime(year, month, day, hour, minute, second, microsecond, tzinfo=timezone.utc)
|
|
except Exception as e:
|
|
print(f"Error parsing GPSU custom format: {e}")
|
|
|
|
try:
|
|
return dateutil.parser.parse(gpsu_str)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def extract_gpmf_gps(filepath):
|
|
parser = MP4Parser(filepath)
|
|
samples = parser.parse()
|
|
if not samples:
|
|
return []
|
|
|
|
gps_points = []
|
|
|
|
with open(filepath, 'rb') as f:
|
|
for sample in samples:
|
|
f.seek(sample['offset'])
|
|
data = f.read(sample['size'])
|
|
|
|
try:
|
|
klv = parse_klv(data)
|
|
|
|
devices = []
|
|
if 'DEVI' in klv:
|
|
devices = klv['DEVI']
|
|
elif 'DEVC' in klv:
|
|
devices = klv['DEVC']
|
|
|
|
for devi in devices:
|
|
if 'STRM' in devi:
|
|
for strm in devi['STRM']:
|
|
if 'GPS5' in strm:
|
|
gps5 = strm['GPS5']
|
|
scal = strm.get('SCAL', 1)
|
|
gpsu = strm.get('GPSU')
|
|
|
|
timestamp = parse_gpsu_time(gpsu)
|
|
|
|
if isinstance(gps5, (tuple, list)) and len(gps5) > 0 and not isinstance(gps5[0], (tuple, list)):
|
|
gps5 = [gps5[i:i+5] for i in range(0, len(gps5), 5)]
|
|
|
|
for idx, pt in enumerate(gps5):
|
|
if len(pt) >= 3:
|
|
lat_scale = scal[0] if isinstance(scal, (list, tuple)) else scal
|
|
lon_scale = scal[1] if isinstance(scal, (list, tuple)) and len(scal) > 1 else lat_scale
|
|
alt_scale = scal[2] if isinstance(scal, (list, tuple)) and len(scal) > 2 else lat_scale
|
|
|
|
lat = pt[0] / lat_scale
|
|
lon = pt[1] / lon_scale
|
|
alt = pt[2] / alt_scale
|
|
|
|
pt_time = timestamp
|
|
if timestamp and len(gps5) > 1:
|
|
pt_time = timestamp + timedelta(seconds=(idx / 18.0))
|
|
|
|
gps_points.append({
|
|
'lat': lat,
|
|
'lon': lon,
|
|
'ele': alt,
|
|
'time': pt_time
|
|
})
|
|
except Exception as e:
|
|
print(f"Error parsing GPMF sample KLV: {e}")
|
|
|
|
if gps_points:
|
|
if all(p['time'] is not None for p in gps_points):
|
|
gps_points.sort(key=lambda x: x['time'])
|
|
|
|
return gps_points
|
|
|