Tips and tutorials

Tips and tutorials for beginners and experts alike

    Formant Cloud Python SDK

    This tutorial gives a very broad overview of the capabilities of the Formant cloud Python SDK, as well as linking to examples for each capability. The cloud Python SDK can ingest and query datapoints directly with the Formant cloud backend. Unlike the Formant agent SDK, it does not require the agent to be running. For more details and latest information, visit our github repo.

    Prerequisites

    1. Create a service user account

    2. Set the service user access credentials as environment variables on your machine

    export FORMANT_EMAIL=<your Formant account email address>
    export FORMANT_PASSWORD=<your Formant account password>

    3. Install formant sdk

    pip install formant

    OR

    pip3 install formant

    Instantiating the Formant Agent client

    Each of the following requires an instantiated, connected Formant API client. That can be done with the following snippet:

    from formant.sdk.cloud.v1 import Client as FormantClient
    fclient = FormantClient()

    Ingesting data

    Use the following method to ingest data into Formant without an agent. If you are ingest on a device with agent, refer to agent Python SDK

    An example to ingest a numeric datapoint below. Please note that you need to ingest this as an administrator. For examples on ingest other datatypes like videos or bitsets, please visit our github repo.

    from formant.sdk.cloud.v1 import Client as FormantClient
    
    
    if __name__ == "__main__":
        fclient = FormantClient()
    
        device_id = "572f5ff3-63e9-4687-b242-c0d8a4891d80"
        timestamp = 1609804800000
        tags = {"location": "sf"}
        ingest_result = fclient.ingest(
            {
                "items": [
                    {
                        "deviceId": device_id,
                        "name": "engine_temp",
                        "type": "numeric",
                        "tags": tags,
                        "points": [[timestamp, 100]],
                    },
                ]
            }
        )

    ingest has the following definition (with example parameters).

    def ingest(self, params):
            {
                deviceId: "ced176ab-f223-4466-b958-ff8d35261529",
                name: "engine_temp",
                type: "numeric",
                tags: {"location":"sf"},
                points: [...],
            }

    Querying data

    Data stored within your organization can be queried using devices or tags or type of streams. start and end are the only parameters required. 

    An example to query all datapoints tagged with location:sf

    from formant.sdk.cloud.v1 import Client as FormantClient
    
    if __name__ == "__main__":
        fclient = FormantClient()
    
        query_result = fclient.query(
            {
                "start": "2021-01-01T00:00:00.000Z",
                "end": "2021-01-10T00:00:00.000Z",
                "tags": {"location": ["sf"]},
            }
        )
        print(query_result)

    query has the following definition (with example parameters). 

    def query(self, params):
            {
                start: "2021-01-01T01:00:00.000Z",
                end: "2021-01-01T02:00:00.000Z",
                deviceIds: ["99e8ee37-0a27-4a11-bba2-521facabefa3"],
                names: ["engine_temp"],
                types: ["numeric"],
                tags: {"location":["sf","la"]},
                notNames: ["speed"],
            }


    Send commands to robot 

    Send commands to the robot from the cloud. For more details on the type of commands, please read out commands deep dive.

    import datetime
    
    from formant.sdk.cloud.v1 import Client as FormantClient
    
    def get_current_isodate():
        return datetime.datetime.now(tz=datetime.timezone.utc).isoformat()
    
    if __name__ == "__main__":
        # to authenticate set FORMANT_EMAIL and FORMANT_PASSWORD
        # environment variables for an existing service account
        fclient = FormantClient()
    
    
        device_id = "43be1d4b-2178-46b8-81b5-61cf865e264b"
    
    
        print("Sending command ...")
        command_result = fclient.create_command(
            {
                "deviceId": device_id,
                "command": "start_recording",
                "parameter": {"scrubberTime": get_current_isodate()},
            }
        )
        print(command_result)
        print("\n")


    Send commands with payload to robot 

    Send commands (directives) to the robot from the Cloud. For more details on commands, please read out commands deep dive

    import datetime
    from formant.sdk.cloud.v1 import Client as FormantClient
    def get_current_isodate():
        return datetime.datetime.now(tz=datetime.timezone.utc).isoformat()
    if __name__ == "__main__":
        # file id obtained by uploading to formant file storage
        # see files.py for an example
        # to authenticate set FORMANT_EMAIL and FORMANT_PASSWORD    
        # environment variables for an existing service account
        fclient = FormantClient(
            admin_api="http://localhost:8080",
        )
        print("Uploading file ...")
        file_id = fclient.upload_file(
            {
                "path": "/tmp/test.log",
            }
        )
        
        print("Sending command with large payload ...")
        command_result = fclient.create_command(
            {
                "deviceId": device_id,
                "command": "snapshot",
                "parameter": {
                    "scrubberTime": get_current_isodate(),
                    "files": [{"id": file_id, "name": "view_0.jpg"}],
                },
            }
        )
    
        print(command_result)
        print("\n")
    
        print("Querying for device's currently undelivered commands ...")
        query_commands_result = fclient.query_commands({"deviceId": device_id})
        print(query_commands_result)


    Query last known value of a device 

    Query last known value of a device from the Cloud.

    from formant.sdk.cloud.v1 import Client as FormantClient
    if __name__ == "__main__":
        # to authenticate set FORMANT_EMAIL and FORMANT_PASSWORD
        # environment variables for an existing service account
        fclient = FormantClient()
        device_id = "572f5ff3-63e9-4687-b242-c0d8a4891d80"
        # query last known value of all streams from a device
        # within a given period
        query_result = fclient.query_stream_current_value(
            {
                "start": "2021-01-01T00:00:00.000Z",
                "end": "2021-01-10T00:00:00.000Z",
                "deviceIds": [device_id],
            }
        )
        print(query_result)
        # query last known value of all streams matching a name
        query_result = fclient.query_stream_current_value(
            {
                "names": ["engine_temp"],
            }
        )
        # query last known value of all streams matching a tag
        query_result = fclient.query_stream_current_value(
            {
                "tags": {"location": ["sf"]},
            }
        )




    © 2020 Formant • 1999 Bryant St · San Francisco, CA 94110