Pridani samplovani dat - neni potreba vracet polohu a rychlost kazdych 10s

src/external/gpx_parser.py

@@ -1,5 +1,6 @@
+from collections import defaultdict
 from datetime import datetime
-from typing import List, Dict
+from typing import List, Dict, Any
 from aiocache import cached
 from lxml import etree
 from lxml.etree import _ElementTree
@@ -11,8 +12,56 @@ class GPXParser:
         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)
+            print("item", time, key, item)
+            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}
@@ -21,34 +70,41 @@ class GPXParser:
         return self.gpx.xpath(path, namespaces=self.namespace)
 
     @cached()
-    async def get_times(self):
+    async def get_times_all(self):
         nodes = self.run_xpath("//gpx:trkpt/gpx:time")
         return [datetime.fromisoformat(node.text).astimezone() for node in nodes]
 
+    @cached()
+    async def get_times(self):
+        times = await self.get_times_all()
+        return await self.sample_times(times)
+
     @cached()
     async def get_coordinates(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]
+        return await self.average_coordinates(
+            [{"lat": float(node.attrib["lat"]), "lng": float(node.attrib['lon'])} for node in nodes]
+        )
 
     @cached()
-    async def get_speed(self) -> List[int]:
+    async def get_speed(self) -> List[float]:
         nodes = self.run_xpath("//gpx:speed")
-        return [int(node.text) for node in nodes]
+        return await self.average_sample_numbers([int(node.text) for node in nodes])
 
     @cached()
-    async def get_magnetic_variation(self) -> List[int]:
+    async def get_magnetic_variation(self) -> List[float]:
         nodes = self.run_xpath("//gpx:magvar")
-        return [int(node.text) for node in nodes]
+        return await self.average_sample_numbers([int(node.text) for node in nodes])
 
     @cached()
     async def get_altitude(self) -> List[float]:
         nodes = self.run_xpath("//gpx:ele")
-        return [float(node.text) for node in nodes]
+        return await self.average_sample_numbers([float(node.text) for node in nodes])
 
     @cached()
     async def get_terrain_elevation(self) -> List[float]:
         nodes = self.run_xpath("//gpx:terrain_elevation")
-        return [float(node.text) for node in nodes]
+        return await self.average_sample_numbers([float(node.text) for node in nodes])
 
     @cached()
     async def get_max_speed(self):