astra.main

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[source]#

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[source]#

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[source]#

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[source]#

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][source]#

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)[source]#

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)[source]#

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()[source]#

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()[source]#

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()[source]#

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)[source]#

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()[source]#

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()[source]#

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()[source]#

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()[source]#

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()[source]#

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()[source]#

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()[source]#

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'))[source]#

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(...))[source]#

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][source]#

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[source]#

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()[source]#

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)[source]#

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)[source]#

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)[source]#

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)[source]#

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)[source]#

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)[source]#

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)[source]#

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 = '')[source]#

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()[source]#

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.

astra.main

Contents

astra.main#