app.api.device_websocket#

Device websocket connection.

This module defines the WebSocket endpoints for managing devices. It includes functionalities for: - Device registration and status updates via WebSocket. - Listening for commands from devices.

Attributes#

`LOG_CALL_DELIMITER`
`DATA_LAKE_DIR`
`DAGSTER_GRAPHQL_URL`
`oauth2_scheme`
`router`
`dict_id_websocket`
`device_last_seen`
`task_sse_queues`
`_SSE_TERMINAL_STATUSES`
`_HEARTBEAT_TIMEOUT_S`
`_HEARTBEAT_CHECK_INTERVAL_S`

Classes#

TaskEvent

Payload for the push-event endpoint — callable by any internal service.

Functions#

`send_json`(websocket, payload)	Send json payload via websocket.
`_broadcast_task_status`(→ None)	Push a status event to all SSE subscribers watching a task.
`push_task_event`(→ dict)	Internal endpoint: any service pushes a status event to all SSE subscribers of a task.
`task_stream`(]) → starlette.responses.StreamingResponse)	SSE endpoint — streams real-time task status updates to the browser.
`trigger_acquisition`(task_id, access_token)	Trigger an MRI acquisition for the given task.
`connection_with_valid_id_and_token`(→ uuid.UUID)	Check if the given device_id and device_token belong to an existing device in the database.
`monitor_devices`()	Monitor device online/offline status via the ping heartbeat.
`websocket_endpoint`(websocket)	Websocket endpoint for device communication.
`handle_register`(→ None)	Handle device registration.
`handle_status_update`(→ None)	Handle device status updates from the device SDK.
`_stream_to_file`(→ tuple[int, str])	Stream bytes from a WebSocket into a temp file; returns (bytes_received, sha256_hex).
`_submit_reconstruction_job`(→ None)	Submit the mrpro_reconstruction_job to Dagster after a file transfer completes.
`handle_file_transfer`(→ None)	Handle file transfer from device to server.
`_pick_result_type`(filename)	Map extensions to enum.

Module Contents#

app.api.device_websocket.LOG_CALL_DELIMITER = '-------------------------------------------------------------------------------'#

app.api.device_websocket.DATA_LAKE_DIR#

app.api.device_websocket.DAGSTER_GRAPHQL_URL = 'http://dagster-webserver:3000/dagster/graphql'#

app.api.device_websocket.oauth2_scheme#

app.api.device_websocket.router#

app.api.device_websocket.dict_id_websocket: dict[uuid.UUID, fastapi.WebSocket]#

app.api.device_websocket.device_last_seen: dict[uuid.UUID, float]#

app.api.device_websocket.task_sse_queues: dict[str, list[asyncio.Queue]]#

async app.api.device_websocket.send_json(websocket, payload: dict)#: Send json payload via websocket.

async app.api.device_websocket._broadcast_task_status(task_id: str, event: dict) → None#: Push a status event to all SSE subscribers watching a task.

app.api.device_websocket._SSE_TERMINAL_STATUSES#

class app.api.device_websocket.TaskEvent#

Bases: pydantic.BaseModel

Payload for the push-event endpoint — callable by any internal service.

source: str#

task_status: str#

progress: int = 0#

message: str = ''#

async app.api.device_websocket.push_task_event(task_id: str, event: TaskEvent) → dict#

Internal endpoint: any service pushes a status event to all SSE subscribers of a task.

Called by Dagster sensors on job success, failure, or cancellation. No authentication required — only reachable on the internal Docker network.

async app.api.device_websocket.task_stream(task_id: str, token: Annotated[str, Query()]) → starlette.responses.StreamingResponse#

SSE endpoint — streams real-time task status updates to the browser.

The browser connects with EventSource and receives JSON events of the form:: {“task_status”: “INPROGRESS”, “progress”: 45}

The stream closes automatically when the task reaches FINISHED or ERROR. A keepalive comment is sent every 25 s to prevent proxy timeouts.

async app.api.device_websocket.trigger_acquisition(task_id: uuid.UUID, access_token: Annotated[str, Depends(oauth2_scheme)])#

Trigger an MRI acquisition for the given task.

Fetches the task from the protocol manager, looks up the assigned sequence and device, sends the scan-start command via the device’s open WebSocket, and marks the task as STARTED.

async app.api.device_websocket.connection_with_valid_id_and_token(websocket: fastapi.WebSocket) → uuid.UUID#: Check if the given device_id and device_token belong to an existing device in the database.

app.api.device_websocket._HEARTBEAT_TIMEOUT_S = 15#

app.api.device_websocket._HEARTBEAT_CHECK_INTERVAL_S = 5#

async app.api.device_websocket.monitor_devices()#

Monitor device online/offline status via the ping heartbeat.

Catches every disconnection a clean WebSocketDisconnect can’t: crashed processes, killed processes (SIGKILL), dropped network links, etc. — any device that stops pinging is marked OFFLINE within roughly _HEARTBEAT_TIMEOUT_S + _HEARTBEAT_CHECK_INTERVAL_S seconds, independent of whether the client ever got a chance to shut down cleanly.

async app.api.device_websocket.websocket_endpoint(websocket: fastapi.WebSocket)#: Websocket endpoint for device communication.

Args#

websocket (WebSocket): The WebSocket connection object. device_id (UUID): The device_id.

async app.api.device_websocket.handle_register(websocket: fastapi.WebSocket, message: dict, device_id: uuid.UUID) → None#: Handle device registration.

async app.api.device_websocket.handle_status_update(websocket: fastapi.WebSocket, message: dict, device_id: uuid.UUID) → None#: Handle device status updates from the device SDK.

async app.api.device_websocket._stream_to_file(websocket: fastapi.WebSocket, tmp_path: pathlib.Path, size_bytes: int) → tuple[int, str]#: Stream bytes from a WebSocket into a temp file; returns (bytes_received, sha256_hex).

app.api.device_websocket._submit_reconstruction_job(task_id: str, task_dir: str, protocol_id: str, user_access_token: str) → None#: Submit the mrpro_reconstruction_job to Dagster after a file transfer completes.

async app.api.device_websocket.handle_file_transfer(websocket: fastapi.WebSocket, header: dict, device_id: uuid.UUID) → None#: Handle file transfer from device to server.

app.api.device_websocket._pick_result_type(filename: str)#: Map extensions to enum.

app.api.device_websocket#

Attributes#

Classes#

Functions#

Module Contents#

Args#