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, force])	Update observatory schedule from web editor.
filters_by_camera(cameras)	Return the filters each camera can load.
fits_explorer_root_redirect()	Redirect /fits_explorer -> /fits_explorer/ so the mounted sub-app handles it.
format_time(ftime)	Format datetime object to HH:MM:SS string.
get_allsky_image([name])	Serve the latest All-Sky camera image.
get_parser()
get_schedule(request)	Serve schedule editor page with current schedule data.
get_timelapse_frame(camera, frame)	Serve one saved time-lapse frame.
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.
list_timelapse_frames(camera)	List the saved time-lapse frames for a camera, oldest first.
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.
parse_schedule_jsonl(schedule_data)	Parse JSONL schedule text into a list of rows.
polling(device_type[, day, since])	Get device polling data from observatory database.
read_log(q)	Run a log query with its own connection, so it works in any thread.
reconnect_devices()	Restart all device processes, connect again, and restart polling.
reconnect_devices_task()	Reconnect all devices, then read the filter wheel names again.
roboticswitch()	Toggle observatory robotic operation mode.
schedule()	Get current observatory schedule with formatted times.
schedule_schema_loader()	Describe the action value fields of every action type.
schedule_template_loader([schemas])	Build the default action value template for every action type.
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.
to_json_safe(value)	Convert numpy/pandas scalar values to JSON-serializable Python types.
update_filter_wheels(obs)	Read the filter names of each filter wheel into the global FWS.
upload_schedule([file])	Upload schedule file to replace current observatory schedule.
validate_schedule([schedule_data])	Check a schedule from the web editor without saving it.
validate_schedule_items(schedule_items)	Check every schedule row by building its action config.
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, WEBCAMFEEDS, and FWS dictionaries.

astra.main.update_filter_wheels(obs: Observatory) → None

Read the filter names of each filter wheel into the global FWS.

Parameters:

obs (Observatory) – Observatory whose filter wheels are read.

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.to_json_safe(value)

Convert numpy/pandas scalar values to JSON-serializable Python types.

astra.main.schedule_schema_loader() → dict

Describe the action value fields of every action type.

The schedule editor builds its form from this, so that the inputs it shows always match the action configs rather than a hand-kept copy of them.

Returns:

dict – Action type mapped to its list of field descriptors.

astra.main.schedule_template_loader(schemas: dict | None = None) → dict

Build the default action value template for every action type.

A nested config marked with flatten metadata is merged into the parent template, because that is the layout the action config reads back.

The schedule editor no longer needs this: it reads the defaults from the field descriptors, which also carry the type and the description. The function stays because it is part of the documented API.

Parameters:

schemas (dict | None) – Field descriptors per action type. Loaded when not given.

Returns:

dict – Action type mapped to its default action value.

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.heartbeat()

Get observatory heartbeat status for health monitoring.

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.

async astra.main.get_allsky_image(name: str | None = None)

Serve the latest All-Sky camera image.

Parameters:

name (str, optional) – Name of the specific all-sky camera. If not provided, returns the first camera.

Returns:

FileResponse – The image file with no-store cache headers.

async astra.main.list_timelapse_frames(camera: str)

List the saved time-lapse frames for a camera, oldest first.

Parameters:

camera (str) – Camera id, as set in the feed’s timelapse_id.

Returns:

dict –

JSON response with frames (list of {file, time}) and

last_error.

async astra.main.get_timelapse_frame(camera: str, frame: str)

Serve one saved time-lapse frame.

Parameters:

• camera (str) – Camera id.

• frame (str) – Frame file name from /api/timelapse/{camera}.

Returns:

FileResponse – The frame image. Frames never change, so browsers may cache them.

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.

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:

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.

Returns:

dict – JSON response with operation status.

astra.main.reconnect_devices_task() → None

Reconnect all devices, then read the filter wheel names again.

astra.main.reconnect_devices()

Restart all device processes, connect again, and restart polling.

Use this to recover from lost devices without restarting Astra. The watchdog stops, every device process is replaced, the devices connect again, polling starts again, and the watchdog starts again. Device changes in the config file are used. The robotic switch stays off.

The work runs in a background thread; the log shows the progress. The request is refused while a schedule, a reconnect, or another task that uses the devices runs (an action or a guider).

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. Refused while a device reconnect runs, because the reconnect starts the watchdog itself.

Returns:

dict – JSON response with operation status.

async astra.main.stop_watchdog()

Stop observatory watchdog monitoring system.

Returns:

dict – JSON response with operation status.

astra.main.roboticswitch()

Toggle observatory robotic operation mode.

Returns:

dict – JSON response with current robotic switch state.

astra.main.start_schedule()

Start executing the observatory’s observation schedule.

Returns:

dict – JSON response with operation status.

astra.main.stop_schedule()

Stop executing the observatory’s observation schedule.

Returns:

dict – JSON response with operation status.

astra.main.schedule()

Get current observatory schedule with formatted times.

Returns:

list –

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

or empty list if no schedule exists.

astra.main.filters_by_camera(cameras: list[str]) → dict

Return the filters each camera can load.

A camera reaches its filters through the wheel it is paired with, so the schedule editor can offer the right names per row rather than every filter in the observatory. A camera with no wheel, or one whose wheel does not report its names, maps to an empty list and the editor leaves that field free text.

Parameters:

cameras (list[str]) – Camera names to look up.

Returns:

dict –

Camera name mapped to its list of filter names. The key

"__all__" holds every filter across all wheels, used as the fallback when a camera cannot be matched to a wheel.

astra.main.validate_schedule_items(schedule_items: list[dict]) → list[dict]

Check every schedule row by building its action config.

This runs the same validation the scheduler runs when it loads the file, so the editor reports a missing required field or a bad value before the row reaches the schedule rather than when the action is due to start.

Parameters:

schedule_items (list[dict]) – Parsed schedule rows.

Returns:

list –

One entry per bad row, with the keys row (zero-based index)

and message.

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

Check a schedule from the web editor without saving it.

Parameters:

schedule_data (str) – JSONL formatted schedule data.

Returns:

dict – Status response whose data holds the per-row errors.

astra.main.parse_schedule_jsonl(schedule_data: str) → list[dict]

Parse JSONL schedule text into a list of rows.

Parameters:

schedule_data (str) – JSONL formatted schedule data.

Returns:

list – The parsed rows.

Raises:

json.JSONDecodeError – If any line is not valid JSON.

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

Update observatory schedule from web editor.

Parses JSONL schedule data and saves to observatory schedule file. A row that fails validation stops the write, because the scheduler would reject the file when it next loads it. Pass force to save a draft anyway.

Parameters:

• schedule_data (str) – JSONL formatted schedule data.

• force (bool) – Save even when a row fails validation.

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:

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

astra.main.sky_data()

Get celestial body positions for all-sky projection.

A plain def, so FastAPI runs it in a worker thread. The astropy calculation and the telescope location read can take seconds on the first call, and in the event loop they would stop the websocket messages.

Returns:

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

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

Get device polling data from observatory database.

Uses an append-only in-memory cache keyed by device_type:

• First request performs a full DB query (up to day days) and stores the result.

• Subsequent requests fetch only the rows newer than last_cached_timestamp - 2 min, reprocess that overlap window to ensure the most-recent 60 s bucket is complete, merge into the cache, and prune rows older than the requested day window.

• Concurrent requests share a single threading.Lock so only one DB fetch runs at a time; any waiter re-uses the freshly-updated cache.

Parameters:

• device_type (str) – Type of device (e.g. 'ObservingConditions').

• day (float) – Days of history to include in the response. Defaults to 1.

• since (str) – Optional ISO-format UTC timestamp; if supplied the response payload is filtered to rows newer than this value (the cache itself always holds the full window).

Returns:

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

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.

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

Get observatory log entries before specified datetime.

Parameters:

• 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.

astra.main.read_log(q: str) → DataFrame

Run a log query with its own connection, so it works in any thread.

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.

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.

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.

Returns:

TemplateResponse – HTML template with schedule editor and data.

async astra.main.fits_explorer_root_redirect()

Redirect /fits_explorer -> /fits_explorer/ so the mounted sub-app handles it.

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.get_parser()

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.