# ruff: noqa: E501
from flask import current_app as app
from flask import jsonify, request
from api.adapters.repositories.ephemeris_repository import SqlAlchemyEphemerisRepository
from api.adapters.repositories.tdm_repository import SqlAlchemyTdmPredictionRepository
from api.adapters.repositories.tle_repository import SqlAlchemyTLERepository
from api.entrypoints.extensions import db, limiter
from api.services.fov_service import (
get_satellite_passes_in_fov,
# get_satellites_above_horizon_range,
get_satellite_passes_in_fov_async,
get_satellite_passes_in_fov_tdm,
get_satellites_above_horizon,
)
from api.services.tasks.fov_tasks import get_fov_task_status
from api.services.validation_service import validate_parameters
from . import api_main, api_source, api_v1, api_version
[docs]
@api_v1.route("/fov/satellite-passes/")
@api_main.route("/fov/satellite-passes/")
@limiter.limit("50 per second, 1000 per minute")
def get_satellite_passes():
"""Get satellites that pass through a specified field of view.
---
tags:
- Field of View
summary: Get satellite passes in a field of view
description: Get satellites that pass through a specified field of view during an observation period. Supports both synchronous and asynchronous processing modes.
parameters:
- name: latitude
in: query
type: number
format: float
required: false
description: Latitude of observation site in decimal degrees (required if site not provided)
example: 48.8566
- name: longitude
in: query
type: number
format: float
required: false
description: Longitude of observation site in decimal degrees (required if site not provided)
example: 2.3522
- name: elevation
in: query
type: number
format: float
required: false
description: Elevation of observation site in meters (required if site not provided)
example: 35.0
- name: site
in: query
type: string
required: false
description: Predefined site name/alias from AstroPy list (https://www.astropy.org/astropy-data/coordinates/sites.json), can be used instead of latitude/longitude/elevation
example: "rubin"
- name: duration
in: query
type: number
format: float
required: true
description: Duration of observation in seconds
example: 60.0
- name: ra
in: query
type: number
format: float
required: true
description: Right ascension of field center in decimal degrees
example: 15.0
- name: dec
in: query
type: number
format: float
required: true
description: Declination of field center in decimal degrees
example: 30.0
- name: fov_radius
in: query
type: number
format: float
required: true
description: Radius of field of view in decimal degrees
example: 0.5
- name: start_time_jd
in: query
type: number
format: float
required: false
description: Start time of observation in Julian date (either this or mid_obs_time_jd must be provided)
example: 2459000.5
- name: mid_obs_time_jd
in: query
type: number
format: float
required: false
description: Mid-observation time in Julian date (either this or start_time_jd must be provided)
example: 2459000.5
- name: group_by
in: query
type: string
required: false
description: Group results by 'satellite' or 'time' (default is 'time' for chronological order)
enum: [satellite, time]
example: satellite
- name: include_tles
in: query
type: boolean
required: false
description: Whether to include TLE data used to calculate the passes in the response
example: true
- name: skip_cache
in: query
type: boolean
required: false
description: Whether to skip the cache and calculate the passes from scratch
example: false
- name: constellation
in: query
type: string
required: false
description: Constellation of the satellites to include in the response
example: "starlink"
- name: data_source
in: query
type: string
required: false
description: Data source to use for orbital data ("celestrak","spacetrack"). Default is any/all sources.
example: "celestrak"
- name: illuminated_only
in: query
type: boolean
required: false
description: Whether to include only illuminated satellites (default is false)
example: true
- name: async
in: query
type: boolean
required: false
description: Whether to process the request asynchronously. If true or omitted, returns a task ID for polling status. If false, returns immediate results.
example: false
- name: tle_only
in: query
type: boolean
required: false
description: Whether to include only TLE data in the response. If true, ephemeris data will not be used.
example: true
responses:
200:
description: Successful response. Returns either immediate satellite pass data (synchronous) or task information (asynchronous) based on the async parameter.
content:
application/json:
schema:
oneOf:
- type: object
description: Synchronous response (when async=false)
properties:
data:
type: object
properties:
satellites:
type: object
description: "Map of satellite pass data. Keys are 'Name (norad_id)', e.g. 'CZ-6A DEB (55190)'."
additionalProperties:
type: object
properties:
name:
type: string
description: Name of the satellite
norad_id:
type: integer
description: NORAD catalog ID of the satellite
positions:
type: array
items:
type: object
properties:
altitude:
type: number
format: float
description: Altitude above horizon in degrees
angle:
type: number
format: float
description: Angular distance from FOV center in degrees
azimuth:
type: number
format: float
description: Azimuth angle in degrees
date_time:
type: string
description: UTC time in YYYY-MM-DD HH:MM:SS TZ format
dec:
type: number
format: float
description: Declination in degrees
julian_date:
type: number
format: float
description: Julian date for this position
ra:
type: number
format: float
description: Right ascension in degrees
orbital_data_epoch:
type: string
nullable: true
description: UTC instant for which the underlying orbital data applies (TLE epoch is in tle_epoch; for ephemeris-based positions this is the ephemeris generation time)
orbital_data_source:
type: string
nullable: true
description: Orbital data source used for the position (e.g. ephemeris)
range_km:
type: number
format: float
description: Distance to the satellite in kilometers
tle_data:
type: object
description: TLE data for this satellite (only included when include_tles=true)
properties:
tle_line1:
type: string
description: First line of the TLE
tle_line2:
type: string
description: Second line of the TLE
source:
type: string
description: Source of the TLE data (celestrak or spacetrack)
total_position_results:
type: integer
description: Total number of position results
total_satellites:
type: integer
description: Total number of satellites found
source:
type: string
description: The source of the satellite position data
example: "IAU CPS SatChecker"
version:
type: string
description: The version of the API
example: "1.X.x"
- type: object
description: Asynchronous response (when async=true or omitted)
properties:
task_id:
type: string
description: Unique identifier for the asynchronous task
example: "abc123-def456-ghi789"
status:
type: string
description: Current status of the task
enum: ["PENDING", "PROGRESS", "SUCCESS", "FAILURE"]
example: "PROGRESS"
message:
type: string
description: Human-readable status message
example: "FOV calculation started. Use the task_id to check status and retrieve results."
progress:
type: number
format: float
description: Progress percentage (only present when status is PROGRESS)
example: 50.0
source:
type: string
description: The source of the satellite position data
example: "IAU CPS SatChecker"
version:
type: string
description: The version of the API
example: "1.X.x"
400:
description: Bad request due to incorrect parameters
500:
description: Internal server error
"""
parameters = [
"latitude",
"longitude",
"elevation",
"site",
"duration",
"ra",
"dec",
"fov_radius",
"start_time_jd",
"mid_obs_time_jd",
"group_by",
"include_tles",
"skip_cache",
"constellation",
"data_source",
"illuminated_only",
"async",
"tle_only",
"use_generated_tles",
]
if "site" not in request.args:
required_parameters = [
"latitude",
"longitude",
"elevation",
"duration",
"ra",
"dec",
"fov_radius",
]
else:
required_parameters = ["site", "duration", "ra", "dec", "fov_radius"]
validated_parameters = validate_parameters(request, parameters, required_parameters)
if validated_parameters["async"] is None:
validated_parameters["async"] = True
if validated_parameters["tle_only"] is None:
validated_parameters["tle_only"] = True
if validated_parameters["use_generated_tles"] is None:
validated_parameters["use_generated_tles"] = False
session = db.session
tle_repo = SqlAlchemyTLERepository(session)
ephem_repo = SqlAlchemyEphemerisRepository(session)
tdm_repo = SqlAlchemyTdmPredictionRepository(session)
try:
if validated_parameters["data_source"] == "aerospace":
# has to use site, not lat/lon/elev, because the TDM predictions are not available for all sites
# return with error if site is not included
if validated_parameters["site"] is None:
return {
"info": "Site is required for TDM predictions",
"api_source": api_source,
"version": api_version,
}
task_response = get_satellite_passes_in_fov_tdm(
tdm_repo,
validated_parameters["site"],
validated_parameters["location"],
validated_parameters["mid_obs_time_jd"],
validated_parameters["start_time_jd"],
validated_parameters["duration"],
validated_parameters["ra"],
validated_parameters["dec"],
validated_parameters["fov_radius"],
validated_parameters["group_by"],
validated_parameters["constellation"],
api_source,
api_version,
)
return jsonify(task_response)
if validated_parameters["async"]:
task_response = get_satellite_passes_in_fov_async(
tle_repo,
validated_parameters["location"],
validated_parameters["mid_obs_time_jd"],
validated_parameters["start_time_jd"],
validated_parameters["duration"],
validated_parameters["ra"],
validated_parameters["dec"],
validated_parameters["fov_radius"],
validated_parameters["group_by"],
validated_parameters["include_tles"],
validated_parameters["skip_cache"],
validated_parameters["constellation"],
validated_parameters["data_source"],
validated_parameters["illuminated_only"],
validated_parameters["tle_only"],
validated_parameters["use_generated_tles"],
api_source,
api_version,
)
return jsonify(task_response)
else:
satellite_passes = get_satellite_passes_in_fov(
tle_repo,
ephem_repo,
validated_parameters["location"],
validated_parameters["mid_obs_time_jd"],
validated_parameters["start_time_jd"],
validated_parameters["duration"],
validated_parameters["ra"],
validated_parameters["dec"],
validated_parameters["fov_radius"],
validated_parameters["group_by"],
validated_parameters["include_tles"],
validated_parameters["skip_cache"],
validated_parameters["constellation"],
validated_parameters["data_source"],
validated_parameters["illuminated_only"],
validated_parameters["tle_only"],
validated_parameters["use_generated_tles"],
api_source,
api_version,
)
if not satellite_passes:
return {
"info": "No position information found with this criteria",
"api_source": api_source,
"version": api_version,
}
return jsonify(satellite_passes)
except ValueError as e:
app.logger.error(e)
return jsonify({"error": "Incorrect parameters"}), 400
except Exception as e:
app.logger.error(e)
return jsonify({"error": str(e)}), 500
[docs]
@api_v1.route("/fov/task-status/<task_id>")
@api_main.route("/fov/task-status/<task_id>")
@limiter.limit("100 per second, 2000 per minute")
def get_fov_task_status_endpoint(task_id):
"""Get the status of a FOV calculation task.
---
tags:
- Field of View
summary: Get FOV task status
description: Check the status of an asynchronous FOV calculation task and retrieve results when complete
parameters:
- name: task_id
in: path
type: string
required: true
description: The task ID returned from the async FOV endpoint
example: "abc123-def456-ghi789"
responses:
200:
description: Task status and result if available
content:
application/json:
schema:
type: object
properties:
status:
type: string
enum: [PENDING, PROGRESS, SUCCESS, FAILURE]
description: Current status of the task
message:
type: string
description: Human-readable status message
result:
type: object
description: FOV calculation results (only present when status is SUCCESS)
performance_metrics:
type: object
description: Performance metrics (only present when status is SUCCESS)
error:
type: string
description: Error message (only present when status is FAILURE)
progress:
type: number
format: float
description: Progress percentage (only present when status is PROGRESS)
404:
description: Task not found
500:
description: Internal server error
"""
try:
task_status = get_fov_task_status(task_id)
return jsonify(task_status)
except Exception as e:
app.logger.error(f"Error checking task status for {task_id}: {str(e)}")
return jsonify({"error": "Failed to check task status"}), 500
[docs]
@api_v1.route("/fov/satellites-above-horizon/")
@api_main.route("/fov/satellites-above-horizon/")
@limiter.limit("50 per second, 1000 per minute")
def get_all_satellites_above_horizon():
"""Get satellites above horizon at a specific time.
---
tags:
- Field of View
summary: Get satellites above horizon at a specific time
description: Get a list of satellites that are above the horizon at a specific Julian date for a given location
parameters:
- name: latitude
in: query
type: number
format: float
required: false
description: Latitude of observation site in decimal degrees (required if site not provided)
example: 48.8566
- name: longitude
in: query
type: number
format: float
required: false
description: Longitude of observation site in decimal degrees (required if site not provided)
example: 2.3522
- name: elevation
in: query
type: number
format: float
required: false
description: Elevation of observation site in meters (required if site not provided)
example: 35.0
- name: site
in: query
type: string
required: false
description: Predefined site name/alias from AstroPy list (https://www.astropy.org/astropy-data/coordinates/sites.json), can be used instead of latitude/longitude/elevation
example: "rubin"
- name: julian_date
in: query
type: number
format: float
required: true
description: Time at which to check for satellites above horizon in Julian date format
example: 2459000.5
- name: min_altitude
in: query
type: number
format: float
required: false
description: Minimum altitude above horizon in degrees (default is 0)
example: 15.0
- name: illuminated_only
in: query
type: boolean
required: false
description: Whether to include only illuminated satellites (default is false)
example: true
- name: min_range
in: query
type: number
format: float
required: false
description: Minimum range of satellites in kilometers (default is 0.0)
example: 300.0
- name: max_range
in: query
type: number
format: float
required: false
description: Maximum range of satellites in kilometers (default is infinity)
example: 500.0
- name: constellation
in: query
type: string
required: false
description: Constellation of the satellites to include in the response
example: "starlink"
responses:
200:
description: Successful response with satellites above horizon
content:
application/json:
schema:
type: object
properties:
count:
type: integer
description: The number of satellites found above the horizon
data:
type: array
description: List of satellites above the horizon
items:
type: object
properties:
name:
type: string
description: Name of the satellite
norad_id:
type: integer
description: NORAD catalog ID of the satellite
julian_date:
type: number
format: float
description: Julian date for the position
altitude:
type: number
format: float
description: Altitude above horizon in degrees
azimuth:
type: number
format: float
description: Azimuth angle in degrees
ra:
type: number
format: float
description: Right ascension in degrees
dec:
type: number
format: float
description: Declination in degrees
range:
type: number
format: float
description: Distance to the satellite in kilometers
tle_epoch:
type: string
description: Epoch date of the TLE used for calculation in YYYY-MM-DD HH:MM:SS TZ format
source:
type: string
description: The source of the satellite position data
example: "IAU CPS SatChecker"
version:
type: string
description: The version of the API
example: "1.X.x"
400:
description: Bad request due to incorrect parameters
500:
description: Internal server error
"""
return _handle_satellites_above_horizon(with_duration=False)
[docs]
@api_v1.route("/fov/satellites-above-horizon-range/")
@api_main.route("/fov/satellites-above-horizon-range/")
@limiter.limit("50 per second, 1000 per minute")
def get_all_satellites_above_horizon_range():
"""Get satellites above horizon over a time range.
---
tags:
- Field of View
summary: Get satellites above horizon over a time range
description: Get a list of satellites that are above the horizon during a specified time range for a given location
parameters:
- name: latitude
in: query
type: number
format: float
required: true
description: Latitude of observation site in decimal degrees (required if site not provided)
example: 48.8566
- name: longitude
in: query
type: number
format: float
required: true
description: Longitude of observation site in decimal degrees (required if site not provided)
example: 2.3522
- name: elevation
in: query
type: number
format: float
required: true
description: Elevation of observation site in meters (required if site not provided)
example: 35.0
- name: site
in: query
type: string
required: false
description: Predefined site name/alias from AstroPy list (https://www.astropy.org/astropy-data/coordinates/sites.json), can be used instead of latitude/longitude/elevation
example: "rubin"
- name: julian_date
in: query
type: number
format: float
required: true
description: Start time of the observation period in Julian date
example: 2459000.5
- name: duration
in: query
type: number
format: float
required: true
description: Duration of observation period in seconds
example: 120.0
- name: min_altitude
in: query
type: number
format: float
required: false
description: Minimum altitude above horizon in degrees (default is 0)
example: 15.0
- name: illuminated_only
in: query
type: boolean
required: false
description: Whether to include only illuminated satellites (default is false)
example: true
- name: min_range
in: query
type: number
format: float
required: false
description: Minimum range of satellites in kilometers (default is 0.0)
example: 300.0
- name: max_range
in: query
type: number
format: float
required: false
description: Maximum range of satellites in kilometers (default is infinity)
example: 500.0
responses:
200:
description: Successful response with satellites above horizon during the specified period
content:
application/json:
schema:
type: object
properties:
count:
type: integer
description: The number of satellites found above the horizon
data:
type: array
description: List of satellites above the horizon
items:
type: object
properties:
tbd
source:
type: string
description: The source of the satellite position data
example: "IAU CPS SatChecker"
version:
type: string
description: The version of the API
example: "1.X.x"
400:
description: Bad request due to incorrect parameters
500:
description: Internal server error
"""
return _handle_satellites_above_horizon(with_duration=True)
[docs]
def _handle_satellites_above_horizon(with_duration=False):
"""Helper function to reduce code duplication between the similar endpoints."""
# Base parameters for both endpoints
parameters = [
"latitude",
"longitude",
"elevation",
"site",
"julian_date",
"min_altitude",
"illuminated_only",
"min_range",
"max_range",
"constellation",
]
# Add duration parameter if needed
if with_duration:
parameters.append("duration")
# Define required parameters based on presence of site and duration
if "site" not in request.args:
required_parameters = ["latitude", "longitude", "elevation", "julian_date"]
else:
required_parameters = ["site", "julian_date"]
# Add duration to required parameters if needed
if with_duration:
required_parameters.append("duration")
validated_parameters = validate_parameters(request, parameters, required_parameters)
session = db.session
tle_repo = SqlAlchemyTLERepository(session)
try:
# Choose the appropriate service function based on whether duration is included
if with_duration:
"""
satellite_passes = get_satellites_above_horizon_range(
tle_repo,
validated_parameters["location"],
validated_parameters["julian_dates"],
validated_parameters["min_altitude"],
validated_parameters["min_range"],
validated_parameters["max_range"],
validated_parameters["illuminated_only"],
validated_parameters["duration"],
api_source,
api_version,
)
"""
pass
else:
satellite_passes = get_satellites_above_horizon(
tle_repo,
validated_parameters["location"],
validated_parameters["julian_dates"],
validated_parameters["min_altitude"],
validated_parameters["min_range"],
validated_parameters["max_range"],
validated_parameters["illuminated_only"],
validated_parameters["constellation"],
api_source,
api_version,
)
if not satellite_passes:
return {
"info": "No position information found with this criteria",
"source": api_source,
"version": api_version,
}
return jsonify(satellite_passes)
except ValueError as e:
app.logger.error(e)
return jsonify({"error": "Incorrect parameters"}), 400
except Exception as e:
app.logger.error(e)
return jsonify({"error": str(e)}), 500