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[satellite_position_fov] | 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>

load_ephemeris(ephemeris: InterpolableEphemeris, ephem_repo: AbstractEphemerisRepository) → None

Load and parse ephemeris data from a file.

Parameters:

• sat_number – Satellite NORAD number

• epoch – Epoch time for the ephemeris

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_fov]

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

class src.api.utils.propagation_strategies.satellite_position_fov

Bases: tuple

satellite_position_fov(ra, dec, covariance, angle, altitude, azimuth, range_km, julian_date, name, norad_id, orbital_data_epoch, orbital_data_source)

static __new__(_cls, ra, dec, covariance, angle, altitude, azimuth, range_km, julian_date, name, norad_id, orbital_data_epoch, orbital_data_source)

Create new instance of satellite_position_fov(ra, dec, covariance, angle, altitude, azimuth, range_km, julian_date, name, norad_id, orbital_data_epoch, orbital_data_source)

_asdict()

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

_field_defaults = {}

_fields = ('ra', 'dec', 'covariance', 'angle', 'altitude', 'azimuth', 'range_km', 'julian_date', 'name', 'norad_id', 'orbital_data_epoch', 'orbital_data_source')

classmethod _make(iterable)

Make a new satellite_position_fov object from a sequence or iterable

_replace(**kwds)

Return a new satellite_position_fov object replacing specified fields with new values

altitude

Alias for field number 4

angle

Alias for field number 3

azimuth

Alias for field number 5

covariance

Alias for field number 2

dec

Alias for field number 1

julian_date

Alias for field number 7

name

Alias for field number 8

norad_id

Alias for field number 9

orbital_data_epoch

Alias for field number 10

orbital_data_source

Alias for field number 11

ra

Alias for field number 0

range_km

Alias for field number 6