diff --git a/CHANGELOG.md b/CHANGELOG.md index 6233251..5b9a7d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,37 @@ The **third number** is for emergencies when we need to start branches for older ## [Unreleased](https://github.com/sgraaf/gpx/compare/2026.3.0...HEAD) +### Added + +- New `operations` module that exposes the editing and merging operations behind the `gpx edit` and `gpx merge` CLI commands as a public, reusable API: + - `crop()`: Crop a GPX to a geographic bounding box + - `trim()`: Trim a GPX to a date/time range + - `reverse()`: Reverse the routes and/or tracks of a GPX + - `strip_metadata()`: Strip metadata (fields) from a GPX + - `reduce_precision()`: Reduce the precision of the coordinates and/or elevations of a GPX + - `filter_points()`: Filter the points of a GPX with an arbitrary predicate + - `merge()`: Merge multiple GPX instances into one + - `split()`: Split track segments at time and/or distance gaps + - `simplify()`: Simplify the tracks and routes of a GPX with the Ramer-Douglas-Peucker algorithm + - `smooth()`: Smooth the coordinates and/or elevations of the tracks and routes of a GPX with a moving average + - `shift_time()`: Shift all point timestamps of a GPX by a time delta + - `strip_extensions()`: Strip all extensions from a GPX + - All operations are pure (they return a new `GPX` instance and never mutate the input) and are importable directly from the top-level package (e.g. `from gpx import crop, merge`). +- New `gpx edit` CLI options exposing the new operations: + - `--split-time-gap SECONDS` and `--split-distance-gap METERS`: Split track segments at gaps + - `--simplify TOLERANCE`: Simplify tracks and routes (tolerance in metres) + - `--smooth WINDOW`: Smooth track and route coordinates and elevations with a moving average + - `--shift-time SECONDS`: Shift all point timestamps (may be negative) + - `--strip-extensions`: Strip all extensions +- New file conversion functions in the `io` module, exposing the operation behind the `gpx convert` CLI command as a public, reusable API: + - `convert_file()`: Convert a file between the GPX, GeoJSON and KML file formats + - `detect_format()`: Detect the file format from a file path's extension + - Both are importable directly from the top-level package (e.g. `from gpx import convert_file`). + +### Changed + +- The CLI (`gpx edit`, `gpx merge` and `gpx convert`) now uses the new `operations` module and `io` conversion functions internally (behavior is unchanged). + ## [2026.3.0](https://github.com/sgraaf/gpx/compare/2026.2.0...2026.3.0) - 2026-05-17 This third release in the year 2026 adds a new `GeoGPXModel` base class for GPX models that carry geometric data. diff --git a/CLAUDE.md b/CLAUDE.md index 26fb6d8..d74781b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -36,7 +36,8 @@ gpx/ │ ├── types.py # Custom types: Latitude, Longitude, Degrees, Fix, DGPSStation │ ├── utils.py # Utility functions for XML parsing and serialization │ ├── convert.py # Conversion functions: from_geo_interface, from_wkb, from_wkt -│ ├── io.py # I/O functions: read_gpx, read_geojson, read_kml +│ ├── io.py # I/O functions: read_gpx, read_geojson, read_kml, convert_file, detect_format +│ ├── operations.py # Edit/merge operations: crop, trim, reverse, strip_metadata, strip_extensions, reduce_precision, filter_points, split, simplify, smooth, shift_time, merge │ ├── cli.py # Command-line interface implementation │ └── py.typed # PEP 561 marker for type hints ├── docs/ # Sphinx documentation (MyST Markdown) @@ -57,6 +58,7 @@ gpx/ │ ├── test_utils.py # Utilities tests │ ├── test_waypoint.py # Waypoint model tests │ ├── test_io_convert.py # IO and conversion tests (read/write/convert) +│ ├── test_operations.py # Edit/merge operations tests │ ├── test_cli.py # CLI tests │ ├── test_main.py # Main entry point tests │ ├── test_extensions.py # Extensions tests diff --git a/README.md b/README.md index 0138a3e..dfe1ddf 100644 --- a/README.md +++ b/README.md @@ -297,6 +297,22 @@ gpx = read_geojson("path/to/file.geojson") gpx = read_kml("path/to/file.kml") ``` +### Converting between file formats + +*gpx* can convert files between the GPX, GeoJSON and KML file formats directly: + +```python +from gpx import convert_file + + +# Formats are auto-detected from the file extensions... +convert_file("input.gpx", "output.geojson") +convert_file("input.kml", "output.gpx") + +# ... or can be given explicitly +convert_file("input.json", "output.gpx", input_format="geojson") +``` + ### Converting from data formats *gpx* can convert from data formats (strings, bytes, objects): @@ -326,6 +342,70 @@ geojson = {"type": "Point", "coordinates": [4.9041, 52.3676]} gpx = from_geo_interface(geojson) ``` +### Editing and merging GPX data + +*gpx* provides operations for editing and merging GPX data. All operations are pure: they return a new `GPX` instance and never mutate the input: + +```python +import datetime as dt + +from gpx import ( + crop, + filter_points, + merge, + read_gpx, + reduce_precision, + reverse, + shift_time, + simplify, + smooth, + split, + strip_extensions, + strip_metadata, + trim, +) + + +gpx = read_gpx("path/to/file.gpx") + +# Crop to a geographic bounding box +cropped = crop(gpx, min_lat=52.0, max_lat=53.0, min_lon=4.0, max_lon=5.0) + +# Trim to a date/time range +trimmed = trim(gpx, start=dt.datetime(2024, 1, 1, 10, 0, 0, tzinfo=dt.UTC)) + +# Reverse routes and/or tracks +reversed_gpx = reverse(gpx) + +# Strip metadata (fields) +anonymous = strip_metadata(gpx, author=True, copyright=True) +bare = strip_metadata(gpx) # removes all metadata + +# Reduce coordinate and/or elevation precision +reduced = reduce_precision(gpx, coordinate_precision=6, elevation_precision=1) + +# Filter points with an arbitrary predicate +with_elevation = filter_points(gpx, lambda point: point.ele is not None) + +# Split track segments at time and/or distance gaps +split_gpx = split(gpx, time_gap=dt.timedelta(minutes=10)) + +# Simplify tracks and routes (Ramer-Douglas-Peucker, tolerance in metres) +simplified = simplify(gpx, tolerance=10.0) + +# Smooth track and route coordinates and elevations (moving average) +smoothed = smooth(gpx, window=5) + +# Shift all point timestamps (e.g., to fix timezone mistakes or clock drift) +shifted = shift_time(gpx, dt.timedelta(hours=-2)) + +# Strip all extensions (e.g., heart rate, cadence, temperature) +without_extensions = strip_extensions(gpx) + +# Merge multiple GPX instances into one +merged = merge([read_gpx("one.gpx"), read_gpx("two.gpx")]) +``` + ### Command-Line Interface *gpx* provides a command-line interface (CLI) for common GPX operations: @@ -344,6 +424,10 @@ gpx edit input.gpx -o output.gpx --min-lat 52.0 --max-lat 53.0 gpx edit input.gpx -o output.gpx --start 2024-01-01T10:00:00 --end 2024-01-01T12:00:00 gpx edit input.gpx -o output.gpx --precision 5 --elevation-precision 1 gpx edit input.gpx -o output.gpx --strip-all-metadata +gpx edit input.gpx -o output.gpx --split-time-gap 600 +gpx edit input.gpx -o output.gpx --simplify 10 --smooth 5 +gpx edit input.gpx -o output.gpx --shift-time -7200 +gpx edit input.gpx -o output.gpx --strip-extensions # Merge multiple GPX files gpx merge file1.gpx file2.gpx file3.gpx -o merged.gpx diff --git a/docs/api.md b/docs/api.md index 03b0841..45fdc8b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -124,6 +124,13 @@ :members: ``` +## `gpx.operations` Module + +```{eval-rst} +.. automodule:: gpx.operations + :members: +``` + ## `gpx.utils` Module ```{eval-rst} diff --git a/docs/cli.md b/docs/cli.md index 1a485ba..7e30805 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -116,10 +116,13 @@ Running `gpx edit --help` or `python -m gpx edit --help` shows a list of all of usage: gpx edit [-h] -o [--min-lat LATITUDE] [--max-lat LATITUDE] [--min-lon LONGITUDE] [--max-lon LONGITUDE] [--start DATETIME] [--end DATETIME] - [--reverse] [--reverse-routes] [--reverse-tracks] - [--strip-name] [--strip-desc] [--strip-author] - [--strip-copyright] [--strip-time] [--strip-keywords] - [--strip-links] [--strip-all-metadata] [--precision DIGITS] + [--split-time-gap SECONDS] [--split-distance-gap METERS] + [--simplify TOLERANCE] [--smooth WINDOW] + [--shift-time SECONDS] [--reverse] [--reverse-routes] + [--reverse-tracks] [--strip-name] [--strip-desc] + [--strip-author] [--strip-copyright] [--strip-time] + [--strip-keywords] [--strip-links] [--strip-all-metadata] + [--strip-extensions] [--precision DIGITS] [--elevation-precision DIGITS] @@ -150,6 +153,35 @@ trim options: --end DATETIME End datetime (ISO 8601 format, e.g., 2024-01-01T12:00:00) +split options: + Split track segments at gaps + + --split-time-gap SECONDS + Split track segments where the time between + consecutive points exceeds this many seconds + --split-distance-gap METERS + Split track segments where the distance between + consecutive points exceeds this many metres + +simplify options: + Simplify tracks and routes + + --simplify TOLERANCE Simplify tracks and routes with the Ramer-Douglas- + Peucker algorithm using this tolerance (in metres) + +smooth options: + Smooth tracks and routes + + --smooth WINDOW Smooth track and route coordinates and elevations with + a moving average over this many points (an odd integer + of at least 3) + +time shift options: + Shift timestamps + + --shift-time SECONDS Shift all point timestamps by this many seconds (may + be negative) + reverse options: Reverse routes and/or tracks @@ -168,6 +200,7 @@ strip options: --strip-keywords Strip metadata keywords --strip-links Strip metadata links --strip-all-metadata Strip all metadata + --strip-extensions Strip all extensions precision options: Reduce coordinate precision diff --git a/pyproject.toml b/pyproject.toml index 1e59a17..008bcb2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -100,5 +100,8 @@ pydocstyle.convention = "google" "PLR2004", # magic-value-comparison (Pylint) ] +[tool.pyrefly] +search_path = ["src", "."] + [tool.mypy] python_version = "3.11" diff --git a/src/gpx/__init__.py b/src/gpx/__init__.py index 12620fc..711afe7 100644 --- a/src/gpx/__init__.py +++ b/src/gpx/__init__.py @@ -7,9 +7,23 @@ from .email import Email from .extensions import Extensions from .gpx import GPX -from .io import read_geojson, read_gpx, read_kml +from .io import convert_file, detect_format, read_geojson, read_gpx, read_kml from .link import Link from .metadata import Metadata +from .operations import ( + crop, + filter_points, + merge, + reduce_precision, + reverse, + shift_time, + simplify, + smooth, + split, + strip_extensions, + strip_metadata, + trim, +) from .person import Person from .route import Route from .track import Track @@ -37,11 +51,25 @@ "Track", "TrackSegment", "Waypoint", + "convert_file", + "crop", + "detect_format", + "filter_points", "from_geo_interface", "from_string", "from_wkb", "from_wkt", + "merge", "read_geojson", "read_gpx", "read_kml", + "reduce_precision", + "reverse", + "shift_time", + "simplify", + "smooth", + "split", + "strip_extensions", + "strip_metadata", + "trim", ] diff --git a/src/gpx/cli.py b/src/gpx/cli.py index 1e5799e..d7b0d67 100644 --- a/src/gpx/cli.py +++ b/src/gpx/cli.py @@ -11,19 +11,29 @@ import datetime as dt import json import sys -from dataclasses import replace -from decimal import Decimal from importlib import metadata from pathlib import Path from typing import TYPE_CHECKING -from .gpx import GPX -from .io import read_geojson, read_gpx, read_kml -from .types import Latitude, Longitude +from .io import convert_file, read_gpx +from .operations import ( + crop, + reduce_precision, + reverse, + shift_time, + simplify, + smooth, + split, + strip_extensions, + strip_metadata, + trim, +) +from .operations import merge as merge_op if TYPE_CHECKING: - from collections.abc import Callable, Sequence + from collections.abc import Sequence + from .gpx import GPX from .route import Route from .track import Track from .waypoint import Waypoint @@ -89,30 +99,25 @@ def _create_parser() -> argparse.ArgumentParser: return parser -# ============================================================================= -# Validate command -# ============================================================================= - - def _add_validate_parser( subparsers: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: """Add the validate subcommand parser.""" - parser = subparsers.add_parser( + validate_parser = subparsers.add_parser( "validate", help="Validate a GPX file", description="Validate a GPX file by attempting to parse it.", ) - parser.add_argument( + validate_parser.add_argument( "input_file", type=Path, help="Path to the input GPX file", metavar="", ) - parser.set_defaults(func=_cmd_validate) + validate_parser.set_defaults(func=validate) -def _cmd_validate(args: argparse.Namespace) -> int: +def validate(args: argparse.Namespace) -> int: """Execute the validate command.""" file_path: Path = args.input_file @@ -135,35 +140,30 @@ def _cmd_validate(args: argparse.Namespace) -> int: return 0 -# ============================================================================= -# Info command -# ============================================================================= - - def _add_info_parser( subparsers: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: """Add the info subcommand parser.""" - parser = subparsers.add_parser( + info_parser = subparsers.add_parser( "info", help="Show information and statistics about a GPX file", description="Display detailed information and statistics about a GPX file.", ) - parser.add_argument( + info_parser.add_argument( "input_file", type=Path, help="Path to the input GPX file", metavar="", ) - parser.add_argument( + info_parser.add_argument( "--json", action="store_true", help="Output information in JSON format", ) - parser.set_defaults(func=_cmd_info) + info_parser.set_defaults(func=info) -def _cmd_info(args: argparse.Namespace) -> int: +def info(args: argparse.Namespace) -> int: """Execute the info command.""" file_path: Path = args.input_file @@ -408,28 +408,23 @@ def _print_gpx_info( # noqa: C901, PLR0912, PLR0915 print(f" Total Duration: {hours:02d}:{minutes:02d}:{seconds:02d}") -# ============================================================================= -# Edit command -# ============================================================================= - - def _add_edit_parser( subparsers: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: """Add the edit subcommand parser.""" - parser = subparsers.add_parser( + edit_parser = subparsers.add_parser( "edit", help="Edit a GPX file with various transformations", description="Edit a GPX file with various transformations like cropping, " "trimming, reversing, and stripping metadata.", ) - parser.add_argument( + edit_parser.add_argument( "input_file", type=Path, help="Path to the input GPX file", metavar="", ) - parser.add_argument( + edit_parser.add_argument( "-o", "--output-file", type=Path, @@ -440,7 +435,7 @@ def _add_edit_parser( ) # Crop options - crop_group = parser.add_argument_group( + crop_group = edit_parser.add_argument_group( "crop options", "Crop to a geographic bounding box" ) crop_group.add_argument( @@ -469,7 +464,9 @@ def _add_edit_parser( ) # Trim options - trim_group = parser.add_argument_group("trim options", "Trim to a date/time range") + trim_group = edit_parser.add_argument_group( + "trim options", "Trim to a date/time range" + ) trim_group.add_argument( "--start", type=str, @@ -483,8 +480,62 @@ def _add_edit_parser( help="End datetime (ISO 8601 format, e.g., 2024-01-01T12:00:00)", ) + # Split options + split_group = edit_parser.add_argument_group( + "split options", "Split track segments at gaps" + ) + split_group.add_argument( + "--split-time-gap", + type=float, + metavar="SECONDS", + help="Split track segments where the time between consecutive points " + "exceeds this many seconds", + ) + split_group.add_argument( + "--split-distance-gap", + type=float, + metavar="METERS", + help="Split track segments where the distance between consecutive points " + "exceeds this many metres", + ) + + # Simplify options + simplify_group = edit_parser.add_argument_group( + "simplify options", "Simplify tracks and routes" + ) + simplify_group.add_argument( + "--simplify", + type=float, + metavar="TOLERANCE", + help="Simplify tracks and routes with the Ramer-Douglas-Peucker algorithm " + "using this tolerance (in metres)", + ) + + # Smooth options + smooth_group = edit_parser.add_argument_group( + "smooth options", "Smooth tracks and routes" + ) + smooth_group.add_argument( + "--smooth", + type=int, + metavar="WINDOW", + help="Smooth track and route coordinates and elevations with a moving " + "average over this many points (an odd integer of at least 3)", + ) + + # Time shift options + shift_group = edit_parser.add_argument_group( + "time shift options", "Shift timestamps" + ) + shift_group.add_argument( + "--shift-time", + type=float, + metavar="SECONDS", + help="Shift all point timestamps by this many seconds (may be negative)", + ) + # Reverse options - reverse_group = parser.add_argument_group( + reverse_group = edit_parser.add_argument_group( "reverse options", "Reverse routes and/or tracks" ) reverse_group.add_argument( @@ -504,7 +555,9 @@ def _add_edit_parser( ) # Strip metadata options - strip_group = parser.add_argument_group("strip options", "Strip metadata fields") + strip_group = edit_parser.add_argument_group( + "strip options", "Strip metadata fields" + ) strip_group.add_argument( "--strip-name", action="store_true", @@ -545,9 +598,14 @@ def _add_edit_parser( action="store_true", help="Strip all metadata", ) + strip_group.add_argument( + "--strip-extensions", + action="store_true", + help="Strip all extensions", + ) # Precision options - precision_group = parser.add_argument_group( + precision_group = edit_parser.add_argument_group( "precision options", "Reduce coordinate precision" ) precision_group.add_argument( @@ -563,10 +621,10 @@ def _add_edit_parser( help="Number of decimal places for elevation (e.g., 1)", ) - parser.set_defaults(func=_cmd_edit) + edit_parser.set_defaults(func=edit) -def _cmd_edit(args: argparse.Namespace) -> int: +def edit(args: argparse.Namespace) -> int: # noqa: C901, PLR0912 """Execute the edit command.""" input_path: Path = args.input_file output_path: Path = args.output_file @@ -581,32 +639,65 @@ def _cmd_edit(args: argparse.Namespace) -> int: if any( v is not None for v in (args.min_lat, args.max_lat, args.min_lon, args.max_lon) ): - gpx = _apply_crop(gpx, args.min_lat, args.max_lat, args.min_lon, args.max_lon) + gpx = crop( + gpx, + min_lat=args.min_lat, + max_lat=args.max_lat, + min_lon=args.min_lon, + max_lon=args.max_lon, + ) # Apply trim if args.start or args.end: start_dt = _parse_datetime(args.start) if args.start else None end_dt = _parse_datetime(args.end) if args.end else None - gpx = _apply_trim(gpx, start_dt, end_dt) + gpx = trim(gpx, start=start_dt, end=end_dt) + + # Apply split + if args.split_time_gap is not None or args.split_distance_gap is not None: + gpx = split( + gpx, + time_gap=dt.timedelta(seconds=args.split_time_gap) + if args.split_time_gap is not None + else None, + distance_gap=args.split_distance_gap, + ) + + # Apply simplify + if args.simplify is not None: + gpx = simplify(gpx, args.simplify) + + # Apply smooth + if args.smooth is not None: + gpx = smooth(gpx, window=args.smooth) + + # Apply time shift + if args.shift_time is not None: + gpx = shift_time(gpx, dt.timedelta(seconds=args.shift_time)) # Apply reverse if args.reverse: - gpx = _apply_reverse(gpx, reverse_routes=True, reverse_tracks=True) + gpx = reverse(gpx, routes=True, tracks=True) else: if args.reverse_routes: - gpx = _apply_reverse(gpx, reverse_routes=True, reverse_tracks=False) + gpx = reverse(gpx, routes=True, tracks=False) if args.reverse_tracks: - gpx = _apply_reverse(gpx, reverse_routes=False, reverse_tracks=True) + gpx = reverse(gpx, routes=False, tracks=True) # Apply strip metadata - if args.strip_all_metadata: - gpx = replace(gpx, metadata=None) - elif gpx.metadata: - gpx = _apply_strip_metadata(gpx, args) + gpx = _apply_strip_metadata(gpx, args) + + # Apply strip extensions + if args.strip_extensions: + gpx = strip_extensions(gpx) # Apply precision reduction if args.precision is not None or args.elevation_precision is not None: - gpx = _apply_precision(gpx, args.precision, args.elevation_precision) + gpx = reduce_precision( + gpx, + coordinate_precision=args.precision, + elevation_precision=args.elevation_precision, + ) # Write output gpx.write_gpx(output_path) @@ -615,26 +706,22 @@ def _cmd_edit(args: argparse.Namespace) -> int: def _apply_strip_metadata(gpx: GPX, args: argparse.Namespace) -> GPX: - """Apply metadata stripping based on args.""" - metadata = gpx.metadata - if not metadata: + """Apply metadata stripping based on the command-line arguments.""" + if args.strip_all_metadata: + return strip_metadata(gpx) + + fields = { + "name": args.strip_name, + "desc": args.strip_desc, + "author": args.strip_author, + "copyright": args.strip_copyright, + "time": args.strip_time, + "keywords": args.strip_keywords, + "links": args.strip_links, + } + if not any(fields.values()): return gpx - - if args.strip_name: - metadata = replace(metadata, name=None) - if args.strip_desc: - metadata = replace(metadata, desc=None) - if args.strip_author: - metadata = replace(metadata, author=None) - if args.strip_copyright: - metadata = replace(metadata, copyright=None) - if args.strip_time: - metadata = replace(metadata, time=None) - if args.strip_keywords: - metadata = replace(metadata, keywords=None) - if args.strip_links: - metadata = replace(metadata, link=[]) - return replace(gpx, metadata=metadata) + return strip_metadata(gpx, **fields) def _parse_datetime(dt_str: str) -> dt.datetime: @@ -665,151 +752,6 @@ def _parse_datetime(dt_str: str) -> dt.datetime: raise ValueError(msg) -def _filter_points(gpx: GPX, predicate: Callable[[Waypoint], bool]) -> GPX: - """Return a new GPX with each point list filtered by ``predicate``. - - Routes / segments / tracks that end up empty are dropped. - """ - new_wpt = [w for w in gpx.wpt if predicate(w)] - - new_rte: list[Route] = [] - for route in gpx.rte: - kept = [p for p in route.rtept if predicate(p)] - if kept: - new_rte.append(replace(route, rtept=kept)) - - new_trk: list[Track] = [] - for track in gpx.trk: - new_trkseg = [ - replace(segment, trkpt=kept) - for segment in track.trkseg - if (kept := [p for p in segment.trkpt if predicate(p)]) - ] - if new_trkseg: - new_trk.append(replace(track, trkseg=new_trkseg)) - - return replace(gpx, wpt=new_wpt, rte=new_rte, trk=new_trk) - - -def _apply_crop( - gpx: GPX, - min_lat: float | None, - max_lat: float | None, - min_lon: float | None, - max_lon: float | None, -) -> GPX: - """Apply geographic crop to GPX data.""" - - def is_in_bounds(point: Waypoint) -> bool: - lat, lon = float(point.lat), float(point.lon) - if min_lat is not None and lat < min_lat: - return False - if max_lat is not None and lat > max_lat: - return False - if min_lon is not None and lon < min_lon: - return False - return not (max_lon is not None and lon > max_lon) - - return _filter_points(gpx, is_in_bounds) - - -def _apply_trim( - gpx: GPX, - start_dt: dt.datetime | None, - end_dt: dt.datetime | None, -) -> GPX: - """Apply time-based trim to GPX data.""" - - def is_in_time_range(point: Waypoint) -> bool: - if point.time is None: - return True # Keep points without timestamps - if start_dt is not None and point.time < start_dt: - return False - return not (end_dt is not None and point.time > end_dt) - - return _filter_points(gpx, is_in_time_range) - - -def _apply_reverse( - gpx: GPX, - *, - reverse_routes: bool, - reverse_tracks: bool, -) -> GPX: - """Apply reversal to routes and/or tracks.""" - new_rte = gpx.rte - new_trk = gpx.trk - - if reverse_routes: - new_rte = [ - replace(route, rtept=list(reversed(route.rtept))) for route in gpx.rte - ] - - if reverse_tracks: - new_trk = [ - replace( - track, - trkseg=[ - replace(segment, trkpt=list(reversed(segment.trkpt))) - for segment in reversed(track.trkseg) - ], - ) - for track in gpx.trk - ] - - return replace(gpx, rte=new_rte, trk=new_trk) - - -def _apply_precision( - gpx: GPX, - coord_precision: int | None, - elevation_precision: int | None, -) -> GPX: - """Apply precision reduction to coordinates and elevations.""" - - def round_point(point: Waypoint) -> Waypoint: - if coord_precision is None and elevation_precision is None: - return point - new_lat = ( - Latitude(str(round(float(point.lat), coord_precision))) - if coord_precision is not None - else point.lat - ) - new_lon = ( - Longitude(str(round(float(point.lon), coord_precision))) - if coord_precision is not None - else point.lon - ) - new_ele = ( - Decimal(str(round(float(point.ele), elevation_precision))) - if elevation_precision is not None and point.ele is not None - else point.ele - ) - return replace(point, lat=new_lat, lon=new_lon, ele=new_ele) - - new_wpt = [round_point(w) for w in gpx.wpt] - new_rte = [ - replace(route, rtept=[round_point(p) for p in route.rtept]) for route in gpx.rte - ] - new_trk = [ - replace( - track, - trkseg=[ - replace(segment, trkpt=[round_point(p) for p in segment.trkpt]) - for segment in track.trkseg - ], - ) - for track in gpx.trk - ] - - return replace(gpx, wpt=new_wpt, rte=new_rte, trk=new_trk) - - -# ============================================================================= -# Merge command -# ============================================================================= - - def _add_merge_parser( subparsers: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: @@ -835,10 +777,10 @@ def _add_merge_parser( metavar="", dest="output_file", ) - parser.set_defaults(func=_cmd_merge) + parser.set_defaults(func=merge) -def _cmd_merge(args: argparse.Namespace) -> int: +def merge(args: argparse.Namespace) -> int: """Execute the merge command.""" files: list[Path] = args.input_files output_path: Path = args.output_file @@ -849,39 +791,18 @@ def _cmd_merge(args: argparse.Namespace) -> int: print(f"Error: File not found: {file_path}", file=sys.stderr) return 1 - # Read all GPX files - gpx_files = [read_gpx(file_path) for file_path in files] - - # Merge - merged_wpt: list[Waypoint] = [] - merged_rte: list[Route] = [] - merged_trk: list[Track] = [] - - for gpx in gpx_files: - merged_wpt.extend(gpx.wpt) - merged_rte.extend(gpx.rte) - merged_trk.extend(gpx.trk) - - merged_gpx = GPX( - wpt=merged_wpt, - rte=merged_rte, - trk=merged_trk, - ) + # Read and merge all GPX files + merged_gpx = merge_op(read_gpx(file_path) for file_path in files) # Write output merged_gpx.write_gpx(output_path) print(f"Merged {len(files)} files into: {output_path}") - print(f" Waypoints: {len(merged_wpt)}") - print(f" Routes: {len(merged_rte)}") - print(f" Tracks: {len(merged_trk)}") + print(f" Waypoints: {len(merged_gpx.wpt)}") + print(f" Routes: {len(merged_gpx.rte)}") + print(f" Tracks: {len(merged_gpx.trk)}") return 0 -# ============================================================================= -# Convert command -# ============================================================================= - - def _add_convert_parser( subparsers: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: @@ -920,10 +841,10 @@ def _add_convert_parser( choices=["gpx", "geojson", "kml"], help="Output format (default: auto-detect from file extension)", ) - parser.set_defaults(func=_cmd_convert) + parser.set_defaults(func=convert) -def _cmd_convert(args: argparse.Namespace) -> int: +def convert(args: argparse.Namespace) -> int: """Execute the convert command.""" input_path: Path = args.input_file output_path: Path = args.output_file @@ -932,72 +853,20 @@ def _cmd_convert(args: argparse.Namespace) -> int: print(f"Error: File not found: {input_path}", file=sys.stderr) return 1 - # Determine input format - input_format = args.from_format or _detect_format(input_path) - if not input_format: - print( - f"Error: Could not detect input format for: {input_path}", file=sys.stderr - ) - return 1 - - # Determine output format - output_format = args.to_format or _detect_format(output_path) - if not output_format: - print( - f"Error: Could not detect output format for: {output_path}", file=sys.stderr + try: + input_format, output_format = convert_file( + input_path, + output_path, + input_format=args.from_format, + output_format=args.to_format, ) - return 1 - - # Read input - gpx = _read_input(input_path, input_format) - if gpx is None: - print(f"Error: Unsupported input format: {input_format}", file=sys.stderr) - return 1 - - # Write output - if not _write_output(gpx, output_path, output_format): - print(f"Error: Unsupported output format: {output_format}", file=sys.stderr) + except ValueError as e: + print(f"Error: {e}", file=sys.stderr) return 1 print(f"Converted {input_path} ({input_format}) to {output_path} ({output_format})") return 0 -def _read_input(input_path: Path, input_format: str) -> GPX | None: - """Read input file based on format.""" - if input_format == "gpx": - return read_gpx(input_path) - if input_format == "geojson": - return read_geojson(input_path) - if input_format == "kml": - return read_kml(input_path) - return None - - -def _write_output(gpx: GPX, output_path: Path, output_format: str) -> bool: - """Write output file based on format.""" - if output_format == "gpx": - gpx.write_gpx(output_path) - elif output_format == "geojson": - gpx.write_geojson(output_path) - elif output_format == "kml": - gpx.write_kml(output_path) - else: - return False - return True - - -def _detect_format(path: Path) -> str | None: - """Detect file format from extension.""" - suffix = path.suffix.lower() - format_map = { - ".gpx": "gpx", - ".geojson": "geojson", - ".json": "geojson", - ".kml": "kml", - } - return format_map.get(suffix) - - if __name__ == "__main__": sys.exit(cli()) diff --git a/src/gpx/convert.py b/src/gpx/convert.py index f475cea..047dc17 100644 --- a/src/gpx/convert.py +++ b/src/gpx/convert.py @@ -171,11 +171,6 @@ def from_wkt(wkt: str, *, creator: str | None = None) -> GPX: return _wkt_geometry_to_gpx(geometry, gpx_kwargs) -# ============================================================================= -# GeoJSON Helper Functions -# ============================================================================= - - def _feature_collection_to_gpx(fc: dict[str, Any], gpx_kwargs: dict[str, Any]) -> GPX: """Convert a GeoJSON FeatureCollection to GPX.""" waypoints: list[Waypoint] = [] @@ -338,11 +333,6 @@ def _coords_to_track( return Track(**kwargs) -# ============================================================================= -# WKB Helper Functions -# ============================================================================= - - def _parse_wkb_geometry(wkb: bytes, offset: int) -> tuple[dict[str, Any], int]: """Parse a WKB geometry and return geometry dict and new offset.""" if offset >= len(wkb): @@ -586,11 +576,6 @@ def _wkb_geometry_to_gpx(geometry: dict[str, Any], gpx_kwargs: dict[str, Any]) - return _geometry_to_gpx(geometry, gpx_kwargs) -# ============================================================================= -# WKT Helper Functions -# ============================================================================= - - def _parse_wkt(wkt: str) -> dict[str, Any]: # noqa: PLR0911 """Parse WKT string into geometry dict.""" wkt = wkt.strip() @@ -737,7 +722,7 @@ def _parse_wkt_multipolygon(coords_part: str, has_z: bool) -> dict[str, Any]: # parsed_rings.append(coords) polygons.append(parsed_rings) current = [] - elif depth > 1 or (depth == 1 and char not in ","): + elif depth > 1 or (depth == 1 and char != ","): current.append(char) return {"type": "MultiPolygon", "coordinates": polygons} diff --git a/src/gpx/extensions.py b/src/gpx/extensions.py index ad4d5ed..233580a 100644 --- a/src/gpx/extensions.py +++ b/src/gpx/extensions.py @@ -157,7 +157,7 @@ def find( # Try to match against the element tag for prefix, uri in namespaces.items(): if path.startswith(f"{prefix}:"): - local_name = path.split(":")[1].split("/")[0] + local_name = path.split(":")[1].split("/", maxsplit=1)[0] if elem.tag == f"{{{uri}}}{local_name}": # If there's more path, continue searching remaining = "/".join(path.split("/")[1:]) @@ -227,7 +227,7 @@ def get_text( for elem in self.elements: for child in elem.iter(): if _matches_tag(child.tag, tag, ns_tag): - return child.text if child.text else default + return child.text or default return default def get_int( diff --git a/src/gpx/gpx.py b/src/gpx/gpx.py index c97c966..d66428f 100644 --- a/src/gpx/gpx.py +++ b/src/gpx/gpx.py @@ -246,10 +246,6 @@ def to_string(self, *, pretty_print: bool = True) -> str: xml_bytes = ET.tostring(element, encoding="utf-8", xml_declaration=True) return xml_bytes.decode("utf-8") - # ========================================================================= - # GPX file methods - # ========================================================================= - def write_gpx(self, file_path: str | Path, *, pretty_print: bool = True) -> None: """Write the GPX to a file. @@ -270,10 +266,6 @@ def write_gpx(self, file_path: str | Path, *, pretty_print: bool = True) -> None tree = ET.ElementTree(element) tree.write(str(file_path), xml_declaration=True, encoding="utf-8") - # ========================================================================= - # GeoJSON file methods - # ========================================================================= - def write_geojson(self, file_path: str | Path, *, indent: int | None = 2) -> None: """Write the GPX to a GeoJSON file. @@ -290,10 +282,6 @@ def write_geojson(self, file_path: str | Path, *, indent: int | None = 2) -> Non with Path(file_path).open("w", encoding="utf-8") as f: json.dump(self.__geo_interface__, f, indent=indent) - # ========================================================================= - # KML file methods - # ========================================================================= - def write_kml(self, file_path: str | Path, *, pretty_print: bool = True) -> None: """Write the GPX to a KML file. @@ -349,10 +337,6 @@ def write_kml(self, file_path: str | Path, *, pretty_print: bool = True) -> None with Path(file_path).open("w", encoding="utf-8") as f: f.write(kml_str) - # ========================================================================= - # WKT conversion methods - # ========================================================================= - def to_wkt(self) -> str: """Convert the GPX to a WKT (Well-Known Text) string. @@ -394,10 +378,6 @@ def to_wkt(self) -> str: return geometries[0] return f"GEOMETRYCOLLECTION ({', '.join(geometries)})" - # ========================================================================= - # WKB conversion methods - # ========================================================================= - def to_wkb(self, *, byte_order: str = "little") -> bytes: """Convert the GPX to WKB (Well-Known Binary). diff --git a/src/gpx/io.py b/src/gpx/io.py index 91fc1c9..7d3a06b 100644 --- a/src/gpx/io.py +++ b/src/gpx/io.py @@ -1,7 +1,7 @@ -"""I/O functions for reading file formats into GPX. +"""I/O functions for reading and converting file formats. This module provides functions for reading GPX, GeoJSON, and KML files from disk -and returning GPX objects. +and returning GPX objects, as well as for converting files between these formats. """ from __future__ import annotations @@ -112,9 +112,104 @@ def read_kml(file_path: str | Path, *, creator: str | None = None) -> GPX: return GPX(wpt=waypoints, rte=routes, trk=tracks, **gpx_kwargs) -# ============================================================================= -# KML Helper Functions -# ============================================================================= +def detect_format(file_path: str | Path) -> str | None: + """Detect the file format from a file path's extension. + + Args: + file_path: The file path to detect the format of. + + Returns: + The detected format ("gpx", "geojson" or "kml"), or None if the + extension is not recognized. + + Example: + >>> from gpx import detect_format + >>> detect_format("route.kml") + 'kml' + + """ + suffix = Path(file_path).suffix.lower() + format_map = { + ".gpx": "gpx", + ".geojson": "geojson", + ".json": "geojson", + ".kml": "kml", + } + return format_map.get(suffix) + + +def convert_file( + input_file: str | Path, + output_file: str | Path, + *, + input_format: str | None = None, + output_format: str | None = None, +) -> tuple[str, str]: + """Convert a file between the GPX, GeoJSON and KML file formats. + + Args: + input_file: Path to the input file. + output_file: Path to the output file. + input_format: The input format ("gpx", "geojson" or "kml"). + Defaults to None (auto-detect from the file extension). + output_format: The output format ("gpx", "geojson" or "kml"). + Defaults to None (auto-detect from the file extension). + + Returns: + A tuple of the input and output formats used. + + Raises: + ValueError: If a format is not given and cannot be detected, or if + a format is not supported. + + Example: + >>> from gpx import convert_file + >>> convert_file("input.gpx", "output.geojson") + ('gpx', 'geojson') + + """ + input_path = Path(input_file) + output_path = Path(output_file) + + input_format = input_format or detect_format(input_path) + if not input_format: + msg = f"Could not detect input format for: {input_path}" + raise ValueError(msg) + + output_format = output_format or detect_format(output_path) + if not output_format: + msg = f"Could not detect output format for: {output_path}" + raise ValueError(msg) + + gpx = _read_file(input_path, input_format) + _write_file(gpx, output_path, output_format) + + return input_format, output_format + + +def _read_file(file_path: Path, file_format: str) -> GPX: + """Read a file in the given format and return a GPX object.""" + if file_format == "gpx": + return read_gpx(file_path) + if file_format == "geojson": + return read_geojson(file_path) + if file_format == "kml": + return read_kml(file_path) + msg = f"Unsupported input format: {file_format}" + raise ValueError(msg) + + +def _write_file(gpx: GPX, file_path: Path, file_format: str) -> None: + """Write a GPX object to a file in the given format.""" + if file_format == "gpx": + gpx.write_gpx(file_path) + elif file_format == "geojson": + gpx.write_geojson(file_path) + elif file_format == "kml": + gpx.write_kml(file_path) + else: + msg = f"Unsupported output format: {file_format}" + raise ValueError(msg) def _find_kml_element(parent: ET.Element, tag: str) -> ET.Element | None: diff --git a/src/gpx/operations.py b/src/gpx/operations.py new file mode 100644 index 0000000..9e1b073 --- /dev/null +++ b/src/gpx/operations.py @@ -0,0 +1,699 @@ +"""Operations for editing and merging GPX models. + +This module provides the operations behind the ``gpx edit`` and ``gpx merge`` +CLI commands as a reusable, importable API. + +All operations are pure: they never mutate the input, but return a new +:class:`~gpx.gpx.GPX` instance instead. Namespace mappings (``nsmap``) and +extensions are preserved. +""" + +from __future__ import annotations + +from dataclasses import replace +from decimal import Decimal +from math import cos, radians, sqrt +from statistics import fmean +from typing import TYPE_CHECKING + +from .gpx import GPX +from .types import Latitude, Longitude + +if TYPE_CHECKING: + import datetime as dt + from collections.abc import Callable, Iterable + + from .route import Route + from .track import Track + from .track_segment import TrackSegment + from .waypoint import Waypoint + + +def filter_points(gpx: GPX, predicate: Callable[[Waypoint], bool]) -> GPX: + """Return a new GPX with each point list filtered by ``predicate``. + + Waypoints, route points and track points for which ``predicate`` returns + False are dropped. Routes, track segments and tracks that end up empty + are dropped as well. + + Args: + gpx: The GPX instance to filter. + predicate: A callable that returns True for points to keep. + + Returns: + A new GPX instance with the filtered points. + + Example: + >>> from gpx import filter_points, read_gpx + >>> gpx = read_gpx("path/to/file.gpx") + >>> with_ele = filter_points(gpx, lambda point: point.ele is not None) + + """ + new_wpt = [w for w in gpx.wpt if predicate(w)] + + new_rte: list[Route] = [] + for route in gpx.rte: + kept = [p for p in route.rtept if predicate(p)] + if kept: + new_rte.append(replace(route, rtept=kept)) + + new_trk: list[Track] = [] + for track in gpx.trk: + new_trkseg = [ + replace(segment, trkpt=kept) + for segment in track.trkseg + if (kept := [p for p in segment.trkpt if predicate(p)]) + ] + if new_trkseg: + new_trk.append(replace(track, trkseg=new_trkseg)) + + return replace(gpx, wpt=new_wpt, rte=new_rte, trk=new_trk) + + +def _map_points(gpx: GPX, transform: Callable[[Waypoint], Waypoint]) -> GPX: + """Return a new GPX with ``transform`` applied to every point. + + Waypoints, route points and track points are all transformed. The + structure of the GPX (routes, tracks, segments) is left unchanged. + """ + new_wpt = [transform(w) for w in gpx.wpt] + new_rte = [ + replace(route, rtept=[transform(p) for p in route.rtept]) for route in gpx.rte + ] + new_trk = [ + replace( + track, + trkseg=[ + replace(segment, trkpt=[transform(p) for p in segment.trkpt]) + for segment in track.trkseg + ], + ) + for track in gpx.trk + ] + return replace(gpx, wpt=new_wpt, rte=new_rte, trk=new_trk) + + +def crop( + gpx: GPX, + *, + min_lat: float | None = None, + max_lat: float | None = None, + min_lon: float | None = None, + max_lon: float | None = None, +) -> GPX: + """Crop a GPX to a geographic bounding box. + + Points outside the given bounds are dropped. Bounds that are None are + not enforced. Routes, track segments and tracks that end up empty are + dropped as well. + + Args: + gpx: The GPX instance to crop. + min_lat: Minimum latitude. Defaults to None (no minimum). + max_lat: Maximum latitude. Defaults to None (no maximum). + min_lon: Minimum longitude. Defaults to None (no minimum). + max_lon: Maximum longitude. Defaults to None (no maximum). + + Returns: + A new GPX instance cropped to the given bounds. + + Example: + >>> from gpx import crop, read_gpx + >>> gpx = read_gpx("path/to/file.gpx") + >>> cropped = crop(gpx, min_lat=52.0, max_lat=53.0) + + """ + + def is_in_bounds(point: Waypoint) -> bool: + lat, lon = float(point.lat), float(point.lon) + if min_lat is not None and lat < min_lat: + return False + if max_lat is not None and lat > max_lat: + return False + if min_lon is not None and lon < min_lon: + return False + return not (max_lon is not None and lon > max_lon) + + return filter_points(gpx, is_in_bounds) + + +def trim( + gpx: GPX, + *, + start: dt.datetime | None = None, + end: dt.datetime | None = None, +) -> GPX: + """Trim a GPX to a date/time range. + + Points with a timestamp outside the given range are dropped; points + without a timestamp are kept. Routes, track segments and tracks that + end up empty are dropped as well. + + Args: + gpx: The GPX instance to trim. + start: Start of the time range. Defaults to None (no start). + end: End of the time range. Defaults to None (no end). + + Returns: + A new GPX instance trimmed to the given time range. + + Example: + >>> import datetime as dt + >>> from gpx import read_gpx, trim + >>> gpx = read_gpx("path/to/file.gpx") + >>> trimmed = trim(gpx, start=dt.datetime(2024, 1, 1, tzinfo=dt.UTC)) + + """ + + def is_in_time_range(point: Waypoint) -> bool: + if point.time is None: + return True # Keep points without timestamps + if start is not None and point.time < start: + return False + return not (end is not None and point.time > end) + + return filter_points(gpx, is_in_time_range) + + +def reverse(gpx: GPX, *, routes: bool = True, tracks: bool = True) -> GPX: + """Reverse the routes and/or tracks of a GPX. + + For tracks, both the order of the segments and the order of the points + within each segment are reversed. + + Args: + gpx: The GPX instance to reverse. + routes: Whether to reverse the routes. Defaults to True. + tracks: Whether to reverse the tracks. Defaults to True. + + Returns: + A new GPX instance with the routes and/or tracks reversed. + + Example: + >>> from gpx import read_gpx, reverse + >>> gpx = read_gpx("path/to/file.gpx") + >>> reversed_gpx = reverse(gpx) + + """ + new_rte = gpx.rte + new_trk = gpx.trk + + if routes: + new_rte = [ + replace(route, rtept=list(reversed(route.rtept))) for route in gpx.rte + ] + + if tracks: + new_trk = [ + replace( + track, + trkseg=[ + replace(segment, trkpt=list(reversed(segment.trkpt))) + for segment in reversed(track.trkseg) + ], + ) + for track in gpx.trk + ] + + return replace(gpx, rte=new_rte, trk=new_trk) + + +def strip_metadata( # noqa: PLR0913 + gpx: GPX, + *, + name: bool = False, + desc: bool = False, + author: bool = False, + copyright: bool = False, # noqa: A002 + time: bool = False, + keywords: bool = False, + links: bool = False, +) -> GPX: + """Strip metadata (fields) from a GPX. + + When no field flags are given, the entire metadata element is removed. + Otherwise, only the selected fields are removed from the metadata. + + Args: + gpx: The GPX instance to strip metadata from. + name: Whether to strip the metadata name. Defaults to False. + desc: Whether to strip the metadata description. Defaults to False. + author: Whether to strip the metadata author. Defaults to False. + copyright: Whether to strip the metadata copyright. Defaults to False. + time: Whether to strip the metadata time. Defaults to False. + keywords: Whether to strip the metadata keywords. Defaults to False. + links: Whether to strip the metadata links. Defaults to False. + + Returns: + A new GPX instance with the metadata (fields) stripped. + + Example: + >>> from gpx import read_gpx, strip_metadata + >>> gpx = read_gpx("path/to/file.gpx") + >>> anonymous = strip_metadata(gpx, author=True, copyright=True) + >>> bare = strip_metadata(gpx) # removes all metadata + + """ + if not any((name, desc, author, copyright, time, keywords, links)): + return replace(gpx, metadata=None) + + metadata = gpx.metadata + if metadata is None: + return gpx + + if name: + metadata = replace(metadata, name=None) + if desc: + metadata = replace(metadata, desc=None) + if author: + metadata = replace(metadata, author=None) + if copyright: + metadata = replace(metadata, copyright=None) + if time: + metadata = replace(metadata, time=None) + if keywords: + metadata = replace(metadata, keywords=None) + if links: + metadata = replace(metadata, link=[]) + return replace(gpx, metadata=metadata) + + +def reduce_precision( + gpx: GPX, + *, + coordinate_precision: int | None = None, + elevation_precision: int | None = None, +) -> GPX: + """Reduce the precision of the coordinates and/or elevations of a GPX. + + Args: + gpx: The GPX instance to reduce the precision of. + coordinate_precision: Number of decimal places for lat/lon + coordinates. Defaults to None (unchanged). + elevation_precision: Number of decimal places for elevations. + Defaults to None (unchanged). + + Returns: + A new GPX instance with the precision reduced. + + Example: + >>> from gpx import read_gpx, reduce_precision + >>> gpx = read_gpx("path/to/file.gpx") + >>> reduced = reduce_precision(gpx, coordinate_precision=6) + + """ + + def round_point(point: Waypoint) -> Waypoint: + if coordinate_precision is None and elevation_precision is None: + return point + new_lat = ( + Latitude(str(round(float(point.lat), coordinate_precision))) + if coordinate_precision is not None + else point.lat + ) + new_lon = ( + Longitude(str(round(float(point.lon), coordinate_precision))) + if coordinate_precision is not None + else point.lon + ) + new_ele = ( + Decimal(str(round(float(point.ele), elevation_precision))) + if elevation_precision is not None and point.ele is not None + else point.ele + ) + return replace(point, lat=new_lat, lon=new_lon, ele=new_ele) + + return _map_points(gpx, round_point) + + +def split( + gpx: GPX, + *, + time_gap: dt.timedelta | None = None, + distance_gap: float | None = None, +) -> GPX: + """Split track segments at time and/or distance gaps. + + Each track segment is split into multiple segments wherever the gap + between two consecutive track points exceeds ``time_gap`` and/or + ``distance_gap``. This is useful when a GPS device kept recording + across a pause or signal loss, since the GPX 1.1 specification calls + for a new track segment for each continuous span of track data. + + Points without a timestamp never trigger a time-based split. Each + resulting segment keeps the original segment's extensions. Waypoints + and routes are left unchanged. + + Args: + gpx: The GPX instance to split. + time_gap: Split where the time between consecutive points exceeds + this duration. Defaults to None (no time-based splitting). + distance_gap: Split where the distance between consecutive points + exceeds this many metres. Defaults to None (no distance-based + splitting). + + Returns: + A new GPX instance with the track segments split. + + Raises: + ValueError: If neither ``time_gap`` nor ``distance_gap`` is given. + + Example: + >>> import datetime as dt + >>> from gpx import read_gpx, split + >>> gpx = read_gpx("path/to/file.gpx") + >>> split_gpx = split(gpx, time_gap=dt.timedelta(minutes=10)) + + """ + if time_gap is None and distance_gap is None: + msg = "At least one of time_gap or distance_gap must be given" + raise ValueError(msg) + + def is_gap(prev: Waypoint, point: Waypoint) -> bool: + if ( + time_gap is not None + and prev.time is not None + and point.time is not None + and point.time - prev.time > time_gap + ): + return True + return distance_gap is not None and prev.distance_to(point) > distance_gap + + new_trk: list[Track] = [] + for track in gpx.trk: + new_trkseg: list[TrackSegment] = [] + for segment in track.trkseg: + if not segment.trkpt: + new_trkseg.append(segment) + continue + parts: list[list[Waypoint]] = [[]] + for point in segment.trkpt: + if parts[-1] and is_gap(parts[-1][-1], point): + parts.append([]) + parts[-1].append(point) + new_trkseg.extend(replace(segment, trkpt=part) for part in parts) + new_trk.append(replace(track, trkseg=new_trkseg)) + + return replace(gpx, trk=new_trk) + + +#: Earth radius (in metres) used for planar projections, matching +#: :meth:`~gpx.waypoint.Waypoint.distance_to`. +_EARTH_RADIUS = 6_378_137 + + +def _perpendicular_distance(point: Waypoint, start: Waypoint, end: Waypoint) -> float: + """Return the distance (in metres) from ``point`` to the ``start``-``end`` segment. + + Uses a local equirectangular projection centered on ``start``, which is + accurate for the small extents typical of GPS tracks. + """ + lat0 = radians(float(start.lat)) + lon0 = radians(float(start.lon)) + cos_lat0 = cos(lat0) + + def project(p: Waypoint) -> tuple[float, float]: + x = (radians(float(p.lon)) - lon0) * cos_lat0 * _EARTH_RADIUS + y = (radians(float(p.lat)) - lat0) * _EARTH_RADIUS + return x, y + + px, py = project(point) + ex, ey = project(end) # start projects to (0, 0) + + segment_length_sq = ex * ex + ey * ey + if segment_length_sq == 0: + return sqrt(px * px + py * py) + + # Clamp the projection of the point onto the segment to its endpoints + t = max(0.0, min(1.0, (px * ex + py * ey) / segment_length_sq)) + dx = px - t * ex + dy = py - t * ey + return sqrt(dx * dx + dy * dy) + + +def _ramer_douglas_peucker(points: list[Waypoint], tolerance: float) -> list[Waypoint]: + """Simplify a list of points with the Ramer-Douglas-Peucker algorithm. + + Uses an explicit stack instead of recursion to handle long tracks. + """ + if len(points) < 3: # noqa: PLR2004 + return list(points) + + keep = [False] * len(points) + keep[0] = keep[-1] = True + stack = [(0, len(points) - 1)] + while stack: + start, end = stack.pop() + max_distance = 0.0 + index = start + for i in range(start + 1, end): + distance = _perpendicular_distance(points[i], points[start], points[end]) + if distance > max_distance: + max_distance = distance + index = i + if max_distance > tolerance: + keep[index] = True + stack.append((start, index)) + stack.append((index, end)) + + return [point for point, kept in zip(points, keep, strict=True) if kept] + + +def simplify(gpx: GPX, tolerance: float) -> GPX: + """Simplify the tracks and routes of a GPX. + + Reduces the number of route points and track points using the + Ramer-Douglas-Peucker algorithm: points closer than ``tolerance`` to the + line between their surviving neighbors are dropped, while the overall + shape is preserved. The first and last point of each route and track + segment are always kept. Waypoints are left unchanged. + + Args: + gpx: The GPX instance to simplify. + tolerance: The maximum allowed deviation (in metres) of a dropped + point from the simplified line. + + Returns: + A new GPX instance with the tracks and routes simplified. + + Raises: + ValueError: If ``tolerance`` is not positive. + + Example: + >>> from gpx import read_gpx, simplify + >>> gpx = read_gpx("path/to/file.gpx") + >>> simplified = simplify(gpx, tolerance=10.0) + + """ + if tolerance <= 0: + msg = f"tolerance must be positive, got {tolerance}" + raise ValueError(msg) + + new_rte = [ + replace(route, rtept=_ramer_douglas_peucker(route.rtept, tolerance)) + for route in gpx.rte + ] + new_trk = [ + replace( + track, + trkseg=[ + replace(segment, trkpt=_ramer_douglas_peucker(segment.trkpt, tolerance)) + for segment in track.trkseg + ], + ) + for track in gpx.trk + ] + + return replace(gpx, rte=new_rte, trk=new_trk) + + +def smooth( + gpx: GPX, + *, + window: int = 5, + coordinates: bool = True, + elevations: bool = True, +) -> GPX: + """Smooth the tracks and routes of a GPX with a moving average. + + Applies a centered moving average to the coordinates and/or elevations + of route points and track points, reducing GPS noise. The window + shrinks symmetrically near the start and end of each route and track + segment, so the first and last point are always left unchanged. Points + without elevation are skipped during elevation smoothing (and do not + contribute to their neighbors' averages). Waypoints are left unchanged. + + Args: + gpx: The GPX instance to smooth. + window: The size of the moving average window (an odd integer of at + least 3). Defaults to 5. + coordinates: Whether to smooth the lat/lon coordinates. Defaults to + True. + elevations: Whether to smooth the elevations. Defaults to True. + + Returns: + A new GPX instance with the tracks and routes smoothed. + + Raises: + ValueError: If ``window`` is not an odd integer of at least 3. + + Example: + >>> from gpx import read_gpx, smooth + >>> gpx = read_gpx("path/to/file.gpx") + >>> smoothed = smooth(gpx, window=5) + + """ + if window < 3 or window % 2 == 0: # noqa: PLR2004 + msg = f"window must be an odd integer of at least 3, got {window}" + raise ValueError(msg) + half = window // 2 + + def smooth_points(points: list[Waypoint]) -> list[Waypoint]: + n = len(points) + if n < 3: # noqa: PLR2004 + return list(points) + new_points: list[Waypoint] = [] + for i, point in enumerate(points): + # Shrink the window symmetrically near the ends + k = min(half, i, n - 1 - i) + if k == 0: + new_points.append(point) + continue + neighbors = points[i - k : i + k + 1] + new_lat, new_lon = point.lat, point.lon + if coordinates: + new_lat = Latitude(str(fmean(float(p.lat) for p in neighbors))) + new_lon = Longitude(str(fmean(float(p.lon) for p in neighbors))) + new_ele = point.ele + if elevations and point.ele is not None: + ele_values = [float(p.ele) for p in neighbors if p.ele is not None] + new_ele = Decimal(str(fmean(ele_values))) + new_points.append(replace(point, lat=new_lat, lon=new_lon, ele=new_ele)) + return new_points + + new_rte = [replace(route, rtept=smooth_points(route.rtept)) for route in gpx.rte] + new_trk = [ + replace( + track, + trkseg=[ + replace(segment, trkpt=smooth_points(segment.trkpt)) + for segment in track.trkseg + ], + ) + for track in gpx.trk + ] + + return replace(gpx, rte=new_rte, trk=new_trk) + + +def shift_time(gpx: GPX, delta: dt.timedelta) -> GPX: + """Shift all point timestamps of a GPX by a time delta. + + The timestamps of all waypoints, route points and track points are + shifted by ``delta``; points without a timestamp are left unchanged. + This is useful for fixing timezone mistakes or GPS clock drift. The + metadata timestamp (the file creation time) is not shifted. + + Args: + gpx: The GPX instance to shift the timestamps of. + delta: The time delta to shift by (may be negative). + + Returns: + A new GPX instance with the timestamps shifted. + + Example: + >>> import datetime as dt + >>> from gpx import read_gpx, shift_time + >>> gpx = read_gpx("path/to/file.gpx") + >>> shifted = shift_time(gpx, dt.timedelta(hours=-2)) + + """ + + def shift_point(point: Waypoint) -> Waypoint: + if point.time is None: + return point + return replace(point, time=point.time + delta) + + return _map_points(gpx, shift_point) + + +def strip_extensions(gpx: GPX) -> GPX: + """Strip all extensions from a GPX. + + Removes the extension elements from the GPX itself, its metadata, and + all waypoints, routes, route points, tracks, track segments and track + points. This is useful for removing vendor-specific data (e.g. heart + rate, cadence, temperature) for privacy or to reduce file size. + + Args: + gpx: The GPX instance to strip the extensions from. + + Returns: + A new GPX instance without extensions. + + Example: + >>> from gpx import read_gpx, strip_extensions + >>> gpx = read_gpx("path/to/file.gpx") + >>> stripped = strip_extensions(gpx) + + """ + + def strip_point(point: Waypoint) -> Waypoint: + if point.extensions is None: + return point + return replace(point, extensions=None) + + stripped = _map_points(gpx, strip_point) + + new_metadata = ( + replace(stripped.metadata, extensions=None) + if stripped.metadata is not None + else None + ) + new_rte = [replace(route, extensions=None) for route in stripped.rte] + new_trk = [ + replace( + track, + extensions=None, + trkseg=[replace(segment, extensions=None) for segment in track.trkseg], + ) + for track in stripped.trk + ] + + return replace( + stripped, metadata=new_metadata, rte=new_rte, trk=new_trk, extensions=None + ) + + +def merge(gpxs: Iterable[GPX], *, creator: str | None = None) -> GPX: + """Merge multiple GPX instances into one. + + The waypoints, routes and tracks of all instances are concatenated, in + order, into a single new GPX instance. + + Args: + gpxs: The GPX instances to merge. + creator: The creator string for the merged GPX. Defaults to None + (uses default). + + Returns: + A new GPX instance with the merged contents. + + Example: + >>> from gpx import merge, read_gpx + >>> merged = merge([read_gpx("one.gpx"), read_gpx("two.gpx")]) + + """ + merged_wpt: list[Waypoint] = [] + merged_rte: list[Route] = [] + merged_trk: list[Track] = [] + + for gpx in gpxs: + merged_wpt.extend(gpx.wpt) + merged_rte.extend(gpx.rte) + merged_trk.extend(gpx.trk) + + merged = GPX(wpt=merged_wpt, rte=merged_rte, trk=merged_trk) + if creator is not None: + merged = replace(merged, creator=creator) + return merged diff --git a/src/gpx/utils.py b/src/gpx/utils.py index 6cc4846..60dad7a 100644 --- a/src/gpx/utils.py +++ b/src/gpx/utils.py @@ -70,7 +70,7 @@ def extract_namespaces_from_string(xml_string: str) -> dict[str, str]: uri = match.group(2) # Use empty string for default namespace (xmlns="...") - key = prefix if prefix else "" + key = prefix or "" namespaces[key] = uri return namespaces diff --git a/tests/conftest.py b/tests/conftest.py index f043743..4eb7ffe 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -9,8 +9,10 @@ import pytest from gpx import ( + GPX, Bounds, Email, + Extensions, Link, Metadata, Person, @@ -72,11 +74,6 @@ def load_fixture(fixture_path: Path) -> str: return fixture_path.read_text(encoding="utf-8") -# ============================================================================= -# GPX string fixtures (loaded from files) -# ============================================================================= - - @pytest.fixture def minimal_gpx_string() -> str: """A minimal valid GPX string.""" @@ -119,11 +116,6 @@ def invalid_gpx_string() -> str: return load_fixture(INVALID_FIXTURES_DIR / "missing_lat_lon.gpx") -# ============================================================================= -# GeoJSON interface fixtures -# ============================================================================= - - @pytest.fixture def waypoint_geo_interface() -> dict[str, Any]: """A GeoJSON-like dict for a waypoint.""" @@ -248,11 +240,6 @@ def full_gpx_geo_interface() -> dict[str, Any]: } -# ============================================================================= -# Programmatically created fixtures -# ============================================================================= - - @pytest.fixture def sample_waypoint() -> Waypoint: """Create a sample waypoint programmatically.""" @@ -353,9 +340,199 @@ def sample_bounds() -> Bounds: ) -# ============================================================================= -# Edge case fixtures (valid) -# ============================================================================= +#: Garmin TrackPointExtension namespace (for extension tests) +GARMIN_TPX_NS = "http://www.garmin.com/xmlschemas/TrackPointExtension/v2" +#: Custom extension namespace (for extension tests) +CUSTOM_NS = "http://example.com/custom/v1" + + +def create_track_point_extension(hr: str = "142", cad: str = "85") -> Extensions: + """Create a Garmin TrackPointExtension. + + Args: + hr: Heart rate value. Defaults to "142". + cad: Cadence value. Defaults to "85". + + Returns: + An Extensions instance containing the TrackPointExtension. + + """ + ET.register_namespace("gpxtpx", GARMIN_TPX_NS) + tpx = ET.Element(f"{{{GARMIN_TPX_NS}}}TrackPointExtension") + hr_elem = ET.SubElement(tpx, f"{{{GARMIN_TPX_NS}}}hr") + hr_elem.text = hr + cad_elem = ET.SubElement(tpx, f"{{{GARMIN_TPX_NS}}}cad") + cad_elem.text = cad + return Extensions(elements=[tpx]) + + +def create_custom_extension(key: str, value: str) -> Extensions: + """Create a custom extension element. + + Args: + key: The element tag name. + value: The element text content. + + Returns: + An Extensions instance containing the custom element. + + """ + elem = ET.Element(f"{{{CUSTOM_NS}}}{key}") + elem.text = value + return Extensions(elements=[elem]) + + +@pytest.fixture +def sample_gpx() -> GPX: + """Create a sample GPX with waypoints, routes, and tracks.""" + waypoint = Waypoint( + lat=Latitude("52.5200"), + lon=Longitude("13.4050"), + ele=Decimal("34.5"), + name="Berlin", + time=dt.datetime(2024, 1, 15, 10, 0, 0, tzinfo=dt.UTC), + ) + + route_points = [ + Waypoint( + lat=Latitude("52.5200"), + lon=Longitude("13.4050"), + ele=Decimal("34.5"), + ), + Waypoint( + lat=Latitude("52.5300"), + lon=Longitude("13.4150"), + ele=Decimal("40.0"), + ), + ] + route = Route(name="City Tour", desc="A tour of the city", rtept=route_points) + + track_points = [ + Waypoint( + lat=Latitude("52.5200"), + lon=Longitude("13.4050"), + ele=Decimal("34.5"), + time=dt.datetime(2024, 1, 15, 10, 0, 0, tzinfo=dt.UTC), + ), + Waypoint( + lat=Latitude("52.5210"), + lon=Longitude("13.4060"), + ele=Decimal("35.0"), + time=dt.datetime(2024, 1, 15, 10, 1, 0, tzinfo=dt.UTC), + ), + Waypoint( + lat=Latitude("52.5220"), + lon=Longitude("13.4070"), + ele=Decimal("36.5"), + time=dt.datetime(2024, 1, 15, 10, 2, 0, tzinfo=dt.UTC), + ), + ] + segment = TrackSegment(trkpt=track_points) + track = Track(name="Morning Run", desc="A morning run", trkseg=[segment]) + + return GPX( + creator="TestApp", + metadata=Metadata(name="Test GPX", desc="Test description"), + wpt=[waypoint], + rte=[route], + trk=[track], + ) + + +@pytest.fixture +def sample_gpx_file(sample_gpx: GPX, tmp_path_factory: pytest.TempPathFactory) -> Path: + """Create a sample GPX file and return its path.""" + temp_dir = tmp_path_factory.mktemp("gpx_files") + temp_file = temp_dir / "sample.gpx" + sample_gpx.write_gpx(temp_file) + return temp_file + + +@pytest.fixture +def gpx_with_extensions() -> GPX: + """Create a GPX with extensions at all levels.""" + # GPX-level extension + gpx_ext = create_custom_extension("source", "TestApp") + + # Metadata extension + metadata_ext = create_custom_extension("version", "1.0") + + # Track extension + track_ext = create_custom_extension("activity", "running") + + # Track segment extension + segment_ext = create_custom_extension("lap", "1") + + # Track point extensions with heart rate + track_points = [ + Waypoint( + lat=Latitude("52.5200"), + lon=Longitude("13.4050"), + ele=Decimal("34.5"), + time=dt.datetime(2024, 1, 15, 10, 0, 0, tzinfo=dt.UTC), + extensions=create_track_point_extension("140", "80"), + ), + Waypoint( + lat=Latitude("52.5210"), + lon=Longitude("13.4060"), + ele=Decimal("35.0"), + time=dt.datetime(2024, 1, 15, 10, 1, 0, tzinfo=dt.UTC), + extensions=create_track_point_extension("145", "85"), + ), + Waypoint( + lat=Latitude("52.5220"), + lon=Longitude("13.4070"), + ele=Decimal("36.5"), + time=dt.datetime(2024, 1, 15, 10, 2, 0, tzinfo=dt.UTC), + extensions=create_track_point_extension("150", "90"), + ), + ] + segment = TrackSegment(trkpt=track_points, extensions=segment_ext) + track = Track(name="Morning Run", trkseg=[segment], extensions=track_ext) + + # Route with extensions + route_ext = create_custom_extension("type", "scenic") + route_points = [ + Waypoint( + lat=Latitude("52.5200"), + lon=Longitude("13.4050"), + extensions=create_custom_extension("poi", "start"), + ), + Waypoint( + lat=Latitude("52.5300"), + lon=Longitude("13.4150"), + extensions=create_custom_extension("poi", "end"), + ), + ] + route = Route(name="City Tour", rtept=route_points, extensions=route_ext) + + # Waypoint with extension + waypoint = Waypoint( + lat=Latitude("52.5200"), + lon=Longitude("13.4050"), + name="Test Point", + extensions=create_custom_extension("rating", "5"), + ) + + return GPX( + creator="TestApp", + metadata=Metadata(name="Test GPX", extensions=metadata_ext), + wpt=[waypoint], + rte=[route], + trk=[track], + extensions=gpx_ext, + ) + + +@pytest.fixture +def gpx_with_extensions_file( + gpx_with_extensions: GPX, tmp_path_factory: pytest.TempPathFactory +) -> Path: + """Create a GPX file with extensions and return its path.""" + temp_dir = tmp_path_factory.mktemp("gpx_ext_files") + temp_file = temp_dir / "with_extensions.gpx" + gpx_with_extensions.write_gpx(temp_file) + return temp_file @pytest.fixture @@ -508,11 +685,6 @@ def large_gpx_string() -> str: return load_fixture(VALID_FIXTURES_DIR / "large_gpx.gpx") -# ============================================================================= -# Edge case fixtures (invalid) -# ============================================================================= - - @pytest.fixture def missing_lat_gpx_string() -> str: """A GPX string with a waypoint missing latitude.""" diff --git a/tests/test_cli.py b/tests/test_cli.py index 0c5b4bf..74b919f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2,113 +2,20 @@ import datetime as dt import json -import xml.etree.ElementTree as ET -from decimal import Decimal from pathlib import Path import pytest -from gpx import ( - GPX, - Extensions, - Metadata, - Route, - Track, - TrackSegment, - Waypoint, - read_gpx, -) -from gpx.cli import ( - _apply_crop, - _apply_precision, - _apply_reverse, - _apply_trim, - _calculate_bounds, - _detect_format, - _gather_gpx_info, - _parse_datetime, - cli, -) +from gpx import GPX, Track, TrackSegment, Waypoint, read_gpx +from gpx.cli import _calculate_bounds, _gather_gpx_info, _parse_datetime, cli from gpx.types import Latitude, Longitude -# Extension namespace for testing -GARMIN_TPX_NS = "http://www.garmin.com/xmlschemas/TrackPointExtension/v2" -CUSTOM_NS = "http://example.com/custom/v1" - -# ============================================================================= -# Test fixtures -# ============================================================================= - - -@pytest.fixture -def sample_gpx() -> GPX: - """Create a sample GPX with waypoints, routes, and tracks.""" - waypoint = Waypoint( - lat=Latitude("52.5200"), - lon=Longitude("13.4050"), - ele=Decimal("34.5"), - name="Berlin", - time=dt.datetime(2024, 1, 15, 10, 0, 0, tzinfo=dt.UTC), - ) - - route_points = [ - Waypoint( - lat=Latitude("52.5200"), - lon=Longitude("13.4050"), - ele=Decimal("34.5"), - ), - Waypoint( - lat=Latitude("52.5300"), - lon=Longitude("13.4150"), - ele=Decimal("40.0"), - ), - ] - route = Route(name="City Tour", desc="A tour of the city", rtept=route_points) - - track_points = [ - Waypoint( - lat=Latitude("52.5200"), - lon=Longitude("13.4050"), - ele=Decimal("34.5"), - time=dt.datetime(2024, 1, 15, 10, 0, 0, tzinfo=dt.UTC), - ), - Waypoint( - lat=Latitude("52.5210"), - lon=Longitude("13.4060"), - ele=Decimal("35.0"), - time=dt.datetime(2024, 1, 15, 10, 1, 0, tzinfo=dt.UTC), - ), - Waypoint( - lat=Latitude("52.5220"), - lon=Longitude("13.4070"), - ele=Decimal("36.5"), - time=dt.datetime(2024, 1, 15, 10, 2, 0, tzinfo=dt.UTC), - ), - ] - segment = TrackSegment(trkpt=track_points) - track = Track(name="Morning Run", desc="A morning run", trkseg=[segment]) - - return GPX( - creator="TestApp", - metadata=Metadata(name="Test GPX", desc="Test description"), - wpt=[waypoint], - rte=[route], - trk=[track], - ) - - -@pytest.fixture -def sample_gpx_file(sample_gpx: GPX, tmp_path_factory: pytest.TempPathFactory) -> Path: - """Create a sample GPX file and return its path.""" - temp_dir = tmp_path_factory.mktemp("gpx_files") - temp_file = temp_dir / "sample.gpx" - sample_gpx.write_gpx(temp_file) - return temp_file - - -# ============================================================================= -# Main function tests -# ============================================================================= +from .conftest import ( + CUSTOM_NS, + GARMIN_TPX_NS, + create_custom_extension, + create_track_point_extension, +) class TestMain: @@ -135,11 +42,6 @@ def test_main_invalid_command(self) -> None: assert exc_info.value.code == 2 -# ============================================================================= -# Validate command tests -# ============================================================================= - - class TestValidateCommand: """Tests for the validate command.""" @@ -173,11 +75,6 @@ def test_validate_invalid_file( assert "Invalid GPX file" in captured.err -# ============================================================================= -# Info command tests -# ============================================================================= - - class TestInfoCommand: """Tests for the info command.""" @@ -213,11 +110,6 @@ def test_info_nonexistent_file(self, capsys: pytest.CaptureFixture[str]) -> None assert "File not found" in captured.err -# ============================================================================= -# Edit command tests -# ============================================================================= - - class TestEditCommand: """Tests for the edit command.""" @@ -272,10 +164,204 @@ def test_edit_strip_all_metadata( gpx = read_gpx(output_file) assert gpx.metadata is None + def test_edit_split_time_gap(self, sample_gpx_file: Path, tmp_path: Path) -> None: + """Test edit with --split-time-gap.""" + output_file = tmp_path / "output.gpx" + # Sample track points are 1 minute apart; split at 30 seconds + result = cli( + [ + "edit", + str(sample_gpx_file), + "-o", + str(output_file), + "--split-time-gap", + "30", + ] + ) + assert result == 0 + gpx = read_gpx(output_file) + assert len(gpx.trk[0].trkseg) == 3 + + def test_edit_split_distance_gap( + self, sample_gpx_file: Path, tmp_path: Path + ) -> None: + """Test edit with --split-distance-gap.""" + output_file = tmp_path / "output.gpx" + # Sample track points are ~130 m apart; split at 50 m + result = cli( + [ + "edit", + str(sample_gpx_file), + "-o", + str(output_file), + "--split-distance-gap", + "50", + ] + ) + assert result == 0 + gpx = read_gpx(output_file) + assert len(gpx.trk[0].trkseg) == 3 + + def test_edit_simplify(self, sample_gpx_file: Path, tmp_path: Path) -> None: + """Test edit with --simplify.""" + output_file = tmp_path / "output.gpx" + result = cli( + [ + "edit", + str(sample_gpx_file), + "-o", + str(output_file), + "--simplify", + "1000", + ] + ) + assert result == 0 + gpx = read_gpx(output_file) + # With a huge tolerance, only the endpoints of the track remain + assert len(gpx.trk[0].trkseg[0].trkpt) == 2 + + def test_edit_smooth(self, sample_gpx_file: Path, tmp_path: Path) -> None: + """Test edit with --smooth.""" + output_file = tmp_path / "output.gpx" + result = cli( + ["edit", str(sample_gpx_file), "-o", str(output_file), "--smooth", "3"] + ) + assert result == 0 + gpx = read_gpx(output_file) + assert len(gpx.trk[0].trkseg[0].trkpt) == 3 + + def test_edit_smooth_invalid_window( + self, sample_gpx_file: Path, tmp_path: Path, capsys: pytest.CaptureFixture[str] + ) -> None: + """Test edit with an invalid --smooth window.""" + output_file = tmp_path / "output.gpx" + result = cli( + ["edit", str(sample_gpx_file), "-o", str(output_file), "--smooth", "2"] + ) + assert result == 1 + captured = capsys.readouterr() + assert "window" in captured.err + + def test_edit_shift_time(self, sample_gpx_file: Path, tmp_path: Path) -> None: + """Test edit with --shift-time.""" + output_file = tmp_path / "output.gpx" + result = cli( + [ + "edit", + str(sample_gpx_file), + "-o", + str(output_file), + "--shift-time", + "3600", + ] + ) + assert result == 0 + original = read_gpx(sample_gpx_file) + shifted = read_gpx(output_file) + original_time = original.wpt[0].time + assert original_time is not None + assert shifted.wpt[0].time == original_time + dt.timedelta(hours=1) + + def test_edit_strip_extensions(self, tmp_path: Path) -> None: + """Test edit with --strip-extensions.""" + input_file = tmp_path / "input.gpx" + gpx = GPX( + wpt=[ + Waypoint( + lat=Latitude("52.0"), + lon=Longitude("4.0"), + extensions=create_custom_extension("rating", "5"), + ) + ], + extensions=create_custom_extension("source", "TestApp"), + ) + gpx.write_gpx(input_file) + + output_file = tmp_path / "output.gpx" + result = cli( + ["edit", str(input_file), "-o", str(output_file), "--strip-extensions"] + ) + assert result == 0 + stripped = read_gpx(output_file) + assert stripped.extensions is None + assert stripped.wpt[0].extensions is None + + def test_edit_crop_with_zero_bound( + self, sample_gpx_file: Path, tmp_path: Path + ) -> None: + """A bound of 0 must trigger the crop (regression: any() treated 0 as falsy). + + sample_gpx points are all near (52.5, 13.4); a min-lat of 0 must keep + them while a min-lat of 53 must drop them. + """ + out_keep = tmp_path / "keep.gpx" + result = cli( + [ + "edit", + str(sample_gpx_file), + "-o", + str(out_keep), + "--min-lat", + "0", + ] + ) + assert result == 0 + kept = read_gpx(out_keep) + assert len(kept.wpt) == 1 + + out_drop = tmp_path / "drop.gpx" + result = cli( + [ + "edit", + str(sample_gpx_file), + "-o", + str(out_drop), + "--min-lat", + "53", + ] + ) + assert result == 0 + dropped = read_gpx(out_drop) + assert len(dropped.wpt) == 0 + + def test_edit_strip_all_metadata_preserves_nsmap(self, tmp_path: Path) -> None: + """`gpx edit --strip-all-metadata` must keep custom namespace prefixes. -# ============================================================================= -# Merge command tests -# ============================================================================= + The prefix must survive an actual file round-trip, which means an + element using that namespace has to be present (ElementTree drops + unused xmlns declarations). + """ + src = tmp_path / "with_ext.gpx" + src.write_text( + '\n' + '\n' + " x\n" + ' \n' + " \n" + " \n" + " 120\n" + " \n" + " \n" + " \n" + "\n", + encoding="utf-8", + ) + + out = tmp_path / "stripped.gpx" + result = cli( + [ + "edit", + str(src), + "-o", + str(out), + "--strip-all-metadata", + ] + ) + assert result == 0 + # Original prefix must survive the round-trip through `gpx edit`. + assert f'xmlns:gpxtpx="{GARMIN_TPX_NS}"' in out.read_text(encoding="utf-8") class TestMergeCommand: @@ -311,11 +397,6 @@ def test_merge_nonexistent_file( assert "File not found" in captured.err -# ============================================================================= -# Convert command tests -# ============================================================================= - - class TestConvertCommand: """Tests for the convert command.""" @@ -381,11 +462,6 @@ def test_convert_nonexistent_file( assert "File not found" in captured.err -# ============================================================================= -# Helper function tests -# ============================================================================= - - class TestHelperFunctions: """Tests for CLI helper functions.""" @@ -410,23 +486,6 @@ def test_parse_datetime_invalid(self) -> None: with pytest.raises(ValueError, match="Invalid datetime format"): _parse_datetime("not a date") - def test_detect_format_gpx(self) -> None: - """Test format detection for GPX.""" - assert _detect_format(Path("file.gpx")) == "gpx" - - def test_detect_format_geojson(self) -> None: - """Test format detection for GeoJSON.""" - assert _detect_format(Path("file.geojson")) == "geojson" - assert _detect_format(Path("file.json")) == "geojson" - - def test_detect_format_kml(self) -> None: - """Test format detection for KML.""" - assert _detect_format(Path("file.kml")) == "kml" - - def test_detect_format_unknown(self) -> None: - """Test format detection for unknown extension.""" - assert _detect_format(Path("file.xyz")) is None - def test_calculate_bounds(self, sample_gpx: GPX) -> None: """Test bounds calculation.""" bounds = _calculate_bounds(sample_gpx) @@ -448,326 +507,6 @@ def test_gather_gpx_info(self, sample_gpx: GPX) -> None: assert info["metadata"]["name"] == "Test GPX" -# ============================================================================= -# Apply function tests -# ============================================================================= - - -class TestApplyFunctions: - """Tests for the apply transformation functions.""" - - def test_apply_crop(self, sample_gpx: GPX) -> None: - """Test applying crop to GPX data.""" - cropped = _apply_crop( - sample_gpx, - min_lat=52.515, - max_lat=52.525, - min_lon=13.400, - max_lon=13.410, - ) - # Should keep some points but filter out those outside bounds - assert len(cropped.wpt) >= 0 - assert len(cropped.trk) >= 0 - - def test_apply_crop_excludes_all(self, sample_gpx: GPX) -> None: - """Test applying crop that excludes all points.""" - cropped = _apply_crop( - sample_gpx, - min_lat=0.0, - max_lat=1.0, - min_lon=0.0, - max_lon=1.0, - ) - assert len(cropped.wpt) == 0 - assert len(cropped.rte) == 0 - assert len(cropped.trk) == 0 - - def test_edit_crop_with_zero_bound( - self, sample_gpx_file: Path, tmp_path: Path - ) -> None: - """A bound of 0 must trigger the crop (regression: any() treated 0 as falsy). - - sample_gpx points are all near (52.5, 13.4); a min-lat of 0 must keep - them while a min-lat of 53 must drop them. - """ - out_keep = tmp_path / "keep.gpx" - result = cli( - [ - "edit", - str(sample_gpx_file), - "-o", - str(out_keep), - "--min-lat", - "0", - ] - ) - assert result == 0 - kept = read_gpx(out_keep) - assert len(kept.wpt) == 1 - - out_drop = tmp_path / "drop.gpx" - result = cli( - [ - "edit", - str(sample_gpx_file), - "-o", - str(out_drop), - "--min-lat", - "53", - ] - ) - assert result == 0 - dropped = read_gpx(out_drop) - assert len(dropped.wpt) == 0 - - def test_apply_trim(self, sample_gpx: GPX) -> None: - """Test applying time trim to GPX data.""" - start = dt.datetime(2024, 1, 15, 10, 0, 30, tzinfo=dt.UTC) - end = dt.datetime(2024, 1, 15, 10, 1, 30, tzinfo=dt.UTC) - trimmed = _apply_trim(sample_gpx, start, end) - # Should keep only points within time range - assert len(trimmed.trk) > 0 - for track in trimmed.trk: - for seg in track.trkseg: - for pt in seg.trkpt: - if pt.time: - assert pt.time >= start - assert pt.time <= end - - def test_apply_reverse_routes(self, sample_gpx: GPX) -> None: - """Test applying reverse to routes.""" - original_first_lat = float(sample_gpx.rte[0].rtept[0].lat) - original_last_lat = float(sample_gpx.rte[0].rtept[-1].lat) - - reversed_gpx = _apply_reverse( - sample_gpx, reverse_routes=True, reverse_tracks=False - ) - - new_first_lat = float(reversed_gpx.rte[0].rtept[0].lat) - new_last_lat = float(reversed_gpx.rte[0].rtept[-1].lat) - - assert new_first_lat == pytest.approx(original_last_lat, rel=1e-3) - assert new_last_lat == pytest.approx(original_first_lat, rel=1e-3) - - def test_apply_reverse_tracks(self, sample_gpx: GPX) -> None: - """Test applying reverse to tracks.""" - original_first_lat = float(sample_gpx.trk[0].trkseg[0].trkpt[0].lat) - original_last_lat = float(sample_gpx.trk[0].trkseg[0].trkpt[-1].lat) - - reversed_gpx = _apply_reverse( - sample_gpx, reverse_routes=False, reverse_tracks=True - ) - - new_first_lat = float(reversed_gpx.trk[0].trkseg[0].trkpt[0].lat) - new_last_lat = float(reversed_gpx.trk[0].trkseg[0].trkpt[-1].lat) - - assert new_first_lat == pytest.approx(original_last_lat, rel=1e-3) - assert new_last_lat == pytest.approx(original_first_lat, rel=1e-3) - - def test_apply_precision(self, sample_gpx: GPX) -> None: - """Test applying precision reduction.""" - reduced = _apply_precision(sample_gpx, coord_precision=2, elevation_precision=0) - - # Check waypoint precision - for wpt in reduced.wpt: - lat_str = str(wpt.lat) - if "." in lat_str: - assert len(lat_str.split(".")[1]) <= 2 - if wpt.ele: - ele_str = str(wpt.ele) - if "." in ele_str: - assert len(ele_str.split(".")[1]) <= 1 - - def test_apply_precision_no_elevation(self) -> None: - """Test applying precision when no elevation.""" - gpx = GPX( - wpt=[Waypoint(lat=Latitude("52.5200123"), lon=Longitude("13.4050456"))] - ) - reduced = _apply_precision(gpx, coord_precision=3, elevation_precision=None) - lat_str = str(reduced.wpt[0].lat) - assert lat_str == "52.52" - - def test_apply_helpers_preserve_nsmap(self, sample_gpx: GPX) -> None: - """All edit transforms must propagate gpx.nsmap so prefixes survive a round-trip. - - Regression test: previously _apply_crop / _apply_trim / _apply_reverse / - _apply_precision / _apply_strip_metadata silently dropped nsmap, which - meant `gpx edit` on a file with custom prefixes (e.g. gpxtpx) emitted - auto-generated ns0/ns1 prefixes instead. - """ - sample_gpx.nsmap = { - "": "http://www.topografix.com/GPX/1/1", - "gpxtpx": GARMIN_TPX_NS, - } - - cropped = _apply_crop( - sample_gpx, min_lat=0.0, max_lat=90.0, min_lon=0.0, max_lon=90.0 - ) - assert cropped.nsmap == sample_gpx.nsmap - - trimmed = _apply_trim(sample_gpx, None, None) - assert trimmed.nsmap == sample_gpx.nsmap - - reversed_gpx = _apply_reverse( - sample_gpx, reverse_routes=True, reverse_tracks=True - ) - assert reversed_gpx.nsmap == sample_gpx.nsmap - - precised = _apply_precision( - sample_gpx, coord_precision=3, elevation_precision=1 - ) - assert precised.nsmap == sample_gpx.nsmap - - def test_edit_strip_all_metadata_preserves_nsmap(self, tmp_path: Path) -> None: - """`gpx edit --strip-all-metadata` must keep custom namespace prefixes. - - The prefix must survive an actual file round-trip, which means an - element using that namespace has to be present (ElementTree drops - unused xmlns declarations). - """ - src = tmp_path / "with_ext.gpx" - src.write_text( - '\n' - '\n' - " x\n" - ' \n' - " \n" - " \n" - " 120\n" - " \n" - " \n" - " \n" - "\n", - encoding="utf-8", - ) - - out = tmp_path / "stripped.gpx" - result = cli( - [ - "edit", - str(src), - "-o", - str(out), - "--strip-all-metadata", - ] - ) - assert result == 0 - # Original prefix must survive the round-trip through `gpx edit`. - assert f'xmlns:gpxtpx="{GARMIN_TPX_NS}"' in out.read_text(encoding="utf-8") - - -# ============================================================================= -# Extensions tests -# ============================================================================= - - -def _create_track_point_extension(hr: str = "142", cad: str = "85") -> Extensions: - """Create a Garmin TrackPointExtension.""" - ET.register_namespace("gpxtpx", GARMIN_TPX_NS) - tpx = ET.Element(f"{{{GARMIN_TPX_NS}}}TrackPointExtension") - hr_elem = ET.SubElement(tpx, f"{{{GARMIN_TPX_NS}}}hr") - hr_elem.text = hr - cad_elem = ET.SubElement(tpx, f"{{{GARMIN_TPX_NS}}}cad") - cad_elem.text = cad - return Extensions(elements=[tpx]) - - -def _create_custom_extension(key: str, value: str) -> Extensions: - """Create a custom extension element.""" - elem = ET.Element(f"{{{CUSTOM_NS}}}{key}") - elem.text = value - return Extensions(elements=[elem]) - - -@pytest.fixture -def gpx_with_extensions() -> GPX: - """Create a GPX with extensions at all levels.""" - # GPX-level extension - gpx_ext = _create_custom_extension("source", "TestApp") - - # Metadata extension - metadata_ext = _create_custom_extension("version", "1.0") - - # Track extension - track_ext = _create_custom_extension("activity", "running") - - # Track segment extension - segment_ext = _create_custom_extension("lap", "1") - - # Track point extensions with heart rate - track_points = [ - Waypoint( - lat=Latitude("52.5200"), - lon=Longitude("13.4050"), - ele=Decimal("34.5"), - time=dt.datetime(2024, 1, 15, 10, 0, 0, tzinfo=dt.UTC), - extensions=_create_track_point_extension("140", "80"), - ), - Waypoint( - lat=Latitude("52.5210"), - lon=Longitude("13.4060"), - ele=Decimal("35.0"), - time=dt.datetime(2024, 1, 15, 10, 1, 0, tzinfo=dt.UTC), - extensions=_create_track_point_extension("145", "85"), - ), - Waypoint( - lat=Latitude("52.5220"), - lon=Longitude("13.4070"), - ele=Decimal("36.5"), - time=dt.datetime(2024, 1, 15, 10, 2, 0, tzinfo=dt.UTC), - extensions=_create_track_point_extension("150", "90"), - ), - ] - segment = TrackSegment(trkpt=track_points, extensions=segment_ext) - track = Track(name="Morning Run", trkseg=[segment], extensions=track_ext) - - # Route with extensions - route_ext = _create_custom_extension("type", "scenic") - route_points = [ - Waypoint( - lat=Latitude("52.5200"), - lon=Longitude("13.4050"), - extensions=_create_custom_extension("poi", "start"), - ), - Waypoint( - lat=Latitude("52.5300"), - lon=Longitude("13.4150"), - extensions=_create_custom_extension("poi", "end"), - ), - ] - route = Route(name="City Tour", rtept=route_points, extensions=route_ext) - - # Waypoint with extension - waypoint = Waypoint( - lat=Latitude("52.5200"), - lon=Longitude("13.4050"), - name="Test Point", - extensions=_create_custom_extension("rating", "5"), - ) - - return GPX( - creator="TestApp", - metadata=Metadata(name="Test GPX", extensions=metadata_ext), - wpt=[waypoint], - rte=[route], - trk=[track], - extensions=gpx_ext, - ) - - -@pytest.fixture -def gpx_with_extensions_file( - gpx_with_extensions: GPX, tmp_path_factory: pytest.TempPathFactory -) -> Path: - """Create a GPX file with extensions and return its path.""" - temp_dir = tmp_path_factory.mktemp("gpx_ext_files") - temp_file = temp_dir / "with_extensions.gpx" - gpx_with_extensions.write_gpx(temp_file) - return temp_file - - class TestExtensionsInCLI: """Tests for extensions handling in CLI commands.""" @@ -996,14 +735,14 @@ def test_merge_preserves_extensions( trk=[ Track( name="Afternoon Run", - extensions=_create_custom_extension("activity", "cycling"), + extensions=create_custom_extension("activity", "cycling"), trkseg=[ TrackSegment( trkpt=[ Waypoint( lat=Latitude("52.6000"), lon=Longitude("13.5000"), - extensions=_create_track_point_extension( + extensions=create_track_point_extension( "160", "95" ), ), @@ -1057,140 +796,3 @@ def test_convert_gpx_to_gpx_preserves_extensions( # Track point extensions preserved assert gpx.trk[0].trkseg[0].trkpt[0].extensions is not None - - -class TestApplyFunctionsWithExtensions: - """Test apply functions preserve extensions.""" - - def test_apply_crop_preserves_waypoint_extensions( - self, gpx_with_extensions: GPX - ) -> None: - """Test that crop preserves extensions on remaining waypoints.""" - cropped = _apply_crop( - gpx_with_extensions, - min_lat=52.0, - max_lat=53.0, - min_lon=13.0, - max_lon=14.0, - ) - - # Waypoint should be preserved with extensions - assert len(cropped.wpt) == 1 - assert cropped.wpt[0].extensions is not None - assert cropped.wpt[0].extensions.get_text("rating", namespace=CUSTOM_NS) == "5" - - def test_apply_crop_preserves_track_extensions( - self, gpx_with_extensions: GPX - ) -> None: - """Test that crop preserves track and track point extensions.""" - cropped = _apply_crop( - gpx_with_extensions, - min_lat=52.0, - max_lat=53.0, - min_lon=13.0, - max_lon=14.0, - ) - - # Track extensions preserved - assert cropped.trk[0].extensions is not None - assert ( - cropped.trk[0].extensions.get_text("activity", namespace=CUSTOM_NS) - == "running" - ) - - # Track point extensions preserved - assert cropped.trk[0].trkseg[0].trkpt[0].extensions is not None - assert ( - cropped.trk[0] - .trkseg[0] - .trkpt[0] - .extensions.get_text("hr", namespace=GARMIN_TPX_NS) - == "140" - ) - - def test_apply_crop_preserves_route_extensions( - self, gpx_with_extensions: GPX - ) -> None: - """Test that crop preserves route and route point extensions.""" - cropped = _apply_crop( - gpx_with_extensions, - min_lat=52.0, - max_lat=53.0, - min_lon=13.0, - max_lon=14.0, - ) - - # Route extensions preserved - assert cropped.rte[0].extensions is not None - assert ( - cropped.rte[0].extensions.get_text("type", namespace=CUSTOM_NS) == "scenic" - ) - - # Route point extensions preserved - assert cropped.rte[0].rtept[0].extensions is not None - - def test_apply_trim_preserves_extensions(self, gpx_with_extensions: GPX) -> None: - """Test that trim preserves extensions on remaining elements.""" - start = dt.datetime(2024, 1, 15, 10, 0, 30, tzinfo=dt.UTC) - end = dt.datetime(2024, 1, 15, 10, 1, 30, tzinfo=dt.UTC) - trimmed = _apply_trim(gpx_with_extensions, start, end) - - # Track extensions preserved - if trimmed.trk: - assert trimmed.trk[0].extensions is not None - assert ( - trimmed.trk[0].extensions.get_text("activity", namespace=CUSTOM_NS) - == "running" - ) - - def test_apply_reverse_preserves_all_extensions( - self, gpx_with_extensions: GPX - ) -> None: - """Test that reverse preserves all extensions.""" - reversed_gpx = _apply_reverse( - gpx_with_extensions, reverse_routes=True, reverse_tracks=True - ) - - # Route extensions preserved - assert reversed_gpx.rte[0].extensions is not None - assert ( - reversed_gpx.rte[0].extensions.get_text("type", namespace=CUSTOM_NS) - == "scenic" - ) - - # Track extensions preserved - assert reversed_gpx.trk[0].extensions is not None - assert ( - reversed_gpx.trk[0].extensions.get_text("activity", namespace=CUSTOM_NS) - == "running" - ) - - # Track segment extensions preserved - assert reversed_gpx.trk[0].trkseg[0].extensions is not None - - # Track point extensions preserved (order reversed) - # First point after reverse was the last point (hr=150) - first_point_ext = reversed_gpx.trk[0].trkseg[0].trkpt[0].extensions - assert first_point_ext is not None - assert first_point_ext.get_text("hr", namespace=GARMIN_TPX_NS) == "150" - - def test_apply_precision_preserves_all_extensions( - self, gpx_with_extensions: GPX - ) -> None: - """Test that precision reduction preserves all extensions.""" - reduced = _apply_precision( - gpx_with_extensions, coord_precision=4, elevation_precision=1 - ) - - # Waypoint extensions preserved - assert reduced.wpt[0].extensions is not None - assert reduced.wpt[0].extensions.get_text("rating", namespace=CUSTOM_NS) == "5" - - # Route extensions preserved - assert reduced.rte[0].extensions is not None - assert reduced.rte[0].rtept[0].extensions is not None - - # Track extensions preserved - assert reduced.trk[0].extensions is not None - assert reduced.trk[0].trkseg[0].extensions is not None - assert reduced.trk[0].trkseg[0].trkpt[0].extensions is not None diff --git a/tests/test_io_convert.py b/tests/test_io_convert.py index 72e299a..2ca4b32 100644 --- a/tests/test_io_convert.py +++ b/tests/test_io_convert.py @@ -15,6 +15,8 @@ Track, TrackSegment, Waypoint, + convert_file, + detect_format, from_geo_interface, from_wkb, from_wkt, @@ -24,10 +26,6 @@ ) from gpx.types import Latitude, Longitude -# ============================================================================= -# Test fixtures -# ============================================================================= - @pytest.fixture def sample_gpx() -> GPX: @@ -137,11 +135,6 @@ def sample_kml() -> str: """ -# ============================================================================= -# GeoJSON conversion tests -# ============================================================================= - - class TestGeoJSONConversion: """Tests for GeoJSON conversion functionality.""" @@ -236,11 +229,6 @@ def test_geojson_roundtrip(self, sample_gpx: GPX) -> None: assert len(gpx2.trk) == len(sample_gpx.trk) -# ============================================================================= -# KML conversion tests -# ============================================================================= - - class TestKMLConversion: """Tests for KML conversion functionality.""" @@ -341,11 +329,6 @@ def test_kml_multigeometry_for_multi_segment_track(self, tmp_path: Path) -> None assert "" in content -# ============================================================================= -# WKT conversion tests -# ============================================================================= - - class TestWKTConversion: """Tests for WKT conversion functionality.""" @@ -458,11 +441,6 @@ def test_wkt_roundtrip(self) -> None: assert float(gpx2.wpt[0].lat) == pytest.approx(float(gpx1.wpt[0].lat), rel=1e-3) -# ============================================================================= -# WKB conversion tests -# ============================================================================= - - class TestWKBConversion: """Tests for WKB conversion functionality.""" @@ -648,11 +626,6 @@ def test_wkb_roundtrip(self) -> None: assert float(gpx2.wpt[0].lat) == pytest.approx(float(gpx1.wpt[0].lat), rel=1e-3) -# ============================================================================= -# GPX file I/O tests -# ============================================================================= - - class TestGPXFileIO: """Tests for GPX file reading and writing.""" @@ -673,11 +646,6 @@ def test_write_gpx_function(self, sample_gpx: GPX, tmp_path: Path) -> None: assert " None: from_wkb(wkb) -# ============================================================================= -# Additional GeoJSON conversion tests for missing coverage -# ============================================================================= - - class TestGeoJSONEdgeCases: """Tests for edge cases in GeoJSON conversion.""" @@ -854,11 +817,6 @@ def test_from_geo_interface_track_with_properties(self) -> None: assert gpx.trk[0].type == "Track Type" -# ============================================================================= -# WKB edge cases for missing coverage -# ============================================================================= - - class TestWKBEdgeCases: """Tests for edge cases in WKB conversion.""" @@ -885,11 +843,6 @@ def test_from_wkb_geometrycollection(self) -> None: assert len(gpx.rte) == 1 -# ============================================================================= -# WKT edge cases for missing coverage -# ============================================================================= - - class TestWKTEdgeCases: """Tests for edge cases in WKT conversion.""" @@ -911,3 +864,97 @@ def test_from_wkt_geometrycollection_complex(self) -> None: gpx = from_wkt(wkt) assert len(gpx.wpt) == 1 assert len(gpx.rte) == 1 + + +class TestDetectFormat: + """Tests for the detect_format function.""" + + def test_detect_format_gpx(self) -> None: + """Test format detection for GPX.""" + assert detect_format(Path("file.gpx")) == "gpx" + + def test_detect_format_geojson(self) -> None: + """Test format detection for GeoJSON.""" + assert detect_format(Path("file.geojson")) == "geojson" + assert detect_format(Path("file.json")) == "geojson" + + def test_detect_format_kml(self) -> None: + """Test format detection for KML.""" + assert detect_format(Path("file.kml")) == "kml" + + def test_detect_format_unknown(self) -> None: + """Test format detection for unknown extension.""" + assert detect_format(Path("file.xyz")) is None + + def test_detect_format_accepts_strings(self) -> None: + """Test format detection with a string path.""" + assert detect_format("file.GPX") == "gpx" + + +class TestConvertFile: + """Tests for the convert_file function.""" + + def test_convert_gpx_to_geojson( + self, sample_gpx_file: Path, tmp_path: Path + ) -> None: + """Test converting GPX to GeoJSON.""" + output_file = tmp_path / "output.geojson" + formats = convert_file(sample_gpx_file, output_file) + assert formats == ("gpx", "geojson") + data = json.loads(output_file.read_text()) + assert data["type"] == "FeatureCollection" + + def test_convert_gpx_to_kml(self, sample_gpx_file: Path, tmp_path: Path) -> None: + """Test converting GPX to KML.""" + output_file = tmp_path / "output.kml" + formats = convert_file(sample_gpx_file, output_file) + assert formats == ("gpx", "kml") + assert " None: + """Test converting GeoJSON to GPX.""" + input_file = tmp_path / "input.geojson" + output_file = tmp_path / "output.gpx" + + sample_gpx.write_geojson(input_file) + + formats = convert_file(input_file, output_file) + assert formats == ("geojson", "gpx") + gpx = read_gpx(output_file) + assert len(gpx.wpt) >= 1 + + def test_convert_with_explicit_formats( + self, sample_gpx_file: Path, tmp_path: Path + ) -> None: + """Test converting with explicitly given formats.""" + output_file = tmp_path / "output.json" + formats = convert_file( + sample_gpx_file, output_file, input_format="gpx", output_format="geojson" + ) + assert formats == ("gpx", "geojson") + + def test_convert_undetectable_input_format(self, tmp_path: Path) -> None: + """An undetectable input format raises a ValueError.""" + with pytest.raises(ValueError, match="Could not detect input format"): + convert_file(tmp_path / "input.xyz", tmp_path / "output.gpx") + + def test_convert_undetectable_output_format( + self, sample_gpx_file: Path, tmp_path: Path + ) -> None: + """An undetectable output format raises a ValueError.""" + with pytest.raises(ValueError, match="Could not detect output format"): + convert_file(sample_gpx_file, tmp_path / "output.xyz") + + def test_convert_unsupported_input_format( + self, sample_gpx_file: Path, tmp_path: Path + ) -> None: + """An unsupported input format raises a ValueError.""" + with pytest.raises(ValueError, match="Unsupported input format"): + convert_file(sample_gpx_file, tmp_path / "output.gpx", input_format="csv") + + def test_convert_unsupported_output_format( + self, sample_gpx_file: Path, tmp_path: Path + ) -> None: + """An unsupported output format raises a ValueError.""" + with pytest.raises(ValueError, match="Unsupported output format"): + convert_file(sample_gpx_file, tmp_path / "output.gpx", output_format="csv") diff --git a/tests/test_operations.py b/tests/test_operations.py new file mode 100644 index 0000000..b222d5e --- /dev/null +++ b/tests/test_operations.py @@ -0,0 +1,762 @@ +"""Tests for gpx.operations module - Edit and merge operations.""" + +import datetime as dt +from decimal import Decimal + +import pytest + +from gpx import ( + GPX, + Metadata, + Person, + Route, + Track, + TrackSegment, + Waypoint, + crop, + filter_points, + merge, + reduce_precision, + reverse, + shift_time, + simplify, + smooth, + split, + strip_extensions, + strip_metadata, + trim, +) +from gpx.types import Latitude, Longitude + +from .conftest import CUSTOM_NS, GARMIN_TPX_NS + + +def make_track_gpx(points: list[Waypoint]) -> GPX: + """Create a GPX with a single track containing a single segment.""" + return GPX(trk=[Track(trkseg=[TrackSegment(trkpt=points)])]) + + +def make_points( + coords: list[tuple[str, str]], + *, + minutes_apart: int | None = 1, +) -> list[Waypoint]: + """Create waypoints from (lat, lon) strings, optionally with timestamps.""" + start = dt.datetime(2024, 1, 15, 10, 0, 0, tzinfo=dt.UTC) + return [ + Waypoint( + lat=Latitude(lat), + lon=Longitude(lon), + time=start + dt.timedelta(minutes=i * minutes_apart) + if minutes_apart is not None + else None, + ) + for i, (lat, lon) in enumerate(coords) + ] + + +class TestFilterPoints: + """Tests for the filter_points operation.""" + + def test_filter_points_keeps_matching(self, sample_gpx: GPX) -> None: + """Points matching the predicate are kept.""" + filtered = filter_points(sample_gpx, lambda _: True) + assert len(filtered.wpt) == len(sample_gpx.wpt) + assert len(filtered.rte) == len(sample_gpx.rte) + assert len(filtered.trk) == len(sample_gpx.trk) + + def test_filter_points_drops_empty_containers(self, sample_gpx: GPX) -> None: + """Routes, segments and tracks that end up empty are dropped.""" + filtered = filter_points(sample_gpx, lambda _: False) + assert filtered.wpt == [] + assert filtered.rte == [] + assert filtered.trk == [] + + def test_filter_points_does_not_mutate_input(self, sample_gpx: GPX) -> None: + """The input GPX is left unchanged.""" + original_wpt_count = len(sample_gpx.wpt) + filter_points(sample_gpx, lambda _: False) + assert len(sample_gpx.wpt) == original_wpt_count + + +class TestCrop: + """Tests for the crop operation.""" + + def test_crop(self, sample_gpx: GPX) -> None: + """Test cropping GPX data.""" + cropped = crop( + sample_gpx, + min_lat=52.515, + max_lat=52.525, + min_lon=13.400, + max_lon=13.410, + ) + # Should keep some points but filter out those outside bounds + assert len(cropped.wpt) >= 0 + assert len(cropped.trk) >= 0 + + def test_crop_excludes_all(self, sample_gpx: GPX) -> None: + """Test cropping with bounds that exclude all points.""" + cropped = crop( + sample_gpx, + min_lat=0.0, + max_lat=1.0, + min_lon=0.0, + max_lon=1.0, + ) + assert len(cropped.wpt) == 0 + assert len(cropped.rte) == 0 + assert len(cropped.trk) == 0 + + def test_crop_partial_bounds(self, sample_gpx: GPX) -> None: + """Bounds that are None are not enforced.""" + kept = crop(sample_gpx, min_lat=0.0) + assert len(kept.wpt) == 1 + + dropped = crop(sample_gpx, min_lat=53.0) + assert len(dropped.wpt) == 0 + + +class TestTrim: + """Tests for the trim operation.""" + + def test_trim(self, sample_gpx: GPX) -> None: + """Test trimming GPX data to a time range.""" + start = dt.datetime(2024, 1, 15, 10, 0, 30, tzinfo=dt.UTC) + end = dt.datetime(2024, 1, 15, 10, 1, 30, tzinfo=dt.UTC) + trimmed = trim(sample_gpx, start=start, end=end) + # Should keep only points within time range + assert len(trimmed.trk) > 0 + for track in trimmed.trk: + for seg in track.trkseg: + for pt in seg.trkpt: + if pt.time: + assert pt.time >= start + assert pt.time <= end + + def test_trim_keeps_points_without_timestamps(self, sample_gpx: GPX) -> None: + """Points without timestamps are kept (route points have no time).""" + start = dt.datetime(2030, 1, 1, tzinfo=dt.UTC) + trimmed = trim(sample_gpx, start=start) + assert len(trimmed.rte) == 1 + + +class TestReverse: + """Tests for the reverse operation.""" + + def test_reverse_routes(self, sample_gpx: GPX) -> None: + """Test reversing routes.""" + original_first_lat = float(sample_gpx.rte[0].rtept[0].lat) + original_last_lat = float(sample_gpx.rte[0].rtept[-1].lat) + + reversed_gpx = reverse(sample_gpx, routes=True, tracks=False) + + new_first_lat = float(reversed_gpx.rte[0].rtept[0].lat) + new_last_lat = float(reversed_gpx.rte[0].rtept[-1].lat) + + assert new_first_lat == pytest.approx(original_last_lat, rel=1e-3) + assert new_last_lat == pytest.approx(original_first_lat, rel=1e-3) + + def test_reverse_tracks(self, sample_gpx: GPX) -> None: + """Test reversing tracks.""" + original_first_lat = float(sample_gpx.trk[0].trkseg[0].trkpt[0].lat) + original_last_lat = float(sample_gpx.trk[0].trkseg[0].trkpt[-1].lat) + + reversed_gpx = reverse(sample_gpx, routes=False, tracks=True) + + new_first_lat = float(reversed_gpx.trk[0].trkseg[0].trkpt[0].lat) + new_last_lat = float(reversed_gpx.trk[0].trkseg[0].trkpt[-1].lat) + + assert new_first_lat == pytest.approx(original_last_lat, rel=1e-3) + assert new_last_lat == pytest.approx(original_first_lat, rel=1e-3) + + def test_reverse_defaults_to_both(self, sample_gpx: GPX) -> None: + """By default, both routes and tracks are reversed.""" + reversed_gpx = reverse(sample_gpx) + assert float(reversed_gpx.rte[0].rtept[0].lat) == pytest.approx( + float(sample_gpx.rte[0].rtept[-1].lat), rel=1e-3 + ) + assert float(reversed_gpx.trk[0].trkseg[0].trkpt[0].lat) == pytest.approx( + float(sample_gpx.trk[0].trkseg[0].trkpt[-1].lat), rel=1e-3 + ) + + +class TestStripMetadata: + """Tests for the strip_metadata operation.""" + + def test_strip_all_metadata(self, sample_gpx: GPX) -> None: + """With no field flags, the entire metadata element is removed.""" + stripped = strip_metadata(sample_gpx) + assert stripped.metadata is None + + def test_strip_selected_fields(self, sample_gpx: GPX) -> None: + """Only the selected fields are removed from the metadata.""" + stripped = strip_metadata(sample_gpx, name=True) + assert stripped.metadata is not None + assert stripped.metadata.name is None + assert stripped.metadata.desc == "Test description" + + def test_strip_author_copyright_time_keywords_links(self) -> None: + """All field flags strip their corresponding metadata field.""" + gpx = GPX( + metadata=Metadata( + name="Name", + desc="Description", + author=Person(name="Author"), + time=dt.datetime(2024, 1, 15, tzinfo=dt.UTC), + keywords="some, keywords", + ) + ) + stripped = strip_metadata( + gpx, author=True, time=True, keywords=True, links=True + ) + assert stripped.metadata is not None + assert stripped.metadata.author is None + assert stripped.metadata.time is None + assert stripped.metadata.keywords is None + assert stripped.metadata.link == [] + assert stripped.metadata.name == "Name" + assert stripped.metadata.desc == "Description" + + def test_strip_fields_without_metadata(self) -> None: + """Stripping fields from a GPX without metadata is a no-op.""" + gpx = GPX() + stripped = strip_metadata(gpx, name=True) + assert stripped.metadata is None + + def test_strip_all_metadata_preserves_nsmap(self, sample_gpx: GPX) -> None: + """Stripping all metadata preserves the namespace mappings.""" + sample_gpx.nsmap = { + "": "http://www.topografix.com/GPX/1/1", + "gpxtpx": GARMIN_TPX_NS, + } + stripped = strip_metadata(sample_gpx) + assert stripped.nsmap == sample_gpx.nsmap + + +class TestReducePrecision: + """Tests for the reduce_precision operation.""" + + def test_reduce_precision(self, sample_gpx: GPX) -> None: + """Test reducing coordinate and elevation precision.""" + reduced = reduce_precision( + sample_gpx, coordinate_precision=2, elevation_precision=0 + ) + + # Check waypoint precision + for wpt in reduced.wpt: + lat_str = str(wpt.lat) + if "." in lat_str: + assert len(lat_str.split(".")[1]) <= 2 + if wpt.ele: + ele_str = str(wpt.ele) + if "." in ele_str: + assert len(ele_str.split(".")[1]) <= 1 + + def test_reduce_precision_no_elevation(self) -> None: + """Test reducing precision when there is no elevation.""" + gpx = GPX( + wpt=[Waypoint(lat=Latitude("52.5200123"), lon=Longitude("13.4050456"))] + ) + reduced = reduce_precision(gpx, coordinate_precision=3) + lat_str = str(reduced.wpt[0].lat) + assert lat_str == "52.52" + + +class TestNsmapPreservation: + """Tests that the edit operations preserve namespace mappings.""" + + def test_operations_preserve_nsmap(self, sample_gpx: GPX) -> None: + """All edit operations must propagate gpx.nsmap so prefixes survive a round-trip. + + Regression test: previously crop / trim / reverse / reduce_precision / + strip_metadata silently dropped nsmap, which meant `gpx edit` on a + file with custom prefixes (e.g. gpxtpx) emitted auto-generated + ns0/ns1 prefixes instead. + """ + sample_gpx.nsmap = { + "": "http://www.topografix.com/GPX/1/1", + "gpxtpx": GARMIN_TPX_NS, + } + + cropped = crop(sample_gpx, min_lat=0.0, max_lat=90.0, min_lon=0.0, max_lon=90.0) + assert cropped.nsmap == sample_gpx.nsmap + + trimmed = trim(sample_gpx) + assert trimmed.nsmap == sample_gpx.nsmap + + reversed_gpx = reverse(sample_gpx) + assert reversed_gpx.nsmap == sample_gpx.nsmap + + reduced = reduce_precision( + sample_gpx, coordinate_precision=3, elevation_precision=1 + ) + assert reduced.nsmap == sample_gpx.nsmap + + split_gpx = split(sample_gpx, time_gap=dt.timedelta(minutes=10)) + assert split_gpx.nsmap == sample_gpx.nsmap + + simplified = simplify(sample_gpx, tolerance=10.0) + assert simplified.nsmap == sample_gpx.nsmap + + smoothed = smooth(sample_gpx, window=3) + assert smoothed.nsmap == sample_gpx.nsmap + + shifted = shift_time(sample_gpx, dt.timedelta(hours=1)) + assert shifted.nsmap == sample_gpx.nsmap + + stripped = strip_extensions(sample_gpx) + assert stripped.nsmap == sample_gpx.nsmap + + +class TestOperationsWithExtensions: + """Test that the edit operations preserve extensions.""" + + def test_crop_preserves_waypoint_extensions(self, gpx_with_extensions: GPX) -> None: + """Test that crop preserves extensions on remaining waypoints.""" + cropped = crop( + gpx_with_extensions, + min_lat=52.0, + max_lat=53.0, + min_lon=13.0, + max_lon=14.0, + ) + + # Waypoint should be preserved with extensions + assert len(cropped.wpt) == 1 + assert cropped.wpt[0].extensions is not None + assert cropped.wpt[0].extensions.get_text("rating", namespace=CUSTOM_NS) == "5" + + def test_crop_preserves_track_extensions(self, gpx_with_extensions: GPX) -> None: + """Test that crop preserves track and track point extensions.""" + cropped = crop( + gpx_with_extensions, + min_lat=52.0, + max_lat=53.0, + min_lon=13.0, + max_lon=14.0, + ) + + # Track extensions preserved + assert cropped.trk[0].extensions is not None + assert ( + cropped.trk[0].extensions.get_text("activity", namespace=CUSTOM_NS) + == "running" + ) + + # Track point extensions preserved + assert cropped.trk[0].trkseg[0].trkpt[0].extensions is not None + assert ( + cropped.trk[0] + .trkseg[0] + .trkpt[0] + .extensions.get_text("hr", namespace=GARMIN_TPX_NS) + == "140" + ) + + def test_crop_preserves_route_extensions(self, gpx_with_extensions: GPX) -> None: + """Test that crop preserves route and route point extensions.""" + cropped = crop( + gpx_with_extensions, + min_lat=52.0, + max_lat=53.0, + min_lon=13.0, + max_lon=14.0, + ) + + # Route extensions preserved + assert cropped.rte[0].extensions is not None + assert ( + cropped.rte[0].extensions.get_text("type", namespace=CUSTOM_NS) == "scenic" + ) + + # Route point extensions preserved + assert cropped.rte[0].rtept[0].extensions is not None + + def test_trim_preserves_extensions(self, gpx_with_extensions: GPX) -> None: + """Test that trim preserves extensions on remaining elements.""" + start = dt.datetime(2024, 1, 15, 10, 0, 30, tzinfo=dt.UTC) + end = dt.datetime(2024, 1, 15, 10, 1, 30, tzinfo=dt.UTC) + trimmed = trim(gpx_with_extensions, start=start, end=end) + + # Track extensions preserved + if trimmed.trk: + assert trimmed.trk[0].extensions is not None + assert ( + trimmed.trk[0].extensions.get_text("activity", namespace=CUSTOM_NS) + == "running" + ) + + def test_reverse_preserves_all_extensions(self, gpx_with_extensions: GPX) -> None: + """Test that reverse preserves all extensions.""" + reversed_gpx = reverse(gpx_with_extensions) + + # Route extensions preserved + assert reversed_gpx.rte[0].extensions is not None + assert ( + reversed_gpx.rte[0].extensions.get_text("type", namespace=CUSTOM_NS) + == "scenic" + ) + + # Track extensions preserved + assert reversed_gpx.trk[0].extensions is not None + assert ( + reversed_gpx.trk[0].extensions.get_text("activity", namespace=CUSTOM_NS) + == "running" + ) + + # Track segment extensions preserved + assert reversed_gpx.trk[0].trkseg[0].extensions is not None + + # Track point extensions preserved (order reversed) + # First point after reverse was the last point (hr=150) + first_point_ext = reversed_gpx.trk[0].trkseg[0].trkpt[0].extensions + assert first_point_ext is not None + assert first_point_ext.get_text("hr", namespace=GARMIN_TPX_NS) == "150" + + def test_reduce_precision_preserves_all_extensions( + self, gpx_with_extensions: GPX + ) -> None: + """Test that precision reduction preserves all extensions.""" + reduced = reduce_precision( + gpx_with_extensions, coordinate_precision=4, elevation_precision=1 + ) + + # Waypoint extensions preserved + assert reduced.wpt[0].extensions is not None + assert reduced.wpt[0].extensions.get_text("rating", namespace=CUSTOM_NS) == "5" + + # Route extensions preserved + assert reduced.rte[0].extensions is not None + assert reduced.rte[0].rtept[0].extensions is not None + + # Track extensions preserved + assert reduced.trk[0].extensions is not None + assert reduced.trk[0].trkseg[0].extensions is not None + assert reduced.trk[0].trkseg[0].trkpt[0].extensions is not None + + +class TestMerge: + """Tests for the merge operation.""" + + def test_merge(self, sample_gpx: GPX) -> None: + """Merging concatenates waypoints, routes and tracks.""" + merged = merge([sample_gpx, sample_gpx]) + assert len(merged.wpt) == 2 + assert len(merged.rte) == 2 + assert len(merged.trk) == 2 + + def test_merge_empty(self) -> None: + """Merging no GPX instances yields an empty GPX.""" + merged = merge([]) + assert merged.wpt == [] + assert merged.rte == [] + assert merged.trk == [] + + def test_merge_with_creator(self, sample_gpx: GPX) -> None: + """The creator can be set on the merged GPX.""" + merged = merge([sample_gpx], creator="MergeApp") + assert merged.creator == "MergeApp" + + def test_merge_accepts_iterables(self, sample_gpx: GPX) -> None: + """Any iterable of GPX instances can be merged.""" + merged = merge(gpx for gpx in (sample_gpx, sample_gpx, sample_gpx)) + assert len(merged.trk) == 3 + + +class TestSplit: + """Tests for the split operation.""" + + def test_split_requires_threshold(self, sample_gpx: GPX) -> None: + """At least one of time_gap or distance_gap must be given.""" + with pytest.raises(ValueError, match="time_gap or distance_gap"): + split(sample_gpx) + + def test_split_time_gap(self) -> None: + """Segments are split where the time gap exceeds the threshold.""" + start = dt.datetime(2024, 1, 15, 10, 0, 0, tzinfo=dt.UTC) + # 30-minute gap between the second and third point + offsets_minutes = (0, 1, 31, 32) + points = [ + Waypoint( + lat=Latitude("52.000"), + lon=Longitude("4.000"), + time=start + dt.timedelta(minutes=offset), + ) + for offset in offsets_minutes + ] + gpx = make_track_gpx(points) + + split_gpx = split(gpx, time_gap=dt.timedelta(minutes=10)) + assert [len(seg.trkpt) for seg in split_gpx.trk[0].trkseg] == [2, 2] + + def test_split_distance_gap(self) -> None: + """Segments are split where the distance gap exceeds the threshold.""" + # ~111 m between consecutive points, then a ~1.1 km jump + coords = [("52.000", "4.000"), ("52.001", "4.000"), ("52.011", "4.000")] + gpx = make_track_gpx(make_points(coords)) + + split_gpx = split(gpx, distance_gap=500.0) + assert [len(seg.trkpt) for seg in split_gpx.trk[0].trkseg] == [2, 1] + + def test_split_no_gap_unchanged(self, sample_gpx: GPX) -> None: + """Without gaps exceeding the thresholds, segments are unchanged.""" + split_gpx = split(sample_gpx, time_gap=dt.timedelta(minutes=10)) + assert len(split_gpx.trk[0].trkseg) == 1 + assert len(split_gpx.trk[0].trkseg[0].trkpt) == 3 + + def test_split_points_without_time_never_split(self) -> None: + """Points without timestamps never trigger a time-based split.""" + gpx = make_track_gpx(make_points([("52.000", "4.000")] * 3, minutes_apart=None)) + split_gpx = split(gpx, time_gap=dt.timedelta(seconds=1)) + assert len(split_gpx.trk[0].trkseg) == 1 + + def test_split_preserves_segment_extensions(self, gpx_with_extensions: GPX) -> None: + """Each resulting segment keeps the original segment's extensions.""" + # Points are 1 minute apart; split at 30 seconds → 3 segments + split_gpx = split(gpx_with_extensions, time_gap=dt.timedelta(seconds=30)) + segments = split_gpx.trk[0].trkseg + assert len(segments) == 3 + for segment in segments: + assert segment.extensions is not None + assert segment.extensions.get_text("lap", namespace=CUSTOM_NS) == "1" + + def test_split_routes_and_waypoints_unchanged(self, sample_gpx: GPX) -> None: + """Waypoints and routes are not split.""" + split_gpx = split(sample_gpx, distance_gap=0.001) + assert len(split_gpx.wpt) == len(sample_gpx.wpt) + assert len(split_gpx.rte) == len(sample_gpx.rte) + assert len(split_gpx.rte[0].rtept) == len(sample_gpx.rte[0].rtept) + + def test_split_does_not_mutate_input(self, sample_gpx: GPX) -> None: + """The input GPX is left unchanged.""" + split(sample_gpx, distance_gap=0.001) + assert len(sample_gpx.trk[0].trkseg) == 1 + + +class TestSimplify: + """Tests for the simplify operation.""" + + def test_simplify_invalid_tolerance(self, sample_gpx: GPX) -> None: + """A non-positive tolerance raises a ValueError.""" + with pytest.raises(ValueError, match="tolerance"): + simplify(sample_gpx, 0) + with pytest.raises(ValueError, match="tolerance"): + simplify(sample_gpx, -1.0) + + def test_simplify_drops_collinear_points(self) -> None: + """Points on a straight line are reduced to the endpoints.""" + coords = [("52.000", "4.000"), ("52.001", "4.000"), ("52.002", "4.000")] + gpx = make_track_gpx(make_points(coords)) + + simplified = simplify(gpx, tolerance=5.0) + points = simplified.trk[0].trkseg[0].trkpt + assert len(points) == 2 + assert str(points[0].lat) == "52.000" + assert str(points[-1].lat) == "52.002" + + def test_simplify_keeps_significant_points(self) -> None: + """Points deviating more than the tolerance are kept.""" + # Middle point deviates ~68 m east of the line + coords = [("52.000", "4.000"), ("52.001", "4.001"), ("52.002", "4.000")] + gpx = make_track_gpx(make_points(coords)) + + simplified = simplify(gpx, tolerance=10.0) + assert len(simplified.trk[0].trkseg[0].trkpt) == 3 + + def test_simplify_routes(self) -> None: + """Route points are simplified as well.""" + gpx = GPX( + rte=[ + Route( + rtept=make_points( + [("52.000", "4.000"), ("52.001", "4.000"), ("52.002", "4.000")], + minutes_apart=None, + ) + ) + ] + ) + simplified = simplify(gpx, tolerance=5.0) + assert len(simplified.rte[0].rtept) == 2 + + def test_simplify_waypoints_unchanged(self, sample_gpx: GPX) -> None: + """Waypoints are left unchanged.""" + simplified = simplify(sample_gpx, tolerance=1000.0) + assert len(simplified.wpt) == len(sample_gpx.wpt) + + def test_simplify_short_lists_unchanged(self) -> None: + """Lists of fewer than three points are left unchanged.""" + coords = [("52.000", "4.000"), ("52.001", "4.000")] + gpx = make_track_gpx(make_points(coords)) + simplified = simplify(gpx, tolerance=1000.0) + assert len(simplified.trk[0].trkseg[0].trkpt) == 2 + + def test_simplify_preserves_extensions_of_kept_points( + self, gpx_with_extensions: GPX + ) -> None: + """Extensions of the points that are kept are preserved.""" + simplified = simplify(gpx_with_extensions, tolerance=1000.0) + first_point = simplified.trk[0].trkseg[0].trkpt[0] + assert first_point.extensions is not None + assert first_point.extensions.get_text("hr", namespace=GARMIN_TPX_NS) == "140" + + +class TestSmooth: + """Tests for the smooth operation.""" + + def test_smooth_invalid_window(self, sample_gpx: GPX) -> None: + """An even or too small window raises a ValueError.""" + for window in (0, 1, 2, 4): + with pytest.raises(ValueError, match="window"): + smooth(sample_gpx, window=window) + + def test_smooth_moves_noisy_point(self) -> None: + """A noisy point is moved towards its neighbors.""" + coords = [("52.000", "4.000"), ("52.002", "4.000"), ("52.000", "4.000")] + gpx = make_track_gpx(make_points(coords)) + + smoothed = smooth(gpx, window=3) + middle = smoothed.trk[0].trkseg[0].trkpt[1] + # Mean of 52.000, 52.002, 52.000 + assert float(middle.lat) == pytest.approx(52.000667, abs=1e-6) + + def test_smooth_endpoints_unchanged(self) -> None: + """The first and last point are always left unchanged.""" + coords = [("52.000", "4.000"), ("52.002", "4.001"), ("52.000", "4.002")] + gpx = make_track_gpx(make_points(coords)) + + smoothed = smooth(gpx, window=3) + points = smoothed.trk[0].trkseg[0].trkpt + assert str(points[0].lat) == "52.000" + assert str(points[0].lon) == "4.000" + assert str(points[-1].lat) == "52.000" + assert str(points[-1].lon) == "4.002" + + def test_smooth_elevations(self) -> None: + """Elevations are smoothed with the same moving average.""" + points = make_points([("52.000", "4.000")] * 3) + points = [ + Waypoint(lat=p.lat, lon=p.lon, time=p.time, ele=Decimal(ele)) + for p, ele in zip(points, ("10.0", "40.0", "10.0"), strict=True) + ] + gpx = make_track_gpx(points) + + smoothed = smooth(gpx, window=3) + middle = smoothed.trk[0].trkseg[0].trkpt[1] + assert middle.ele is not None + assert float(middle.ele) == pytest.approx(20.0) + + def test_smooth_coordinates_flag(self) -> None: + """With coordinates=False, the coordinates are left unchanged.""" + coords = [("52.000", "4.000"), ("52.002", "4.000"), ("52.000", "4.000")] + gpx = make_track_gpx(make_points(coords)) + + smoothed = smooth(gpx, window=3, coordinates=False) + assert str(smoothed.trk[0].trkseg[0].trkpt[1].lat) == "52.002" + + def test_smooth_points_without_elevation(self) -> None: + """Points without elevation are left without elevation.""" + coords = [("52.000", "4.000"), ("52.002", "4.000"), ("52.000", "4.000")] + gpx = make_track_gpx(make_points(coords)) + + smoothed = smooth(gpx, window=3) + assert all(p.ele is None for p in smoothed.trk[0].trkseg[0].trkpt) + + def test_smooth_short_lists_unchanged(self) -> None: + """Lists of fewer than three points are left unchanged.""" + coords = [("52.000", "4.000"), ("52.002", "4.000")] + gpx = make_track_gpx(make_points(coords)) + + smoothed = smooth(gpx, window=3) + assert str(smoothed.trk[0].trkseg[0].trkpt[1].lat) == "52.002" + + def test_smooth_preserves_other_fields(self, gpx_with_extensions: GPX) -> None: + """Timestamps and extensions are preserved on smoothed points.""" + smoothed = smooth(gpx_with_extensions, window=3) + middle = smoothed.trk[0].trkseg[0].trkpt[1] + assert middle.time == gpx_with_extensions.trk[0].trkseg[0].trkpt[1].time + assert middle.extensions is not None + assert middle.extensions.get_text("hr", namespace=GARMIN_TPX_NS) == "145" + + +class TestShiftTime: + """Tests for the shift_time operation.""" + + def test_shift_time_forward(self, sample_gpx: GPX) -> None: + """All point timestamps are shifted by the delta.""" + delta = dt.timedelta(hours=2) + shifted = shift_time(sample_gpx, delta) + + original_time = sample_gpx.wpt[0].time + assert original_time is not None + assert shifted.wpt[0].time == original_time + delta + for original, new in zip( + sample_gpx.trk[0].trkseg[0].trkpt, + shifted.trk[0].trkseg[0].trkpt, + strict=True, + ): + assert original.time is not None + assert new.time == original.time + delta + + def test_shift_time_negative(self, sample_gpx: GPX) -> None: + """Negative deltas shift the timestamps backwards.""" + delta = dt.timedelta(minutes=-30) + shifted = shift_time(sample_gpx, delta) + original_time = sample_gpx.wpt[0].time + assert original_time is not None + assert shifted.wpt[0].time == original_time + delta + + def test_shift_time_points_without_time(self, sample_gpx: GPX) -> None: + """Points without a timestamp are left unchanged.""" + shifted = shift_time(sample_gpx, dt.timedelta(hours=1)) + assert all(p.time is None for p in shifted.rte[0].rtept) + + def test_shift_time_metadata_unchanged(self) -> None: + """The metadata timestamp is not shifted.""" + time = dt.datetime(2024, 1, 15, tzinfo=dt.UTC) + gpx = GPX(metadata=Metadata(time=time)) + shifted = shift_time(gpx, dt.timedelta(hours=1)) + assert shifted.metadata is not None + assert shifted.metadata.time == time + + def test_shift_time_does_not_mutate_input(self, sample_gpx: GPX) -> None: + """The input GPX is left unchanged.""" + original_time = sample_gpx.wpt[0].time + shift_time(sample_gpx, dt.timedelta(hours=1)) + assert sample_gpx.wpt[0].time == original_time + + +class TestStripExtensions: + """Tests for the strip_extensions operation.""" + + def test_strip_extensions_all_levels(self, gpx_with_extensions: GPX) -> None: + """Extensions are removed from all levels of the GPX.""" + stripped = strip_extensions(gpx_with_extensions) + + assert stripped.extensions is None + assert stripped.metadata is not None + assert stripped.metadata.extensions is None + assert all(w.extensions is None for w in stripped.wpt) + for route in stripped.rte: + assert route.extensions is None + assert all(p.extensions is None for p in route.rtept) + for track in stripped.trk: + assert track.extensions is None + for segment in track.trkseg: + assert segment.extensions is None + assert all(p.extensions is None for p in segment.trkpt) + + def test_strip_extensions_without_extensions(self, sample_gpx: GPX) -> None: + """Stripping a GPX without extensions is a no-op.""" + stripped = strip_extensions(sample_gpx) + assert stripped.extensions is None + assert len(stripped.wpt) == len(sample_gpx.wpt) + assert len(stripped.trk) == len(sample_gpx.trk) + + def test_strip_extensions_does_not_mutate_input( + self, gpx_with_extensions: GPX + ) -> None: + """The input GPX is left unchanged.""" + strip_extensions(gpx_with_extensions) + assert gpx_with_extensions.extensions is not None + assert gpx_with_extensions.trk[0].trkseg[0].trkpt[0].extensions is not None