22961397810

Committed 11 Mar 2026 03:49PM UTC coverage: 65.598% (-0.8%) from 66.351%

Build # 22961397810

Build Type

Pull #29

github

Committed by

web-flow

Commit Message

Merge 580e87afc into c58dfbb3f

Pull Request Pull Request #29: Added wave processing from RBSP data

Coverage Stats

8 of 16 new or added lines in 2 files covered. (50.0%)

16 existing lines in 1 file now uncovered.

1739 of 2651 relevant lines covered (65.6%)

1.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

39.13

/el_paso/processing/tle.py

# SPDX-FileCopyrightText: 2026 GFZ Helmholtz Centre for Geosciences
# SPDX-FileContributor: Sahil Jhawar
#
# SPDX-License-Identifier: Apache-2.0

import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path

import numpy as np
from numpy.typing import NDArray
from skyfield.api import EarthSatellite, load

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TLE:
    """A class to track satellite positions using TLE data.

    Args:
        tle_filename (str | Path): The file path containing the TLE data.

    Attributes:
        tle_filename (str | Path): The file path containing the TLE data.
        tle_data (list of tuple): A list containing pairs of TLE lines for each
            satellite.
        satellite_name (str): The name of the satellite based on the first line of
            the TLE data.
    """

    tle_filename: str | Path
    tle_data: list[tuple[str, str]] = field(init=False)
    satellite_name: str = field(init=False)
    tle_time_list: list[datetime] = field(init=False)

    def __post_init__(self) -> None:
        """Initialize parsed and derived values from the input TLE file."""
        tle_path = Path(self.tle_filename)
        tle_data, satellite_name = self._read_tle_file(tle_path)
        tle_time_list = self.get_tle_time(tle_data)

        object.__setattr__(self, "tle_data", tle_data)
        object.__setattr__(self, "satellite_name", satellite_name)
        object.__setattr__(self, "tle_time_list", tle_time_list)

    def _read_tle_file(self, filename: Path) -> tuple[list[tuple[str, str]], str]:
        """Read TLE data from a file.

        Args:
            filename (Path): The file path containing the TLE data.

        Returns:
            tuple[list[tuple[str, str]], str]: A tuple containing:
                - tle_data: A list of satellite TLE line pairs.
                - satellite_name: The name of the satellite.
        """
        with filename.open() as file:
            lines = file.readlines()

        tle_data = [(lines[i].strip(), lines[i + 1].strip()) for i in range(0, len(lines), 2)]
        satellite_name = tle_data[0][0].split()[1]

        return tle_data, satellite_name

    def get_tle_time(self, tle_data: list[tuple[str, str]]) -> list[datetime]:
        """Generate a list of UTC datetime objects from the TLE data.

        Args:
            tle_data (list[tuple[str, str]]): The TLE data for the satellite.

        Returns:
            list[datetime]: A list of UTC datetime objects.
        """
        tle_times = []
        for tle in tle_data:
            year, doy = str(tle[0].split()[3])[:2], str(tle[0].split()[3])[2:]
            tle_times.append(
                datetime(2000 + int(year), 1, 1, 0, 0, 0, tzinfo=timezone.utc) + timedelta(days=float(doy))
            )

        return tle_times

    def calculate_geo_coords(self) -> NDArray[np.float64]:
        """Calculate GEO coordinates (x, y, z) in kilometers using Skyfield.

        Returns:
            NDArray[np.float64]: GEO coordinates for each TLE epoch in a shape
                `(n, 3)` array, where columns are `(x, y, z)` in kilometers.
        """
        timescale = load.timescale()
        geo_coordinates = []

        for tle_lines, tle_time in zip(self.tle_data, self.tle_time_list, strict=True):
            satellite = EarthSatellite(tle_lines[0], tle_lines[1], self.satellite_name)
            geocentric = satellite.at(timescale.from_datetime(tle_time))
            xyz = geocentric.xyz.km
            geo_coordinates.append(xyz)

        geo_coordinates = np.asarray(geo_coordinates, dtype=np.float64)

        if np.isnan(geo_coordinates).any():
            nan_indices = np.where(np.isnan(geo_coordinates).any(axis=1))[0]
            logger.warning(
                f"NaN values found in GEO coordinates at indices: {', '.join(str(idx) for idx in nan_indices)}. "
                "Check the TLE file at these indices."
            )

        return np.asarray(geo_coordinates, dtype=np.float64)

1	# SPDX-FileCopyrightText: 2026 GFZ Helmholtz Centre for Geosciences
2	# SPDX-FileContributor: Sahil Jhawar
3	#
4	# SPDX-License-Identifier: Apache-2.0
5
6	import logging	2✔
7	from dataclasses import dataclass, field	2✔
8	from datetime import datetime, timedelta, timezone	2✔
9	from pathlib import Path	2✔
10
11	import numpy as np	2✔
12	from numpy.typing import NDArray	2✔
13	from skyfield.api import EarthSatellite, load	2✔
14
15	logger = logging.getLogger(__name__)	2✔
16
17
18	@dataclass(frozen=True)	2✔
19	class TLE:	2✔
20	"""A class to track satellite positions using TLE data.
21
22	Args:
23	tle_filename (str \| Path): The file path containing the TLE data.
24
25	Attributes:
26	tle_filename (str \| Path): The file path containing the TLE data.
27	tle_data (list of tuple): A list containing pairs of TLE lines for each
28	satellite.
29	satellite_name (str): The name of the satellite based on the first line of
30	the TLE data.
31	"""
32
33	tle_filename: str \| Path	2✔
34	tle_data: list[tuple[str, str]] = field(init=False)	2✔
35	satellite_name: str = field(init=False)	2✔
36	tle_time_list: list[datetime] = field(init=False)	2✔
37
38	def __post_init__(self) -> None:	2✔
39	"""Initialize parsed and derived values from the input TLE file."""
UNCOV 40	tle_path = Path(self.tle_filename)	×
UNCOV 41	tle_data, satellite_name = self._read_tle_file(tle_path)	×
UNCOV 42	tle_time_list = self.get_tle_time(tle_data)	×
43
UNCOV 44	object.__setattr__(self, "tle_data", tle_data)	×
UNCOV 45	object.__setattr__(self, "satellite_name", satellite_name)	×
UNCOV 46	object.__setattr__(self, "tle_time_list", tle_time_list)	×
47
48	def _read_tle_file(self, filename: Path) -> tuple[list[tuple[str, str]], str]:	2✔
49	"""Read TLE data from a file.
50
51	Args:
52	filename (Path): The file path containing the TLE data.
53
54	Returns:
55	tuple[list[tuple[str, str]], str]: A tuple containing:
56	- tle_data: A list of satellite TLE line pairs.
57	- satellite_name: The name of the satellite.
58	"""
UNCOV 59	with filename.open() as file:	×
UNCOV 60	lines = file.readlines()	×
61
UNCOV 62	tle_data = [(lines[i].strip(), lines[i + 1].strip()) for i in range(0, len(lines), 2)]	×
UNCOV 63	satellite_name = tle_data[0][0].split()[1]	×
64
UNCOV 65	return tle_data, satellite_name	×
66
67	def get_tle_time(self, tle_data: list[tuple[str, str]]) -> list[datetime]:	2✔
68	"""Generate a list of UTC datetime objects from the TLE data.
69
70	Args:
71	tle_data (list[tuple[str, str]]): The TLE data for the satellite.
72
73	Returns:
74	list[datetime]: A list of UTC datetime objects.
75	"""
UNCOV 76	tle_times = []	×
UNCOV 77	for tle in tle_data:	×
UNCOV 78	year, doy = str(tle[0].split()[3])[:2], str(tle[0].split()[3])[2:]	×
UNCOV 79	tle_times.append(	×
80	datetime(2000 + int(year), 1, 1, 0, 0, 0, tzinfo=timezone.utc) + timedelta(days=float(doy))
81	)
82
UNCOV 83	return tle_times	×
84
85	def calculate_geo_coords(self) -> NDArray[np.float64]:	2✔
86	"""Calculate GEO coordinates (x, y, z) in kilometers using Skyfield.
87
88	Returns:
89	NDArray[np.float64]: GEO coordinates for each TLE epoch in a shape
90	`(n, 3)` array, where columns are `(x, y, z)` in kilometers.
91	"""
92	timescale = load.timescale()	×
93	geo_coordinates = []	×
94
95	for tle_lines, tle_time in zip(self.tle_data, self.tle_time_list, strict=True):	×
96	satellite = EarthSatellite(tle_lines[0], tle_lines[1], self.satellite_name)	×
97	geocentric = satellite.at(timescale.from_datetime(tle_time))	×
98	xyz = geocentric.xyz.km	×
99	geo_coordinates.append(xyz)	×
100
101	geo_coordinates = np.asarray(geo_coordinates, dtype=np.float64)	×
102
103	if np.isnan(geo_coordinates).any():	×
104	nan_indices = np.where(np.isnan(geo_coordinates).any(axis=1))[0]	×
105	logger.warning(	×
106	f"NaN values found in GEO coordinates at indices: {', '.join(str(idx) for idx in nan_indices)}. "
107	"Check the TLE file at these indices."
108	)
109
110	return np.asarray(geo_coordinates, dtype=np.float64)	×

GFZ / EL_PASO / 22961397810

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous