The Cloud SDK is part of the Formant Python package. It allows you to interact with the data and capabilities of the Formant cloud, which contains your organization's data and issues commands. Using the Cloud SDK, you can interact with the Formant platform programmatically, without having to use the web application or UI.
When should I use the Cloud SDK?You should use the Cloud SDK...
- To query datapoints from your Formant organization.
- To ingest datapoints to your Formant organization without using the Formant agent.
- For cloud-to-cloud integrations (AWS Lambda, Azure Functions, GCP, backend services) that send data on behalf of devices.
- For batch or historical data uploads from files or databases.
Installation
Prerequisite: Authenticate your session
You'll need to authenticate your session with the Cloud SDK. You can do this using your Formant credentials, or by creating a service account.
You can create a service account in the Formant web application, or using fctl, Formant's command-line tool.
Once you've generated your credentials, set the following environment variables:
export FORMANT_EMAIL="<your-email>"
export FORMANT_PASSWORD="<your-password>"The Formant Python package can be installed via pip:
pip install formantUsage
This section gives a broad overview of the capabilities of the Cloud SDK.
The full set of methods, syntax, and return values can be found in the Cloud SDK reference.
Instantiate the cloud client
To use the Cloud SDK, instantiate a single client and access APIs through admin, query, and ingest:
from formant.sdk.cloud.v2 import Client
client = Client()
admin_api = client.admin
query_api = client.query
ingest_api = client.ingestIngest data
Option 1: ingest_datapoints (dictionary payload)
ingest_datapoints (dictionary payload)Use ingest_datapoints with a simple {"items": [...]} payload:
from formant.sdk.cloud.v2 import Client
client = Client()
response = client.ingest_datapoints({
"items": [
{
"deviceId": "ced176ab-f223-4466-b958-ff8d35261529",
"name": "engine_temp",
"type": "numeric",
"tags": {"location": "sf"},
"points": [[1609804800000, 100]],
}
]
})Option 2: typed IngestionRequest
IngestionRequestFor typed helpers per datapoint type, use client.ingest.ingest.post():
from formant.sdk.cloud.v2 import Client
from formant.sdk.cloud.v2.src.resources.ingest import IngestionRequest
client = Client()
request = IngestionRequest("ced176ab-f223-4466-b958-ff8d35261529", tags={"location": "sf"})
request.add_numeric("engine_temp", 100)
response = client.ingest.ingest.post(request)
result = response.parsedFor more ingest examples, see GitHub: ingest.py.
Query datapoints
Use client.query.queries.query() with a Query model:
from formant.sdk.cloud.v2 import Client
from formant.sdk.cloud.v2.src.resources.queries import Query
import dateutil.parser as parser
client = Client()
query = Query(
start=parser.isoparse("2021-01-01T01:00:00.000Z"),
end=parser.isoparse("2021-01-01T02:00:00.000Z"),
device_ids=["99e8ee37-0a27-4a11-bba2-521facabefa3"],
names=["engine_temp"],
types=["numeric"],
tags={"location": ["sf", "la"]},
)
response = client.query.queries.query(query)
result = response.parsedFor more query examples, see GitHub: query.py.
Use
not_namesandnot_tagsonQueryto exclude stream names or tag sets from the query.
Query current stream value
Use client.query.stream_current.query() with a ScopeFilter:
from formant.sdk.cloud.v2 import Client
from formant.sdk.cloud.v2.src.resources.stream_current import ScopeFilter
import dateutil.parser as parser
client = Client()
scope_filter = ScopeFilter(
start=parser.isoparse("2021-01-01T00:00:00.000Z"),
end=parser.isoparse("2021-01-10T00:00:00.000Z"),
device_ids=["58d7f6e1-899d-4a8a-8c02-4c805cc8227f"],
)
response = client.query.stream_current.query(scope_filter=scope_filter)
result = response.parsedSee GitHub: query_stream_latest.py.
Query device online status
Query the $.agent.health stream over a recent time window:
from formant.sdk.cloud.v2 import Client
from formant.sdk.cloud.v2.src.resources.queries import Query
import datetime
client = Client()
now = datetime.datetime.now(tz=datetime.timezone.utc)
query = Query(
start=now - datetime.timedelta(seconds=10),
end=now,
device_ids=["58d7f6e1-899d-4a8a-8c02-4c805cc8227f"],
names=["$.agent.health"],
)
response = client.query.queries.query(query)
online = len(response.parsed.items) > 0See GitHub: online.py.
Commands
Use client.admin.commands.create() to send commands and client.admin.commands.query() to list undelivered commands:
from formant.sdk.cloud.v2 import Client
from formant.sdk.cloud.v2.src.resources.commands import Command, CommandParameter, CommandQuery
import datetime
client = Client()
admin_api = client.admin
device_id = "58d7f6e1-899d-4a8a-8c02-4c805cc8227f"
command = Command(
device_id=device_id,
command="start_recording",
parameter=CommandParameter(scrubber_time=datetime.datetime.now(tz=datetime.timezone.utc)),
organization_id=admin_api.organization_id,
)
response = admin_api.commands.create(command)
command_query = CommandQuery(device_id=device_id)
undelivered = admin_api.commands.query(command_query).parsedSee GitHub: commands.py.
Intervention requests
Use client.admin.interventions.create_request() and create_response():
from formant.sdk.cloud.v2 import Client
from formant.sdk.cloud.v2.src.resources.interventions import (
InterventionRequest,
CreateInterventionResponse,
)
client = Client()
request = InterventionRequest(
device_id="58d7f6e1-899d-4a8a-8c02-4c805cc8227f",
intervention_type="teleop",
message="Teleop requested",
)
response = client.admin.interventions.create_request(request)