routers

device_router

class ochra.manager.lab.routers.device_router.DeviceRouter(scheduler)

Bases: APIRouter

DeviceRouter is responsible for handling device-related API endpoints.

async call_device(identifier, args)

Call a method on the device.

Parameters:

  • identifier (str) – The ID or name of the device.

  • args (ObjectCallRequest) – The method call parameters.

Returns:

A dict representing the Operation model in JSON format.

Return type:

Dict[str, Any]

async construct_device(args)

Construct a new device in the lab.

Parameters:

args (ObjectConstructionRequest) – The construction parameters for the device.

Returns:

The ID of the constructed device.

Return type:

str

async delete_device(identifier)

Delete a device by its ID or name.

Parameters:

identifier (str) – The ID or name of the device.

Returns:

A message indicating the result of the deletion.

Return type:

Dict

async get_device(identifier)

Get a device by its ID or name.

Parameters:

identifier (str) – The ID or name of the device.

Returns:

The device data model.

Return type:

DataModel

async get_device_property(identifier, args)

Get properties of a device.

Parameters:
Returns:

The requested properties of the device.

Return type:

Any

async modify_device_property(identifier, args)

Modify properties of a device.

Parameters:
Returns:

True if the modification was successful, False otherwise.

Return type:

bool

lab_router

class ochra.manager.lab.routers.lab_router.LabRouter

Bases: APIRouter

LabRouter is responsible for handling lab-related API endpoints.

async get_lab_object(object_type, identifier)

Get a specific lab object by its identifier.

Parameters:

  • object_type (str) – The type of the lab object (e.g., “stations”).

  • identifier (str) – The unique identifier of the lab object (UUID or name).

Returns:

The lab object data.

Return type:

dict

Raises:

HTTPException – If the lab object is not found (404).

async get_lab_objects(object_type)

Get all lab objects of a specific type.

Parameters:

object_type (str) – The type of the lab objects (e.g., “stations”).

Returns:

A list of lab object data.

Return type:

list

Raises:

HTTPException – If no lab objects are found (404).

operation_results_router

class ochra.manager.lab.routers.operation_results_router.OperationResultRouter(folderpath)

Bases: APIRouter

OperationResultRouter is responsible for handling operation result-related API endpoints.

Parameters:

folderpath (str)

async construct_result(args)

Construct a new operation result in the lab.

Parameters:

args (ObjectConstructionRequest) – The construction parameters for the operation result.

Returns:

The ID of the constructed operation result.

Return type:

str

async get_data(identifier, background_tasks)

Get the data file associated with an operation result.

Parameters:

  • identifier (str) – The ID of the operation result.

  • background_tasks (BackgroundTasks) – Background tasks for cleanup.

Returns:

The file response containing the data.

Return type:

FileResponse

async get_property(identifier, args)

Get properties of an operation result.

Parameters:
Returns:

The requested properties of the operation result.

Return type:

Any

async get_result(identifier)

Get an operation result by its ID.

Parameters:

identifier (str) – The ID of the operation result.

Returns:

The operation result data model.

Return type:

DataModel

async modify_property(identifier, args)

Modify properties of an operation result.

Parameters:
Returns:

True if the modification was successful, False otherwise.

Return type:

bool

async put_data(identifier, file=File(PydanticUndefined))

Upload and associate a data file with an operation result.

Parameters:

  • identifier (str) – The ID of the operation result.

  • file (UploadFile) – The uploaded file.

Return type:

None

operation_router

class ochra.manager.lab.routers.operation_router.OperationRouter

Bases: APIRouter

OperationRouter is responsible for handling operation-related API endpoints.

async construct_op(args)

Construct a new operation in the lab.

Parameters:

args (ObjectConstructionRequest) – The construction parameters for the operation.

Returns:

The ID of the constructed operation.

Return type:

str

async get_op(identifier)

Get an operation by its ID.

Parameters:

identifier (str) – The ID of the operation.

Returns:

The operation data model.

Return type:

DataModel

async get_op_property(identifier, args)

Get properties of an operation.

Parameters:
Returns:

The requested properties of the operation.

Return type:

Any

async modify_op_property(identifier, args)

Modify properties of an operation.

Parameters:
Returns:

True if the modification was successful, False otherwise.

Return type:

bool

robot_router

class ochra.manager.lab.routers.robot_router.RobotRouter(scheduler)

Bases: APIRouter

RobotRouter is responsible for handling robot-related API endpoints.

async call_robot(identifier, args)

Call a method on the robot.

Parameters:

  • identifier (str) – The ID or name of the robot.

  • args (ObjectCallRequest) – The method call parameters.

Returns:

A dict representing the Operation model in JSON format.

Return type:

Dict[str, Any]

async construct_robot(args)

Construct a new robot in the lab.

Parameters:

args (ObjectConstructionRequest) – The construction parameters for the robot.

Returns:

The ID of the constructed robot.

Return type:

str

async delete_robot(identifier)

Delete a robot by its ID or name.

Parameters:

identifier (str) – The ID or name of the robot.

Returns:

A message indicating the result of the deletion.

Return type:

Dict

async get_property(identifier, args)

Get properties of a robot.

Parameters:
Returns:

The requested properties of the robot.

Return type:

Any

async get_robot(identifier)

Get a robot by its ID or name.

Parameters:

identifier (str) – The ID or name of the robot.

Returns:

The robot data model.

Return type:

DataModel

async modify_property(identifier, args)

Modify properties of a robot.

Parameters:
Returns:

True if the modification was successful, False otherwise.

Return type:

bool

station_router

class ochra.manager.lab.routers.station_router.StationRouter(scheduler)

Bases: APIRouter

StationRouter is responsible for handling station-related API endpoints.

async call_method(identifier, args)

Call a method on the station.

Parameters:

  • identifier (str) – The ID or name of the station.

  • args (ObjectCallRequest) – The method call parameters.

Returns:

A dict representing the Operation model in JSON format.

Return type:

Dict[str, Any]

async construct_station(args, request)

Construct a new station in the lab.

Parameters:
Returns:

The ID of the constructed station.

Return type:

str

async delete_station(identifier)

Delete a station by its ID or name.

Parameters:

identifier (str) – The ID or name of the station.

Returns:

A message indicating the result of the deletion.

Return type:

Dict

async get_station(identifier)

Get a station by its ID or name.

Parameters:

identifier (str) – The ID or name of the station.

Returns:

The station data model.

Return type:

DataModel

async get_station_property(identifier, args)

Get properties of a station.

Parameters:
Returns:

The requested properties of the station.

Return type:

Any

async modify_property(identifier, args)

Modify properties of a station.

Parameters:
Returns:

True if the modification was successful, False otherwise.

Return type:

bool

storage_router

class ochra.manager.lab.routers.storage_router.StorageRouter

Bases: APIRouter

StorageRouter is responsible for handling storage-related API endpoints.

async construct_storage_item(object_type, args)

Construct a new storage item in the lab.

Parameters:
Returns:

The ID of the constructed storage item.

Return type:

str

async delete_storage_item(object_type, identifier)

Delete a storage item by its ID.

Parameters:

  • identifier (str) – The ID of the storage item.

  • object_type (str)

Returns:

A message indicating the result of the deletion.

Return type:

Dict

async get_storage_item(object_type, identifier)

Get a storage item by its ID.

Parameters:

  • identifier (str) – The ID of the storage item.

  • object_type (str)

Returns:

The storage item data model.

Return type:

DataModel

async get_storage_item_property(object_type, identifier, args)

Get properties of a storage item.

Parameters:

  • identifier (str) – The ID of the storage item.

  • args (ObjectPropertyGetRequest) – The properties to retrieve.

  • object_type (str)

Returns:

The requested properties of the storage item.

Return type:

Any

async modify_storage_item_property(object_type, identifier, args)

Modify properties of a storage item.

Parameters:

  • identifier (str) – The ID of the storage item.

  • args (ObjectPropertyPatchRequest) – The properties to modify.

  • object_type (str)

Returns:

True if the modification was successful, False otherwise.

Return type:

bool

ui_router

class ochra.manager.lab.routers.ui_router.WebAppRouter(templates)

Bases: APIRouter

Parameters:

templates (Jinja2Templates)

build_table_fields()

Constructs a list of station information for rendering in the UI.

Returns:

A list of dictionaries containing station information.

Return type:

list[dict]

async get_device(request, station_id, device_id)

Handles GET requests to retrieve a specific device’s information.

Parameters:

  • request (Request) – The incoming HTTP request.

  • station_id (str) – The ID of the station the device belongs to.

  • device_id (str) – The ID of the device to retrieve.

Returns:

The rendered HTML response containing the device’s information.

Return type:

HTMLResponse

async get_device_view(request, station_id, device_id)

Handles GET requests to retrieve a specific device’s information.

Parameters:

  • request (Request) – The incoming HTTP request.

  • station_id (str) – The ID of the station the device belongs to.

  • device_id (str) – The ID of the device to retrieve.

Returns:

The rendered HTML response containing the device’s information.

Return type:

HTMLResponse

async get_login(request)

Renders the login page. :param request: The incoming HTTP request. :type request: Request

Returns:

The rendered HTML response for the login page.

Return type:

HTMLResponse

Parameters:

request (Request)

async get_register(request)

Renders the registration page.

Parameters:

request (Request) – The incoming HTTP request.

Returns:

The rendered HTML response for the registration page.

Return type:

HTMLResponse

async get_settings(request, edit=False)

Renders the settings page.

Parameters:

  • request (Request) – The incoming HTTP request.

  • edit (bool) – Flag to indicate if the settings are in edit mode. Default is False.

Returns:

The rendered HTML response for the settings page.

Return type:

HTMLResponse

async get_station(request, station_id)

Handles GET requests to retrieve a specific station’s information.

Parameters:

  • request (Request) – The incoming HTTP request.

  • station_id (str) – The ID of the station to retrieve.

Returns:

The rendered HTML response containing the station’s information.

Return type:

HTMLResponse

async get_stations(request)

Handles GET requests to retrieve stations information.

Parameters:

request (Request) – The incoming HTTP request.

Returns:

The rendered HTML response containing station information.

Return type:

HTMLResponse

async get_workflows(request)

Renders the workflows page.

Parameters:

request (Request) – The incoming HTTP request.

Returns:

The rendered HTML response for the workflows page.

Return type:

HTMLResponse

isHXRequest(request)

Checks if the request is an HTMX request.

Parameters:

request (Request) – The incoming HTTP request.

Returns:

True if the request is an HTMX request, False otherwise.

Return type:

bool

async post_command(request, station_id, device_id)

Handles POST requests to send a command to a specific device.

Parameters:

  • request (Request) – The incoming HTTP request.

  • station_id (str) – The ID of the station the device belongs to.

  • device_id (str) – The ID of the device to send the command to.

Returns:

The response indicating the result of the command.

Return type:

Response

Raises:

HTTPException – If the args format is invalid.

async post_login(username=Form(PydanticUndefined), password=Form(PydanticUndefined), db=Depends(get_db))

Handles user login.

Parameters:

  • username (str) – The username.

  • password (str) – The password.

  • db (Session) – Database session.

Returns:

A JSON response indicating success or failure.

Return type:

JSONResponse

Raises:

HTTPException – If authentication fails.

async post_logout(request, db=Depends(get_db))

Handles user logout by deleting the session token.

Parameters:

  • request (Request) – The incoming HTTP request.

  • db (Session) – Database session.

Returns:

A JSON response indicating successful logout.

Return type:

JSONResponse

Raises:

HTTPException – If no session token is found or session does not exist.

async post_register(username=Form(PydanticUndefined), email=Form(PydanticUndefined), password=Form(PydanticUndefined), confirm=Form(PydanticUndefined), db=Depends(get_db))

Handles user registration.

Parameters:

  • username (str) – The desired username.

  • email (str) – The user’s email address.

  • password (str) – The desired password.

  • confirm (str) – Password confirmation.

  • db (Session) – Database session.

Returns:

A JSON response indicating success or failure.

Return type:

JSONResponse

Raises:

HTTPException – If passwords do not match or username is already taken.

async put_settings(request, username=Form(PydanticUndefined), email=Form(PydanticUndefined), db=Depends(get_db))

Handles updating user settings.

Parameters:

  • request (Request) – The incoming HTTP request.

  • username (str) – The new username.

  • email (str) – The new email address.

  • db (Session) – Database session. Defaults to a dependency injection of get_db.

Returns:

The rendered HTML response for the settings page.

Return type:

HTMLResponse

Raises:

HTTPException – If the user is not authenticated or not found.