123 lines
4.3 KiB
Python
123 lines
4.3 KiB
Python
from collections import defaultdict
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Dict, Any
|
|
from lxml import etree
|
|
|
|
|
|
class GPXParser:
|
|
def __init__(self, file: str):
|
|
self.file = file
|
|
self.gpx = etree.parse(self.file)
|
|
self.namespace = None
|
|
self.precision_digits = 6
|
|
self.max_samples_per_minute = 2
|
|
self.set_namespace()
|
|
|
|
def get_timestamp_key(self, dt: datetime):
|
|
ts = round(dt.replace(microsecond=0).timestamp())
|
|
return ts - (ts % (60 / self.max_samples_per_minute))
|
|
|
|
async def group_data(self, data: list):
|
|
times = await self.get_times_all()
|
|
|
|
indexed_data: dict[datetime, Any] = {}
|
|
for time, item in zip(times, data):
|
|
indexed_data[time] = item
|
|
|
|
sampled_data = defaultdict(list)
|
|
for time, item in indexed_data.items():
|
|
key = self.get_timestamp_key(time)
|
|
sampled_data[key].append(item)
|
|
|
|
return sampled_data
|
|
|
|
async def average_sample_numbers(self, data: list[int | float]):
|
|
sampled_data = await self.group_data(data)
|
|
return [sum(items) / len(items) for items in sampled_data.values()]
|
|
|
|
async def sample_times(self, times: List[datetime]):
|
|
sampled_times = set()
|
|
for time in times:
|
|
original_tz = time.tzinfo
|
|
sampled_times.add(datetime.fromtimestamp(self.get_timestamp_key(time), tz=original_tz))
|
|
|
|
times_sorted = list(sampled_times)
|
|
times_sorted.sort()
|
|
return times_sorted
|
|
|
|
async def average_coordinates(self, coordinates: list[dict[str, float]]):
|
|
sampled_data = await self.group_data(coordinates)
|
|
coordinates_aggregated = []
|
|
for coordinates_group in sampled_data.values():
|
|
lat = [c['lat'] for c in coordinates_group]
|
|
lng = [c['lng'] for c in coordinates_group]
|
|
|
|
coordinates_aggregated.append({
|
|
"lat": sum(lat) / len(lat),
|
|
"lng": sum(lng) / len(lng),
|
|
})
|
|
|
|
return coordinates_aggregated
|
|
|
|
def set_namespace(self):
|
|
namespace = self.gpx.getroot().nsmap.get(None)
|
|
self.namespace = {'gpx': namespace}
|
|
|
|
def run_xpath(self, path: str):
|
|
return self.gpx.xpath(path, namespaces=self.namespace)
|
|
|
|
async def get_times_all(self):
|
|
nodes = self.run_xpath("//gpx:trkpt/gpx:time")
|
|
return [datetime.fromisoformat(node.text).astimezone() for node in nodes]
|
|
|
|
async def get_times(self):
|
|
times = await self.get_times_all()
|
|
return await self.sample_times(times)
|
|
|
|
async def get_total_duration(self) -> timedelta:
|
|
times = await self.get_times_all()
|
|
return times[-1] - times[0]
|
|
|
|
async def get_coordinates(self) -> List[Dict[str, float]]:
|
|
return await self.average_coordinates(await self.get_coordinates_all())
|
|
|
|
async def get_coordinates_all(self) -> List[Dict[str, float]]:
|
|
nodes = self.run_xpath("//gpx:trkpt")
|
|
return [{"lat": float(node.attrib["lat"]), "lng": float(node.attrib['lon'])} for node in nodes]
|
|
|
|
async def get_speed(self) -> List[float]:
|
|
nodes = self.run_xpath("//gpx:speed")
|
|
return await self.average_sample_numbers([float(node.text) for node in nodes])
|
|
|
|
async def get_magnetic_variation(self) -> List[float]:
|
|
nodes = self.run_xpath("//gpx:magvar")
|
|
return await self.average_sample_numbers([int(node.text) for node in nodes])
|
|
|
|
async def get_altitude(self) -> List[float]:
|
|
nodes = self.run_xpath("//gpx:ele")
|
|
return await self.average_sample_numbers([float(node.text) for node in nodes])
|
|
|
|
async def get_terrain_elevation(self) -> List[float]:
|
|
nodes = self.run_xpath("//gpx:terrain_elevation")
|
|
return await self.average_sample_numbers([float(node.text) for node in nodes])
|
|
|
|
async def get_max_speed(self):
|
|
return max(await self.get_speed()) or 0
|
|
|
|
async def get_min_speed(self):
|
|
return min(await self.get_speed()) or 0
|
|
|
|
async def get_avg_speed(self):
|
|
speeds = await self.get_speed()
|
|
if not speeds:
|
|
return 0
|
|
|
|
return round(sum(speeds) / len(speeds), 2)
|
|
|
|
async def get_max_altitude(self):
|
|
return max(await self.get_altitude()) or 0
|
|
|
|
async def get_avg_altitude(self):
|
|
altitudes = await self.get_altitude()
|
|
return round(sum(altitudes) / len(altitudes), 2)
|