ephys_link.server

WebSocket server and communication handler

Manages the WebSocket server and handles connections and events from the client. For every event, the server does the following:

Extract the arguments passed in the event
Log that the event was received
Call the appropriate function in ephys_link.sensapex_handler with arguments
Relay the response from ephys_link.sensapex_handler to the callback function

Classes

Server

class Server[source]

Bases: object

async connect(sid, _, __) → bool[source]

Acknowledge connection to the server.

Parameters:

sid (str) – Socket session ID.
_ (dict) – WSGI formatted dictionary with request info (unused).
__ (dict) – Authentication details (unused).

Returns:

False on error to refuse connection. True otherwise.

Return type:

bool

async disconnect(sid) → None[source]

Acknowledge disconnection from the server.

Parameters:: sid (str) – Socket session ID.
Returns:: None

async static get_version(_) → str[source]

Get the version number of the server.

Parameters:: _ (str) – Socket session ID (unused).
Returns:: Version number as defined in ephys_link.__about__.
Return type:: str

async get_manipulators(_) → str[source]

Get the list of discoverable manipulators.

Parameters:: _ (str) – Socket session ID (unused).
Returns:: ephys_link.common.GetManipulatorsOutputData as JSON formatted string.
Return type:: str

async register_manipulator(_, manipulator_id: str) → str[source]

Register a manipulator with the server.

Parameters:

_ (str) – Socket session ID (unused).
manipulator_id (str) – ID of the manipulator to register.

Returns:

Error message on error, empty string otherwise.

Return type:

str

async unregister_manipulator(_, manipulator_id: str) → str[source]

Unregister a manipulator from the server.

Parameters:

_ (str) – Socket session ID (unused)
manipulator_id (str) – ID of the manipulator to unregister.

Returns:

Error message on error, empty string otherwise.

Return type:

str

async get_pos(_, manipulator_id: str) → str[source]

Position of manipulator request.

Parameters:

_ (str) – Socket session ID (unused).
manipulator_id (str) – ID of manipulator to pull position from.

Returns:

ephys_link.common.PositionalOutputData as JSON formatted string.

Return type:

str

async get_angles(_, manipulator_id: str)[source]

Angles of manipulator request.

Parameters:

_ (str) – Socket session ID (unused).
manipulator_id (str) – ID of manipulator to pull angles from.

Returns:

ephys_link.common.AngularOutputData as JSON formatted string.

Return type:

str

async get_shank_count(_, manipulator_id: str) → str[source]

Number of shanks of manipulator request.

Parameters:

_ (str) – Socket session ID (unused).
manipulator_id (str) – ID of manipulator to pull number of shanks from.

Returns:

ephys_link.common.ShankCountOutputData as JSON formatted string.

Return type:

str

async goto_pos(_, data: str) → str[source]

Move manipulator to position.

Parameters:

_ (str) – Socket session ID (unused).
data (str) – ephys_link.common.GotoPositionInputDataFormat as JSON formatted string.

Returns:

ephys_link.common.PositionalOutputData as JSON formatted string.

Return type:

str

async drive_to_depth(_, data: str) → str[source]

Drive to depth.

Parameters:

_ (str) – Socket session ID (unused).
data (str) – ephys_link.common.DriveToDepthInputDataFormat as JSON formatted string.

Returns:

ephys_link.common.DriveToDepthOutputData as JSON formatted string.

Return type:

str

async set_inside_brain(_, data: str) → str[source]

Set the inside brain state.

Parameters:

_ (str) – Socket session ID (unused).
data (str) – ephys_link.common.InsideBrainInputDataFormat as JSON formatted string.

Returns:

ephys_link.common.StateOutputData as JSON formatted string.

Return type:

str

async calibrate(_, manipulator_id: str) → str[source]

Calibrate manipulator.

Parameters:

_ (str) – Socket session ID (unused).
manipulator_id (str) – ID of manipulator to calibrate.

Returns:

Error message on error, empty string otherwise.

Return type:

str

async bypass_calibration(_, manipulator_id: str) → str[source]

Bypass calibration of manipulator.

Parameters:

_ (str) – Socket session ID (unused).
manipulator_id (str) – ID of manipulator to bypass calibration.

Returns:

Error message on error, empty string otherwise.

Return type:

str

async set_can_write(_, data: str) → str[source]

Set manipulator can_write state.

Parameters:

_ (str) – Socket session ID (unused)
data (str) – ephys_link.common.CanWriteInputDataFormat as JSON formatted string.

Returns:

ephys_link.common.StateOutputData as JSON formatted string.

Return type:

str

stop(_) → bool[source]

Stop all manipulators.

Parameters:: _ (str) – Socket session ID (unused).
Returns:: True if successful, False otherwise.
Return type:: bool

async static catch_all(_, __, data: Any) → str[source]

Catch all event.

Parameters:

_ (str) – Socket session ID (unused).
__ (str) – Client ID (unused).
data (Any) – Data received from client.

Returns:

“UNKNOWN_EVENT” response message.

Return type:

str

launch(platform_type: str, server_port: int, pathfinder_port: int | None = None, ignore_updates: bool = False) → None[source]

Launch the server.

Parameters:

platform_type (str) – Parsed argument for platform type.
server_port (int) – HTTP port to serve the server.
pathfinder_port (int) – Port New Scale Pathfinder’s server is on.
ignore_updates (bool) – Flag to ignore checking for updates.

Returns:

None

close_server(_, __) → None[source]: Close the server.