Formant Python SDK

class formant.sdk.agent.v1.Client(agent_url='unix:///var/lib/formant/agent.sock', enable_logging=True, ignore_throttled=False, ignore_unavailable=False)

A client for interacting with the Formant agent. Automatically handles connection and reconnection to the agent. There are methods for:

  • Ingesting telemetry datapoints

  • Creating events

  • Handling commands

  • Ingesting transform frames

  • Reading application configuration

  • Handling teleop control datapoints

__init__()
Parameters

  • agent_url – The address of the Formant agent API

  • enable_logging – If True, this client will log some information to stdout

  • ignore_throttled – If True, telemetry datapoint throttle errors will not raise Exceptions. Throttled datapoints are still valid for teleop

  • ignore_unavailable – If True, Formant agent unavailable errors will not raise Exceptions.

post_text(stream, value, tags=None, timestamp=None)

Post a text datapoint to a stream.

Parameters

  • stream (str) – The name of the Formant stream to post the datapoint on

  • value (str) – The text datapoint value

  • tags (Optional[Dict[str, str]]) – Tags to include on the posted datapoint

  • timestamp (Optional[int]) – Unix timestamp for the posted datapoint. Uses the current time by default

from formant.sdk.agent.v1 import Client

fclient = Client()
fclient.post_text(
    "example.text",
    "Processed 9 items"
)
Return type

None

post_json(stream, value, tags=None, timestamp=None)

Post a JSON datapoint to a telemetry stream.

Parameters

  • stream (str) – The name of the Formant stream to post the datapoint on

  • value (str) – The encoded JSON datapoint value

  • tags (Optional[Dict[str, str]]) – Tags to include on the posted datapoint

  • timestamp (Optional[int]) – Unix timestamp for the posted datapoint. Uses the current time by default

Return type

None

post_numeric(stream, value, tags=None, timestamp=None)

Post a numeric datapoint to a telemetry stream.

Parameters

  • stream (str) – The name of the Formant stream to post the datapoint on

  • value (Union[float, int]) – The numeric datapoint value

  • tags (Optional[Dict[str, str]]) – Tags to include on the posted datapoint

  • timestamp (Optional[int]) – Unix timestamp for the posted datapoint. Uses the current time by default

Return type

None

post_numericset(stream, numerics_dict, tags=None, timestamp=None)

Post a numeric set datapoint to a telemetry stream. Numeric sets are collections of related numeric datapoints.

Parameters

  • stream (str) – The name of the Formant stream to post the datapoint on

  • numerics_dict (Dict[str, Tuple[Union[float, int], Optional[str]]]) – The numeric set datapoint value, a dictionary mapping names to (numeric value, units) tuples.

  • tags (Optional[Dict[str, str]]) – Tags to include on the posted datapoint

  • timestamp (Optional[int]) – Unix timestamp for the posted datapoint. Uses the current time by default

from formant.sdk.agent.v1 import Client

fclient = Client()
fclient.post_numericset(
    "example.numericset2",
    {
        "frequency": (998, "Hz"),
        "usage": (30, "percent"),
        "warp factor": (6.0, None),
    },
)
Return type

None

post_image(stream, value=None, url=None, content_type='image/jpg', tags=None, timestamp=None)

Post an image datapoint to a telemetry stream.

Parameters

  • stream (str) – The name of the Formant stream to post the datapoint on

  • value (Optional[bytes]) – The datapoint value: raw bytes of an encoded image or frame

  • url (Optional[str]) – The datapoint url: path to local file or valid remote URL for remote files

  • content_type (Literal[‘image/jpg’, ‘image/png’, ‘video/h264’]) – The format of the encoded image or frame. Defaults to “image/jpg”.

  • tags (Optional[Dict[str, str]]) – Tags to include on the posted datapoint

  • timestamp (Optional[int]) – Unix timestamp for the posted datapoint. Uses the current time by default

Return type

None

post_bitset(stream, bitset_dict, tags=None, timestamp=None)

Post a bitset datapoint to a telemetry stream. A bitset is a collection of related boolean states.

Parameters

  • stream (str) – The name of the Formant stream to post the datapoint on

  • bitset_dict (Dict[str, bool]) – The datapoint value, a dictionary mapping names to booleans

  • tags (Optional[Dict[str, str]]) – Tags to include on the posted datapoint

  • timestamp (Optional[int]) – Unix timestamp for the posted datapoint. Uses the current time by default

from formant.sdk.agent.v1 import Client

fclient = Client()
fclient.post_bitset(
    "example.bitset",
    {
        "standing": False,
        "walking": False,
        "sitting": True
    }
)
Return type

None

post_geolocation(stream, latitude, longitude, tags=None, timestamp=None)

Post a geolocation datapoint to a telemetry stream.

Parameters

  • stream (str) – The name of the Formant stream to post the datapoint on

  • latitude (Union[float, int]) – The datapoint value’s latitude

  • longitude (Union[float, int]) – The datapoint value’s longitude

  • tags (Optional[Dict[str, str]]) – Tags to include on the posted datapoint

  • timestamp (Optional[int]) – Unix timestamp for the posted datapoint. Uses the current time by default

Return type

None

post_battery(stream, percentage, voltage=None, current=None, charge=None, tags=None, timestamp=None)

Post a battery datapoint to a telemetry stream. Only percentage is required.

Parameters

  • stream (str) – The name of the Formant stream to post the datapoint on

  • percentage (Union[int, float]) – The battery charge percentage

  • voltage (Union[int, float, None]) – The battery voltage

  • current (Union[int, float, None]) – The battery current

  • charge (Union[int, float, None]) – The battery charge

  • tags (Optional[Dict[str, str]]) – Tags to include on the posted datapoint

  • timestamp (Optional[int]) – Unix timestamp for the posted datapoint. Uses the current time by default

Return type

None

post_transform_frame(parent_frame, child_frame, tx, ty, tz, rx, ry, rz, rw)

Adds a transform frame, used to position datapoints in 3D space.

Parameters

  • parent_frame (str) – The parent frame of the posted transform

  • child_frame (str) – The child frame of the posted transform

  • tx (Union[int, float]) – x-translation

  • ty (Union[int, float]) – y-translation

  • tz (Union[int, float]) – z-translation

  • rx (Union[int, float]) – x-rotation (quaternion)

  • ry (Union[int, float]) – y-rotation (quaternion)

  • rz (Union[int, float]) – z-rotation (quaternion)

  • rw (Union[int, float]) – w-rotation (quaternion)

Return type

None

create_event(message, tags=None, timestamp=None, end_timestamp=None, notify=False, severity='info')

Creates and ingests an event.

Parameters

  • message (str) – The text payload of the event

  • tags (Optional[Dict[str, str]]) – Tags to include on the event

  • timestamp (Optional[int]) – Unix starting timestamp for the event. Uses the current time by default

  • end_timestamp (Optional[int]) – Unix ending timestamp for the event. Must be greater than timestamp. If end_timestamp is supplied, the event will span a length of time

  • notify (bool) – If True, the created event will trigger a Formant notification

  • severity (Literal[‘info’, ‘warning’, ‘critical’, ‘error’]) – The severity level of the event

from formant.sdk.agent.v1 import Client

fclient = Client()
fclient.create_event(
    "Confinement beam to warp frequency 0.4e17 hz",
    tags={"Region": "North"},
    notify=True,
    severity="warning"
)
Return type

None

get_command_request(command_filter=None)

If there is a command request in the agent’s queue whose “command” value matches an element of the given command filter, takes and returns the command request. Otherwise, returns None if there are no matching command requests in the agent’s queue.

Parameters

command_filter (Optional[List[str]]) – A list of command names. This method only returns commands whose names are in this list.

Return type

Optional[CommandRequest]

send_command_response(request_id, success, datapoint=None)

Sends a command response for an identified command request to Formant. Returns an error if there was a problem sending the command response.

Parameters

  • request_id (str) – The ID of the command request to which this method responds

  • request_id – Whether the command was successfully executed

  • datapoint (Optional[Datapoint]) – A datapoint related to the command

Return type

None

register_command_request_callback(f, command_filter=None)

Command requests issued to the agent whose “command” value matches an element of the given command filter will be streamed into the provided callback. If no command filter is provided, all command requests will be handled.

Parameters

  • f (Callable[[CommandRequest], None]) – A callback that will be executed on command requests as they are received by the Formant agent.

  • command_filter (Optional[List[str]]) – A list of command names. The provided callback is only exectued on commands whose names are in this list

Return type

None

unregister_command_request_callback(f)

Unregisters previously registered command request callback.

Parameters

f (Callable[[CommandRequest], None]) – The command request callback to be unregistered

Return type

None

register_teleop_callback(f, stream_filter=None)

Control datapoints received from teleop whose “stream” value matches an element of the given stream filter will be streamed into the provided callback. If no stream filter is provided, control datapoints from all streams will be received.

Parameters

  • f (Callable[[ControlDatapoint], None]) – A callback that will be executed on teleop control datapoints as they are received by the Formant agent

  • stream_filter (Optional[List[str]]) – A list of stream names. The provided callback is only exectued on control datapoints whose names are in this list

Return type

None

unregister_teleop_callback(f)

Unregisters previously registered teleop callback.

Parameters

f (Callable[[ControlDatapoint], None]) – The teleop callback to be unregistered

Return type

None

get_teleop_info()

Returns current information about teleop connection count.

Return type

GetTeleopInfoResponse

register_teleop_heartbeat_callback(f)

The provided callback will be called once each time a heartbeat is received over Formant teleop. Heartbeats are streamed from the operator machine at 20Hz on a UDP-like channel. This method can be used to quickly detect teleop disconnections.

Parameters

f – A callback that will be called when a heartbeat is received

unregister_teleop_heartbeat_callback(f)

Unregisters previously registered teleop heartbeat callback.

Parameters

f – The teleop heartbeat callback to be unregistered

get_app_config(key, *args)

Returns the value for the given key that was set in Formant application configuration for this device, or returns the given default value.

Parameters

  • key (str) – The application configuration key

  • args (Any) – (One additional argument) The default value to return if the key is not found.

Return type

Optional[str]

get_config_blob_data()

Returns the blob data defined in the device configuration.

Return type

str

register_config_update_callback(f)

Adds a function to the list of callbacks that are executed by the client when this device receives updated configuration from Formant.

Parameters

f (Callable) – The configuration update callback to be registered

Return type

None

unregister_config_update_callback(f)

Removes a function from the list of callbacks that are executed by the client when this device receives updated configuration from Formant.

Parameters

f (Callable) – The configuration update callback to be unregistered

Return type

None