src.api.utils.propagation_strategies module

class src.api.utils.propagation_strategies.BasePropagationStrategy

Bases: ABC

Base class for all propagation strategies.

_abc_impl = <_abc._abc_data object>

abstract propagate(julian_dates: float | list[float] | ndarray, tle_line_1: str, tle_line_2: str, latitude: float, longitude: float, elevation: float, **kwargs) → satellite_position | list[satellite_position] | list[dict[str, Any]]

Propagate satellite positions.

Parameters:

• julian_dates – Single Julian date or array of Julian dates

• tle_line_1 – First line of TLE

• tle_line_2 – Second line of TLE

• latitude – Observer latitude in degrees

• longitude – Observer longitude in degrees

• elevation – Observer elevation in meters

• **kwargs – Additional strategy-specific parameters

Returns:

Propagated position(s) in the strategy’s format

class src.api.utils.propagation_strategies.FOVParallelPropagationStrategy

Bases: object

Propagate satellite positions and check if they fall within FOV.

Supports two execution modes: - In-process: uses ThreadPoolExecutor (sync path; GIL-limited) - Distributed: batch_executor runs batches (e.g. Celery chord in fov_service);

batch_serializer converts TLE batches to serializable form for the executor

propagate(all_tles, jd_times, location, fov_center, fov_radius, batch_size=1000, max_workers=None, include_tles=True, illuminated_only=False, progress_callback=None, *, batch_executor: Callable[[...], list[tuple[list, int, float]]] | None = None, batch_serializer: Callable[[list], list] | None = None) → tuple[list[dict[str, Any]], float, int]

Propagate satellite positions and check if they fall within FOV.

Parameters:

• all_tles – List of TLE objects

• jd_times – Array of Julian dates

• location – Observer’s location

• fov_center – Tuple of (RA, Dec) in degrees. Defaults to (0,0)

• fov_radius – FOV radius in degrees. Defaults to 0

• batch_size – Number of satellites to process in each batch

• max_workers – Maximum number of workers (in-process mode only)

• include_tles – Whether to include TLE data in results

• illuminated_only – Whether to only include illuminated satellites

• progress_callback – Optional callback for progress updates

• batch_executor – Callable(serialized_batches, common_args) -> list of (results, sats). When provided, runs distributed (e.g. Celery).

• batch_serializer – Callable(batch) -> serialized_batch. Required when batch_executor is provided; converts TLE batches for message passing.

Returns:

(

results: List of dictionaries containing position data for points in FOV, execution_time: Total execution time in seconds, satellites_processed: Number of satellites processed

)

Return type:

tuple

class src.api.utils.propagation_strategies.FOVPropagationStrategy

Bases: BasePropagationStrategy

_abc_impl = <_abc._abc_data object>

propagate(julian_dates: float | list[float] | ndarray, tle_line_1: str, tle_line_2: str, latitude: float, longitude: float, elevation: float, fov_center: tuple[float, float] = (0.0, 0.0), fov_radius: float = 0.0, **kwargs) → list[dict[str, Any]]

Propagate satellite positions and check if they fall within FOV.

Parameters:

• julian_dates – Single Julian date or array of Julian dates

• tle_line_1 – First line of TLE

• tle_line_2 – Second line of TLE

• latitude – Observer latitude in degrees

• longitude – Observer longitude in degrees

• elevation – Observer elevation in meters

• fov_center – Tuple of (RA, Dec) in degrees. Defaults to (0,0)

• fov_radius – FOV radius in degrees. Defaults to 0

• **kwargs – Additional parameters (not used in this strategy)

Returns:

List of dictionaries containing position data for points in FOV

Raises:

• RuntimeError – If propagation fails due to invalid TLE

• or numerical instability –

class src.api.utils.propagation_strategies.KroghPropagationStrategy

Bases: BasePropagationStrategy

__init__()

Initialize the Krogh propagation strategy.

_abc_impl = <_abc._abc_data object>

_create_chunked_krogh_interpolator(x: ndarray, y: ndarray, chunk_size: int = 14, overlap: int = 8) → list

Create a series of overlapping Krogh interpolators to handle large datasets with improved stability.

This method splits the input data into overlapping chunks and creates a Krogh interpolator for each chunk. This approach helps avoid numerical instability that can occur when interpolating over many points using a single interpolator.

The method uses a sliding window approach with overlap to ensure smooth transitions between chunks. For each chunk: - First chunk: Valid from start to just before end - Middle chunks: Valid in middle portion, leaving overlap areas for adjacent chunks - Last chunk: Valid from just after start to end

Parameters:

• x (np.ndarray) – Independent variable values (e.g., times)

• y (np.ndarray) – Dependent variable values to interpolate

• chunk_size (int, optional) – Number of points to use in each interpolation

• 14. (chunk. Defaults to) –

• overlap (int, optional) – Number of points to overlap between chunks. Defaults to 8.

Returns:

List of dictionaries, each containing:

• interpolator (KroghInterpolator): The interpolator for this chunk

• range (tuple): Valid range for this interpolator (lower, upper)

Return type:

list

Note

• If input data length is less than chunk_size, returns a single

interpolator - Overlap should be less than chunk_size to ensure progress - The valid ranges are slightly narrower than the actual chunks to ensure

smooth transitions between interpolators

_generate_and_propagate_sigma_points(data: dict) → dict

Generate and propagate sigma points using the Unscented Transform for improved numerical stability.

This method implements the Unscented Transform to generate sigma points from the state vectors and covariance matrices in the ephemeris data. The sigma points are used to capture the mean and covariance of the state distribution more accurately than linearization methods.

The method uses optimized parameters for the Unscented Transform: - alpha = 0.001 (reduced for better numerical stability) - beta = 2.0 (optimal for Gaussian distributions) - kappa = 3-n (modified for better stability)

Parameters:

data (dict) – Dictionary containing ephemeris data with keys: - timestamps (np.ndarray): Array of datetime objects - positions (np.ndarray): Array of position vectors - velocities (np.ndarray): Array of velocity vectors - covariances (np.ndarray): Array of 6x6 covariance matrices

Returns:

A dictionary mapping Julian dates to sigma point information:

• sigma_points (np.ndarray): Array of 13 sigma points

(6D state vectors) - weights (dict): Dictionary containing mean and covariance weights - epoch (datetime): Timestamp for these sigma points - state_vector (np.ndarray): Original state vector - covariance (np.ndarray): Original covariance matrix

Return type:

dict

Raises:

• ValueError – If no sigma points could be generated successfully

• np.linalg.LinAlgError – If covariance matrix is not positive definite

Note

The method uses Cholesky decomposition for numerical stability, falling back to matrix square root if Cholesky fails. Each sigma point set contains 13 points: - 1 mean state point - 6 points from positive Cholesky decomposition - 6 points from negative Cholesky decomposition

_get_interpolated_sigma_points_KI(interpolated_splines: dict, julian_date: float) → ndarray

Get interpolated sigma points at a specific Julian date using optimal chunk selection.

This method interpolates the position and velocity components of all 13 sigma points at the requested Julian date. It uses a sophisticated chunk selection algorithm that prefers interpolators where the requested time is in the middle of their valid range, rather than at the edges, to minimize interpolation errors.

The method handles both position and velocity components (x, y, z) for each sigma point, selecting the most appropriate interpolator chunk for each component based on the requested time’s position within the chunk’s valid range.

Parameters:

• interpolated_splines (dict) – Dictionary containing interpolation splines: - positions (list): List of lists of position interpolators - velocities (list): List of lists of velocity interpolators - time_range (tuple): (start_time, end_time) in Julian dates

• julian_date (float) – The Julian date at which to interpolate the sigma

• points –

Returns:

Array of shape (13, 6) containing the interpolated sigma points:

• First 13 rows: One row per sigma point

• 6 columns: [x, y, z, vx, vy, vz] for each point

• dtype: np.float64 for high precision

Return type:

np.ndarray

Raises:

ValueError – If the requested Julian date is outside the interpolation range

Note

• The method uses a centrality score to select the best interpolator chunk

• For times outside any chunk’s range, the nearest chunk is used

• All calculations are performed in double precision (np.float64)

• The method assumes the input splines are valid and properly structured

_interpolate_sigma_pointsKI(sigma_points_dict: dict) → dict

Create high-precision interpolation splines for sigma point trajectories using chunked Krogh interpolation.

This method processes the sigma points dictionary to create interpolators for each component of the position and velocity vectors. It handles 13 sigma points (1 mean + 6 positive + 6 negative Cholesky points) and creates separate interpolators for each component (x, y, z) of both position and velocity.

The method uses chunked Krogh interpolation to maintain numerical stability when dealing with long time series. Each component is interpolated independently, and invalid or non-finite values are handled gracefully.

Parameters:

sigma_points_dict (dict) – Dictionary mapping Julian dates to sigma point information: - sigma_points (np.ndarray): Array of 13 sigma points (6D state vectors) - weights (dict): Dictionary containing mean and covariance weights - epoch (datetime): Timestamp for these sigma points - state_vector (np.ndarray): Original state vector - covariance (np.ndarray): Original covariance matrix

Returns:

Dictionary containing interpolation splines and time range:

• positions (list): List of lists of position interpolators:

  • Outer list: One entry per sigma point (13 total)

  • Inner list: One entry per component (x, y, z)

  • Each entry: List of chunked Krogh interpolators

• velocities (list): List of lists of velocity interpolators:

  • Outer list: One entry per sigma point (13 total)

  • Inner list: One entry per component (x, y, z)

  • Each entry: List of chunked Krogh interpolators

• time_range (tuple): (start_time, end_time) in Julian dates

Return type:

dict

Note

• Each component (x, y, z) of position and velocity has its own set of interpolators

• Interpolators are created only for valid (finite) data points

• The chunking parameters (chunk_size=14, overlap=8) are optimized

for stability - None is returned for components with no valid data points

_parse_ephemeris_file(filename: str) → dict

Parse a satellite ephemeris file in UVW frame format.

The file format is expected to be: - First 3 lines: Header information in key:value format - Line 4: Must contain ‘UVW’ to specify the coordinate frame - Remaining lines: State vectors and covariance matrices in blocks of 4 lines:

• Line 1: Timestamp (YYYYDDDHHMMSSmmm) and state vector

(position and velocity) * Lines 2-4: Lower triangular portion of 6x6 covariance matrix

Parameters:

filename (str) – Path to the ephemeris file

Returns:

A dictionary containing:

• headers (dict): Key-value pairs from the file header

• timestamps (np.ndarray): Array of datetime objects

• positions (np.ndarray): Array of position vectors in UVW frame

• velocities (np.ndarray): Array of velocity vectors in UVW frame

• covariances (np.ndarray): Array of 6x6 covariance matrices

Return type:

dict

Raises:

• ValueError – If the file doesn’t specify UVW frame or has invalid format

• FileNotFoundError – If the file doesn’t exist

• IndexError – If the file has an invalid format or is truncated

_reconstruct_covariance_at_time(interpolated_points: ndarray) → tuple

Reconstruct the mean state and covariance matrix from interpolated sigma points using the Unscented Transform.

This method implements the Unscented Transform to reconstruct the mean state and covariance matrix from a set of interpolated sigma points. It uses optimized parameters for numerical stability and accuracy in the presence of non-linear transformations.

The method uses the following Unscented Transform parameters: - alpha = 0.001: Reduced for better numerical stability - beta = 2.0: Optimal for Gaussian distributions - kappa = 3-n: Modified for better stability - lambda = alpha²(n+kappa) - n: Scaling parameter

Parameters:

• interpolated_points (np.ndarray) – Array of shape (13, 6) containing

• points (the interpolated sigma) –

  • 13 rows: One row per sigma point (1 mean + 6 positive + 6

  negative Cholesky points) - 6 columns: [x, y, z, vx, vy, vz] state components - dtype: np.float64 for high precision

• where –

  • 13 rows: One row per sigma point (1 mean + 6 positive + 6

  negative Cholesky points) - 6 columns: [x, y, z, vx, vy, vz] state components - dtype: np.float64 for high precision

Returns:

(mean_state, covariance) where:

• mean_state (np.ndarray): Array of shape (6,) containing the

mean state vector - covariance (np.ndarray): Array of shape (6, 6) containing the symmetric covariance matrix

Return type:

tuple

Note

• The mean state is taken directly from the first sigma point for stability

• The covariance matrix is computed using weighted outer products

• The final covariance matrix is symmetrized to ensure numerical stability

• All calculations are performed in double precision (np.float64)

load_ephemeris(ephemeris_file: str) → None

Load and parse ephemeris data from a file.

Parameters:

ephemeris_file – Path to the ephemeris file

propagate(julian_dates: float | list[float] | ndarray, tle_line_1: str, tle_line_2: str, latitude: float, longitude: float, elevation: float, **kwargs) → list[satellite_position]

Propagate satellite positions using Krogh interpolation.

Parameters:

• julian_dates – Single Julian date or array of Julian dates

• tle_line_1 – First line of TLE (not used in this strategy)

• tle_line_2 – Second line of TLE (not used in this strategy)

• latitude – Observer latitude in degrees

• longitude – Observer longitude in degrees

• elevation – Observer elevation in meters

• **kwargs – Additional parameters (not used in this strategy)

Returns:

List of satellite positions

class src.api.utils.propagation_strategies.PropagationInfo

Bases: object

__init__(propagation_strategy, tle_line_1, tle_line_2, julian_date, latitude, longitude, elevation)

propagate()

class src.api.utils.propagation_strategies.SGP4PropagationStrategy

Bases: BasePropagationStrategy

_abc_impl = <_abc._abc_data object>

propagate(julian_dates: float | list[float] | ndarray, tle_line_1: str, tle_line_2: str, latitude: float, longitude: float, elevation: float, **kwargs) → satellite_position | list[satellite_position]

Propagates satellite and observer states using the SGP4 propagation model.

Parameters:

• julian_dates – Single Julian date or array of Julian dates

• tle_line_1 – First line of TLE

• tle_line_2 – Second line of TLE

• latitude – Observer latitude in degrees

• longitude – Observer longitude in degrees

• elevation – Observer elevation in meters above the WGS84 ellipsoid.

• **kwargs – Additional parameters (not used in this strategy)

Returns:

Single satellite position or list of positions

class src.api.utils.propagation_strategies.SkyfieldPropagationStrategy

Bases: BasePropagationStrategy

_abc_impl = <_abc._abc_data object>

propagate(julian_dates: float | list[float] | ndarray, tle_line_1: str, tle_line_2: str, latitude: float, longitude: float, elevation: float, **kwargs) → list[satellite_position]

Use Skyfield (https://rhodesmill.org/skyfield/earth-satellites.html) to propagate satellite and observer states.

Parameters:

• julian_dates – Single Julian date or array of Julian dates

• tle_line_1 – TLE line 1

• tle_line_2 – TLE line 2

• latitude – The observer WGS84 latitude in degrees

• longitude – The observers WGS84 longitude in degrees (positive value represents east, negative value represents west)

• elevation – The observer elevation above WGS84 ellipsoid in meters

• **kwargs – Additional parameters (not used in this strategy)

Returns:

List of propagated positions

Raises:

• RuntimeError – If propagation fails due to invalid TLE

• or numerical instability –

class src.api.utils.propagation_strategies.TestPropagationStrategy

Bases: BasePropagationStrategy

_abc_impl = <_abc._abc_data object>

propagate(julian_dates: float | list[float] | ndarray, tle_line_1: str, tle_line_2: str, latitude: float, longitude: float, elevation: float, **kwargs) → satellite_position | list[satellite_position]

Test propagation strategy that uses Skyfield implementation.

Parameters:

• julian_dates – Single Julian date or array of Julian dates

• tle_line_1 – First line of TLE

• tle_line_2 – Second line of TLE

• latitude – Observer latitude in degrees

• longitude – Observer longitude in degrees

• elevation – Observer elevation in meters

• **kwargs – Additional parameters (not used in this strategy)

Returns:

Single satellite position or list of positions

src.api.utils.propagation_strategies._run_batches_threadpool(args_list: list[tuple], max_workers: int, progress_callback: Callable[[...], None] | None) → tuple[list[dict[str, Any]], int]

Run batches using ThreadPoolExecutor (sync path).

src.api.utils.propagation_strategies.get_timescale()

src.api.utils.propagation_strategies.process_satellite_batch(args: tuple[Any, ...]) → tuple[list[dict[str, Any]], int, float]

Process a batch of satellites for FOV calculations.

Returns:

(batch_results, satellites_processed, execution_time)

Return type:

tuple

class src.api.utils.propagation_strategies.satellite_position

Bases: tuple

satellite_position(ra, dec, dracosdec, ddec, alt, az, distance, ddistance, phase_angle, sat_altitude_km, solar_elevation_deg, solar_azimuth_deg, illuminated, satellite_gcrs, observer_gcrs, julian_date)

static __new__(_cls, ra, dec, dracosdec, ddec, alt, az, distance, ddistance, phase_angle, sat_altitude_km, solar_elevation_deg, solar_azimuth_deg, illuminated, satellite_gcrs, observer_gcrs, julian_date)

Create new instance of satellite_position(ra, dec, dracosdec, ddec, alt, az, distance, ddistance, phase_angle, sat_altitude_km, solar_elevation_deg, solar_azimuth_deg, illuminated, satellite_gcrs, observer_gcrs, julian_date)

_asdict()

Return a new dict which maps field names to their values.

_field_defaults = {}

_fields = ('ra', 'dec', 'dracosdec', 'ddec', 'alt', 'az', 'distance', 'ddistance', 'phase_angle', 'sat_altitude_km', 'solar_elevation_deg', 'solar_azimuth_deg', 'illuminated', 'satellite_gcrs', 'observer_gcrs', 'julian_date')

classmethod _make(iterable)

Make a new satellite_position object from a sequence or iterable

_replace(**kwds)

Return a new satellite_position object replacing specified fields with new values

alt

Alias for field number 4

az

Alias for field number 5

ddec

Alias for field number 3

ddistance

Alias for field number 7

dec

Alias for field number 1

distance

Alias for field number 6

dracosdec

Alias for field number 2

illuminated

Alias for field number 12

julian_date

Alias for field number 15

observer_gcrs

Alias for field number 14

phase_angle

Alias for field number 8

ra

Alias for field number 0

sat_altitude_km

Alias for field number 9

satellite_gcrs

Alias for field number 13

solar_azimuth_deg

Alias for field number 11

solar_elevation_deg

Alias for field number 10