aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--fietsboek/transformers/__init__.py7
-rw-r--r--fietsboek/transformers/elevation.py81
2 files changed, 65 insertions, 23 deletions
diff --git a/fietsboek/transformers/__init__.py b/fietsboek/transformers/__init__.py
index 7df700c..b1a0245 100644
--- a/fietsboek/transformers/__init__.py
+++ b/fietsboek/transformers/__init__.py
@@ -10,10 +10,10 @@ function to load and apply transformers.
"""
from abc import ABC, abstractmethod
-from collections.abc import Callable, Iterable, Mapping
+from collections.abc import Mapping
from typing import Literal, NamedTuple, TypeVar
-from gpxpy.gpx import GPX, GPXTrackPoint
+from gpxpy.gpx import GPX
from pydantic import BaseModel
from pyramid.i18n import TranslationString
from pyramid.request import Request
@@ -134,10 +134,11 @@ def list_transformers() -> list[type[Transformer]]:
"""
# pylint: disable=import-outside-toplevel,cyclic-import
from .breaks import RemoveBreaks
- from .elevation import FixNullElevation
+ from .elevation import FixElevationJumps, FixNullElevation
return [
FixNullElevation,
+ FixElevationJumps,
RemoveBreaks,
]
diff --git a/fietsboek/transformers/elevation.py b/fietsboek/transformers/elevation.py
index 0e6f3b0..0af5161 100644
--- a/fietsboek/transformers/elevation.py
+++ b/fietsboek/transformers/elevation.py
@@ -1,6 +1,6 @@
"""Transformers that deal with elevation changes in the track."""
from collections.abc import Callable, Iterable
-from itertools import islice
+from itertools import islice, zip_longest
from gpxpy.gpx import GPX, GPXTrackPoint
from pyramid.i18n import TranslationString
@@ -9,6 +9,26 @@ from . import Parameters, Transformer
_ = TranslationString
+MAX_ORGANIC_SLOPE: float = 1.0
+
+
+def slope(point_a: GPXTrackPoint, point_b: GPXTrackPoint) -> float:
+ """Returns the slope between two GPX points.
+
+ This is defined as delta_h / euclid_distance.
+
+ :param point_a: First point.
+ :param point_b: Second point.
+ :return: The slope, as percentage.
+ """
+ if point_a.elevation is None or point_b.elevation is None:
+ return 0.0
+ delta_h = abs(point_a.elevation - point_b.elevation)
+ dist = point_a.distance_2d(point_b)
+ if dist == 0.0 or dist is None:
+ return 0.0
+ return delta_h / dist
+
class FixNullElevation(Transformer):
"""A transformer that fixes points with zero elevation."""
@@ -67,15 +87,13 @@ class FixNullElevation(Transformer):
:param points: A function that generates the iterable of points.
"""
- max_slope = 1.0
-
bad_until = 0
final_elevation = 0.0
for i, (point, next_point) in enumerate(zip(points(), islice(points(), 1, None))):
if (
point.elevation is not None
and point.elevation != 0.0
- and cls.slope(point, next_point) < max_slope
+ and slope(point, next_point) < MAX_ORGANIC_SLOPE
):
bad_until = i
final_elevation = point.elevation
@@ -84,23 +102,46 @@ class FixNullElevation(Transformer):
for point in islice(points(), None, bad_until):
point.elevation = final_elevation
- @staticmethod
- def slope(point_a: GPXTrackPoint, point_b: GPXTrackPoint) -> float:
- """Returns the slope between two GPX points.
- This is defined as delta_h / euclid_distance.
+class FixElevationJumps(Transformer):
+ """A transformer that fixes big jumps in the elevation."""
- :param point_a: First point.
- :param point_b: Second point.
- :return: The slope, as percentage.
- """
- if point_a.elevation is None or point_b.elevation is None:
- return 0.0
- delta_h = abs(point_a.elevation - point_b.elevation)
- dist = point_a.distance_2d(point_b)
- if dist == 0.0 or dist is None:
- return 0.0
- return delta_h / dist
+ @classmethod
+ def identifier(cls) -> str:
+ return "fix-elevation-jumps"
+
+ @classmethod
+ def name(cls) -> TranslationString:
+ return _("transformers.fix-elevation-jumps")
+
+ @classmethod
+ def description(cls) -> TranslationString:
+ return _("transformers.fix-elevation-jumps.description")
+
+ @classmethod
+ def parameter_model(cls) -> type[Parameters]:
+ return Parameters
+
+ @property
+ def parameters(self) -> Parameters:
+ return Parameters()
+
+ @parameters.setter
+ def parameters(self, value):
+ pass
+
+ def execute(self, gpx: GPX):
+ current_adjustment = 0.0
+
+ points = gpx.walk(only_points=True)
+ next_points = gpx.walk(only_points=True)
+
+ for current_point, next_point in zip_longest(points, islice(next_points, 1, None)):
+ point_adjustment = current_adjustment
+ if next_point and slope(current_point, next_point) > MAX_ORGANIC_SLOPE:
+ current_adjustment += current_point.elevation - next_point.elevation
+ print(f"{current_adjustment=}")
+ current_point.elevation += point_adjustment
-__all__ = ["FixNullElevation"]
+__all__ = ["FixNullElevation", "FixElevationJumps"]