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:

  1. Extract the arguments passed in the event

  2. Log that the event was received

  3. Call the appropriate function in ephys_link.sensapex_handler with arguments

  4. 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 get_pinpoint_id() → str[source]

Get the pinpoint ID.

Returns:

Pinpoint ID and whether the client is a requester.

Return type:

tuple[str, bool]

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:

vbl_aquarium.models.ephys_link.GetManipulatorsResponse 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:

vbl_aquarium.models.ephys_link.PositionalResponse as JSON formatted string.

Return type:

str

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

Angles of manipulator request.

Parameters:

  • _ (str) – Socket session ID (unused).

  • manipulator_id (str) – ID of manipulator to pull angles from.

Returns:

vbl_aquarium.models.ephys_link.AngularResponse 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:

vbl_aquarium.models.ephys_link.ShankCountResponse 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) – vbl_aquarium.models.ephys_link.GotoPositionRequest as JSON formatted string.

Returns:

vbl_aquarium.models.ephys_link.PositionalResponse 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) – vbl_aquarium.models.ephys_link.DriveToDepthRequest as JSON formatted string.

Returns:

vbl_aquarium.models.ephys_link.DriveToDepthResponse 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) – vbl_aquarium.models.ephys_link.InsideBrainRequest as JSON formatted string.

Returns:

vbl_aquarium.models.ephys_link.BooleanStateResponse 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) – vbl_aquarium.models.ephys_link.CanWriteRequest as JSON formatted string.

Returns:

vbl_aquarium.models.ephys_link.BooleanStateResponse 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

async launch_setup(platform_type: str, pathfinder_port: int, ignore_updates) → None[source]
async launch_for_proxy(proxy_address: str, port: int, platform_type: str, pathfinder_port: int | None, ignore_updates: bool) → None[source]

Launch the server in proxy mode.

Parameters:

  • proxy_address (str) – Proxy IP address.

  • port (int) – Port to serve the server.

  • platform_type (str) – Parsed argument for platform type.

  • pathfinder_port (int) – Port New Scale Pathfinder’s server is on.

  • ignore_updates (bool) – Flag to ignore checking for updates.

Returns:

None

launch(platform_type: str, port: int, pathfinder_port: int | None, ignore_updates: bool) → None[source]

Launch the server.

Parameters:

  • platform_type (str) – Parsed argument for platform type.

  • 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

bind_events() → None[source]

Bind Ephys Link events to the server.

close_server(_, __) → None[source]

Close the server.