astra.main

Main FastAPI application for Astra observatory automation system.

This module provides a web-based interface for controlling and monitoring multiple astronomical observatories. It handles real-time device status updates, schedule management, image display, logging, and WebSocket communications for the observatory control system.

Key features: - Multi-observatory support with device monitoring - Real-time WebSocket updates for device status - Schedule upload and editing capabilities - Image conversion and display (FITS to JPEG) - Database logging and telemetry storage - Safety monitoring and robotic operation control

Functions

autofocus(request)	Autofocus web interface endpoint (TODO: Implement).
calculate_celestial_data(obs_location)	Calculate positions of celestial objects for all-sky projection.
calculate_twilight_periods(start_time, ...)	Calculate twilight periods for the given time range.
clean_up()	Clean up and stop all observatory devices before shutdown.
close_observatory()	Close observatory and stop all operations safely.
complete_headers()	Complete FITS header processing for observatory images.
convert_fits_to_preview(fits_file)	Convert FITS astronomical image to JPEG bytes for web display.
cool_camera(device_name)	Initiate camera cooling to configured target temperature.
edit_schedule([schedule_data])	Update observatory schedule from web editor.
format_time(ftime)	Format datetime object to HH:MM:SS string.
get_schedule(request)	Serve schedule editor page with current schedule data.
get_video(request, filename)	Proxy video streams from observatory webcams.
guiding_data([day, since, telescope])	Get autoguider log data for plotting guiding performance.
heartbeat()	Get observatory heartbeat status for health monitoring.
latest_image_preview()	Serve the latest FITS image as a JPEG preview.
lifespan(app)	FastAPI lifespan context manager for startup and shutdown.
load_observatories()	Load and initialize all observatory configurations.
log(datetime[, limit])	Get observatory log entries before specified datetime.
main()	Main entry point for Astra observatory automation system.
observatory_db()	Get database connection for observatory logging.
polling(device_type[, day, since])	Get device polling data from observatory database.
roboticswitch()	Toggle observatory robotic operation mode.
schedule()	Get current observatory schedule with formatted times.
serve_files(request[, path])	Serve static files and main application interface.
sky_data()	Get celestial body positions for all-sky projection.
start_schedule()	Start executing the observatory's observation schedule.
start_watchdog()	Start observatory watchdog monitoring system.
stop_schedule()	Stop executing the observatory's observation schedule.
stop_watchdog()	Stop observatory watchdog monitoring system.
upload_schedule([file])	Upload schedule file to replace current observatory schedule.
websocket_endpoint(websocket)	Main WebSocket endpoint for real-time observatory status updates.
websocket_log(websocket)	WebSocket endpoint for real-time log streaming.

astra.main.observatory_db() → Connection

Get database connection for observatory logging.

Parameters:

name (str) – Observatory name for database file.

Returns:

sqlite3.Connection – Database connection object.

astra.main.load_observatories() → None

Load and initialize all observatory configurations.

Discovers observatory config files, creates Observatory instances, establishes device connections, and sets up filter wheel mappings. Updates global OBSERVATORY, WEBCAMFEED, and FWS dictionaries.

astra.main.clean_up() → None

Clean up and stop all observatory devices before shutdown.

Iterates through all observatories and device types to safely stop all connected devices. Handles exceptions during shutdown.

astra.main.format_time(ftime: datetime) → str | None

Format datetime object to HH:MM:SS string.

Parameters:

ftime (datetime.datetime) – Datetime object to format.

Returns:

str | None – Formatted time string or None if formatting fails.

astra.main.convert_fits_to_preview(fits_file: str) → tuple[bytes, dict]

Convert FITS astronomical image to JPEG bytes for web display.

Opens FITS file, extracts image data and headers, applies Z-scale normalization, and returns JPEG as bytes (no disk I/O).

Parameters:

fits_file (str) – Path to FITS file to convert.

Returns:

tuple[bytes, dict] – JPEG image bytes and extracted FITS headers.

astra.main.lifespan(app: FastAPI)

FastAPI lifespan context manager for startup and shutdown.

Handles application startup (loading observatories) and shutdown (cleaning up device connections) lifecycle events.

Parameters:

app (FastAPI) – FastAPI application instance.

Yields:

None – Application runs between yield statements.

async astra.main.get_video(request: Request, filename: str)

Proxy video streams from observatory webcams.

Forwards HTTP requests to webcam feeds, handling both MP4 video streams and HTML content with appropriate media types.

Parameters:

• request (Request) – FastAPI request object with headers.

• observatory (str) – Observatory name for webcam lookup.

• filename (str) – Video filename or path to stream.

Returns:

StreamingResponse – Proxied video content with appropriate headers.

async astra.main.heartbeat()

Get observatory heartbeat status for health monitoring.

Parameters:

observatory (str) – Observatory name to check.

Returns:

dict – JSON response with heartbeat status data.

async astra.main.latest_image_preview()

Serve the latest FITS image as a JPEG preview.

Returns the most recent observatory image converted to JPEG format, generated in-memory without disk I/O.

Returns:

StreamingResponse – JPEG image data with appropriate headers.

astra.main.close_observatory()

Close observatory and stop all operations safely.

Stops running schedule if active and closes the observatory. Logs all actions for audit trail.

Parameters:

observatory (str) – Observatory name to close.

Returns:

dict – JSON response with operation status.

astra.main.cool_camera(device_name: str)

Initiate camera cooling to configured target temperature.

Gets camera configuration and starts cooling process to the specified set temperature with defined tolerance.

Parameters:

• observatory (str) – Observatory name containing the camera.

• device_name (str) – Camera device name to cool.

Returns:

dict – JSON response with operation status and cooling details.

astra.main.complete_headers()

Complete FITS header processing for observatory images.

Parameters:

observatory (str) – Observatory name to process headers for.

Returns:

dict – JSON response with operation status.

async astra.main.start_watchdog()

Start observatory watchdog monitoring system.

Resets error states and starts the watchdog process for continuous observatory health monitoring.

Parameters:

observatory (str) – Observatory name to start watchdog for.

Returns:

dict – JSON response with operation status.

async astra.main.stop_watchdog()

Stop observatory watchdog monitoring system.

Parameters:

observatory (str) – Observatory name to stop watchdog for.

Returns:

dict – JSON response with operation status.

async astra.main.roboticswitch()

Toggle observatory robotic operation mode.

Parameters:

observatory (str) – Observatory name to toggle robotic mode for.

Returns:

dict – JSON response with current robotic switch state.

async astra.main.start_schedule()

Start executing the observatory’s observation schedule.

Parameters:

observatory (str) – Observatory name to start schedule for.

Returns:

dict – JSON response with operation status.

async astra.main.stop_schedule()

Stop executing the observatory’s observation schedule.

Parameters:

observatory (str) – Observatory name to stop schedule for.

Returns:

dict – JSON response with operation status.

async astra.main.schedule()

Get current observatory schedule with formatted times.

Parameters:

observatory (str) – Observatory name to get schedule for.

Returns:

list –

Schedule items with start/end times formatted as HH:MM:SS,

or empty list if no schedule exists.

async astra.main.edit_schedule(schedule_data: str = Body(..., media_type='text/plain'))

Update observatory schedule from web editor.

Parses JSONL schedule data and saves to observatory schedule file.

Parameters:

• observatory (str) – Observatory name to update schedule for.

• schedule_data (str) – JSONL formatted schedule data.

Returns:

dict – Status response with success/error information.

async astra.main.upload_schedule(file: UploadFile = File(...))

Upload schedule file to replace current observatory schedule.

Parameters:

• observatory (str) – Observatory name to upload schedule for.

• file (UploadFile) – Uploaded schedule file in JSONL format.

Returns:

dict – Upload status response with success/error information.

astra.main.calculate_twilight_periods(start_time: datetime, end_time: datetime, obs_location: EarthLocation) → list[dict]

Calculate twilight periods for the given time range.

Parameters:

• start_time – Start of time range (UTC)

• end_time – End of time range (UTC)

• obs_location – Observatory location

Returns:

List of period dictionaries with start, end, and phase

astra.main.calculate_celestial_data(obs_location: EarthLocation) → dict

Calculate positions of celestial objects for all-sky projection.

Parameters:

obs_location – Observatory location as EarthLocation

Returns:

Dictionary with observatory info, UTC time, and celestial body data

async astra.main.sky_data()

Get celestial body positions for all-sky projection.

Returns:

dict – JSON response with observatory location, time, and celestial body positions

async astra.main.polling(device_type: str, day: float = 1, since: str | None = None)

Get device polling data from observatory database.

Retrieves and processes telemetry data for specific device types, including pivot formatting, safety limits, and statistical grouping.

Parameters:

• observatory (str) – Observatory name to query data for.

• device_type (str) – Type of device (e.g., ‘ObservingConditions’).

• day (float) – Number of days back to retrieve data. Defaults to 1.

• since (str) – Optional timestamp to get only newer records.

Returns:

dict – Processed polling data with safety limits and latest values.

async astra.main.guiding_data(day: float = 1, since: str | None = None, telescope: str | None = None)

Get autoguider log data for plotting guiding performance.

Retrieves guiding corrections (post_pid_x, post_pid_y) from the autoguider_log table for visualization.

Parameters:

• day (float) – Number of days back to retrieve data. Defaults to 1.

• since (str) – Optional timestamp to get only newer records.

• telescope (str) – Optional telescope name to filter data.

Returns:

dict –

JSON response with guiding data including datetime,

telescope_name, post_pid_x, and post_pid_y values.

async astra.main.log(datetime: str, limit: int = 100)

Get observatory log entries before specified datetime.

Parameters:

• observatory (str) – Observatory name to query logs for.

• datetime (str) – Upper limit datetime for log entries.

• limit (int) – Maximum number of log entries to return. Defaults to 100.

Returns:

list – Log entries as dictionary records.

async astra.main.websocket_log(websocket: WebSocket)

WebSocket endpoint for real-time log streaming.

Provides initial log history and streams new log entries as they are added to the database. Also includes schedule modification time.

Parameters:

• websocket (WebSocket) – WebSocket connection object.

• observatory (str) – Observatory name for log streaming.

async astra.main.websocket_endpoint(websocket: WebSocket)

Main WebSocket endpoint for real-time observatory status updates.

Streams comprehensive observatory status including device polling data, system health, operational status, and latest images. Handles FITS to JPEG conversion for image display.

Parameters:

• websocket (WebSocket) – WebSocket connection object.

• observatory (str) – Observatory name for status monitoring.

async astra.main.autofocus(request: Request)

Autofocus web interface endpoint (TODO: Implement).

Placeholder for autofocus functionality that will process CSV files with FITS image references for focus analysis.

Parameters:

request (Request) – FastAPI request object.

Returns:

TemplateResponse – HTML template for autofocus interface.

async astra.main.get_schedule(request: Request)

Serve schedule editor page with current schedule data.

Loads raw JSONL schedule file preserving original datetime format for the web-based schedule editor interface.

Parameters:

• request (Request) – FastAPI request object.

• observatory (str) – Observatory name to load schedule for.

Returns:

TemplateResponse – HTML template with schedule editor and data.

async astra.main.serve_files(request: Request, path: str = '')

Serve static files and main application interface.

Handles routing for the main dashboard, favicon, JavaScript files, and frontend assets. Returns 404 for unknown paths.

Parameters:

• request (Request) – FastAPI request object.

• path (str) – Requested file path. Defaults to empty string for root.

Returns:

Union[TemplateResponse, FileResponse, HTMLResponse] –

Appropriate response

based on requested path.

astra.main.main()

Main entry point for Astra observatory automation system.

Parses command line arguments, configures logging, handles configuration reset, and starts the FastAPI server with specified options.