`smartem_backend.api_client`#

Members

`EntityConverter`	Handles conversions between EPU data model and API request/response models.
`SSEAgentClient`	SSE client for agents to receive real-time instructions from the backend.
`SmartEMAPIClient`	SmartEM API client that provides synchronous HTTP interface.

class smartem_backend.api_client.EntityConverter[source]#

Handles conversions between EPU data model and API request/response models. Separating this conversion logic keeps the main client code cleaner.

static acquisition_to_request(entity: AcquisitionData) → AcquisitionCreateRequest[source]#: Convert EPU session data to acquisition request model

static grid_to_request(entity: GridData, lowmag: bool = False) → GridCreateRequest[source]#: Convert Grid data to grid request model

static gridsquare_to_request(entity: GridSquareData, lowmag: bool = False) → GridSquareCreateRequest[source]#: Convert GridSquareData to grid square request model

static foilhole_to_request(entity: FoilHoleData) → FoilHoleCreateRequest[source]#: Convert FoilHoleData to foil hole request model

static micrograph_to_request(entity: MicrographData) → MicrographCreateRequest[source]#: Convert MicrographData to micrograph request model

static atlas_to_request(entity: AtlasData) → AtlasCreateRequest[source]#: Convert AtlasData to atlas request model

static atlas_tile_to_request(entity: AtlasTileData) → AtlasTileCreateRequest[source]#: Convert AtlasTileData to atlas tile request model

static gridsquare_position_to_request(entity: AtlasTileGridSquarePositionData) → GridSquarePositionRequest[source]#: Convert AtlasTileData to atlas tile request model

class smartem_backend.api_client.SmartEMAPIClient(base_url: str, timeout: float = 10.0, logger=None)[source]#

SmartEM API client that provides synchronous HTTP interface.

This client handles all API communication with the SmartEM Core API, provides data conversion between EPU data models and API request/response models, and maintains a cache of entity IDs.

Initialize the SmartEM API client

Parameters:

base_url – Base URL for the API
timeout – Request timeout in seconds
logger – Optional custom logger instance

close() → None[source]#: Close the client connection

get_status() → dict[str, object][source]#: Get API status information

get_health() → dict[str, object][source]#: Get API health check information

get_acquisitions() → list[AcquisitionResponse][source]#: Get all acquisitions

create_acquisition(acquisition: AcquisitionData) → AcquisitionResponse[source]#: Create a new acquisition

get_acquisition(acquisition_uuid: str) → AcquisitionResponse[source]#: Get a single acquisition by ID

update_acquisition(acquisition: AcquisitionData) → AcquisitionResponse[source]#: Update an acquisition

delete_acquisition(acquisition_uuid: str) → None[source]#: Delete an acquisition

get_grids() → list[GridResponse][source]#: Get all grids

get_grid(grid_uuid: str) → GridResponse[source]#: Get a single grid by ID

update_grid(grid: GridData) → GridResponse[source]#: Update a grid

delete_grid(grid_uuid: str) → None[source]#: Delete a grid

get_acquisition_grids(acquisition_uuid: str) → list[GridResponse][source]#: Get all grids for a specific acquisition

create_acquisition_grid(grid: GridData) → GridResponse[source]#: Create a new grid for a specific acquisition

get_atlases() → list[AtlasResponse][source]#: Get all atlases

get_atlas(atlas_uuid: str) → AtlasResponse[source]#: Get a single atlas by ID

update_atlas(atlas: AtlasData) → AtlasResponse[source]#: Update an atlas

delete_atlas(atlas_uuid: str) → None[source]#: Delete an atlas

get_grid_atlas(grid_uuid: str) → AtlasResponse[source]#: Get the atlas for a specific grid

create_grid_atlas(atlas: AtlasData) → AtlasResponse[source]#: Create a new atlas for a grid

get_atlas_tiles() → list[AtlasTileResponse][source]#: Get all atlas tiles

get_atlas_tile(tile_uuid: str) → AtlasTileResponse[source]#: Get a single atlas tile by ID

update_atlas_tile(tile: AtlasTileData) → AtlasTileResponse[source]#: Update an atlas tile

delete_atlas_tile(tile_uuid: str) → None[source]#: Delete an atlas tile

get_atlas_tiles_by_atlas(atlas_uuid: str) → list[AtlasTileResponse][source]#: Get all tiles for a specific atlas

create_atlas_tile_for_atlas(tile: AtlasTileData) → AtlasTileResponse[source]#: Create a new tile for a specific atlas

link_atlas_tile_and_gridsquare(gridsquare_position: AtlasTileGridSquarePositionData) → AtlasTileGridSquarePositionResponse[source]#: Link a grid square to a tile

link_atlas_tile_and_gridsquares(gridsquare_positions: list[AtlasTileGridSquarePositionData]) → list[AtlasTileGridSquarePositionResponse][source]#: Link multiple grid squares to a tile

get_gridsquares() → list[GridSquareResponse][source]#: Get all grid squares

get_gridsquare(gridsquare_uuid: str) → GridSquareResponse[source]#: Get a single grid square by ID

update_gridsquare(gridsquare: GridSquareData, lowmag: bool = False) → GridSquareResponse[source]#: Update a grid square

delete_gridsquare(gridsquare_uuid: str) → None[source]#: Delete a grid square

get_grid_gridsquares(grid_uuid: str) → list[GridSquareResponse][source]#: Get all grid squares for a specific grid

create_grid_gridsquare(gridsquare: GridSquareData, lowmag: bool = False) → GridSquareResponse[source]#: Create a new grid square for a specific grid

get_foilholes() → list[FoilHoleResponse][source]#: Get all foil holes

get_foilhole(foilhole_uuid: str) → FoilHoleResponse[source]#: Get a single foil hole by ID

update_foilhole(foilhole: FoilHoleData) → FoilHoleResponse[source]#: Update a foil hole

delete_foilhole(foilhole_uuid: str) → None[source]#: Delete a foil hole

get_gridsquare_foilholes(gridsquare_uuid: str) → list[FoilHoleResponse][source]#: Get all foil holes for a specific grid square

create_gridsquare_foilholes(gridsquare_uuid: str, foilholes: list[FoilHoleData], allow_on_grid_bar: bool = False) → list[FoilHoleResponse][source]#: Create a new foil hole for a specific grid square

get_micrographs() → list[MicrographResponse][source]#: Get all micrographs

get_micrograph(micrograph_uuid: str) → MicrographResponse[source]#: Get a single micrograph by ID

update_micrograph(micrograph: MicrographData) → MicrographResponse[source]#: Update a micrograph

delete_micrograph(micrograph_id: str) → None[source]#: Delete a micrograph

get_foilhole_micrographs(foilhole_id: str) → list[MicrographResponse][source]#: Get all micrographs for a specific foil hole

create_foilhole_micrograph(micrograph: MicrographData) → MicrographResponse[source]#: Create a new micrograph for a specific foil hole

acknowledge_instruction(agent_id: str, session_id: str, instruction_id: str, acknowledgement: AgentInstructionAcknowledgement) → AgentInstructionAcknowledgementResponse[source]#: Acknowledge an instruction from the agent

get_active_connections() → dict[source]#: Get active agent connections (debug endpoint)

get_session_instructions(session_id: str) → dict[source]#: Get instructions for a session (debug endpoint)

class smartem_backend.api_client.SSEAgentClient(base_url: str, agent_id: str, session_id: str, timeout: int = 30, max_retries: int = 10, initial_retry_delay: float = 1.0, max_retry_delay: float = 60.0)[source]#

SSE client for agents to receive real-time instructions from the backend. This is separate from the main ApiClient as it handles long-lived connections.

Initialize SSE client for agent communication

Parameters:

base_url – Base URL of the API server
agent_id – Unique identifier for this agent/microscope
session_id – Current microscopy session ID
timeout – Connection timeout in seconds
max_retries – Maximum number of reconnection attempts
initial_retry_delay – Initial delay between retries in seconds
max_retry_delay – Maximum delay between retries in seconds

stream_instructions(instruction_callback: Callable[[dict], None], connection_callback: Callable[[dict], None] | None = None, error_callback: Callable[[Exception], None] | None = None) → None[source]#

Start streaming instructions via SSE (blocking)

Parameters:

instruction_callback – Called when an instruction is received
connection_callback – Called when connection events occur (optional)
error_callback – Called when errors occur (optional)

async stream_instructions_async(instruction_callback: Callable[[dict], None], connection_callback: Callable[[dict], None] | None = None, error_callback: Callable[[Exception], None] | None = None, auto_retry: bool = True) → None[source]#

Start streaming instructions via SSE (async with auto-retry and exponential backoff)

Parameters:

instruction_callback – Called when an instruction is received
connection_callback – Called when connection events occur (optional)
error_callback – Called when errors occur (optional)
auto_retry – Whether to automatically retry on connection failures

acknowledge_instruction(instruction_id: str, status: str, result: str | None = None, error_message: str | None = None, processing_time_ms: int | None = None, retry_count: int = 3) → AgentInstructionAcknowledgementResponse[source]#

Acknowledge an instruction with retry logic

Parameters:

instruction_id – ID of the instruction to acknowledge
status – Status of acknowledgement (‘received’, ‘processed’, ‘failed’, ‘declined’)
result – Optional result message
error_message – Optional error message if status is ‘failed’
processing_time_ms – Time taken to process the instruction in milliseconds
retry_count – Number of retry attempts for acknowledgement

get_stats() → dict[source]#: Get client connection and performance statistics.

reset_stats() → None[source]#: Reset client statistics.

is_connected() → bool[source]#: Check if the client is currently connected and streaming.

send_heartbeat(retry_count: int = 3) → bool[source]#

Send a heartbeat to the backend to update connection health status

Parameters:: retry_count – Number of retry attempts for heartbeat
Returns:: True if heartbeat was sent successfully, False otherwise
Return type:: bool

stop()[source]#: Stop the SSE stream

smartem_backend.api_client#

This Page

`smartem_backend.api_client`#