Nominal can connect to a variety of 3rd-party API services, time series databases, and cloud file stores (e.g. S3). These connections are displayed on the Connections page of your Nominal tenant.

In addition to support for connecting to external data sources, Nominal also has its own internal time series database. This guide details how to establish a connection to this database and stream channel data to it in Python.

Streaming Data to Nominal

Prerequisites

Make sure you have the nominal Python packages installed. You can install it using:

1
pip3 install nominal

Connect to Nominal

Get your Nominal API token from your User settings page.

See the Quickstart for more details on connecting to Nominal from Python.

1
import nominal.nominal as nm
2
3
nm.set_token(
4
    base_url = 'https://api.gov.nominal.io/api',
5
    token = '* * *' # Replace with your Access Token from 
6
                    # https://app.gov.nominal.io/settings/user?tab=tokens
7
)

Creating a Connection to Nominal’s Internal Time Series Database

To stream data to a Connection, first create a connection to Nominal’s internal time series database. The datasource_id must be unique within your organization, and connection_name is a friendly name that will appear on the Nominal platform under all connections.

1
connection = nm.create_streaming_connection(
2
    datasource_id="my-unique-datasource-id",
3
    connection_name="my-connection-name"
4
)

Streaming Data Non-Blocking to the Connection

The Nominal client provides a convenient way to write data to the connection without blocking your application during network requests, which can be slow compared to your sample rate.

Writing a Single Data Point

First, we’ll write a single data point to verify that it arrives correctly on the Nominal platform.

1
from datetime import datetime
2
3
with connection.get_nominal_write_stream(batch_size=1) as stream:
4
    now = datetime.now()
5
    stream.enqueue(channel_name="my_channel_name", timestamp=now, value=0.0)

In this example:

• channel_name: You can choose any name for your channel.
• timestamp: Can be a Python datetime object, a string in ISO format (e.g., “2024-12-31T15:00:00.001Z”), or a time in nanoseconds since the Unix epoch.

Viewing the Data Point in Nominal

After executing this code snippet, we will create a Run and a Workbook, so we can see the data point.

Creating a run

1
run = nm.create_run(name="my_run_name", start=now, end=None)
2
run.add_connection("some_ref", connection)

Creating a Workbook

1. Navigate to the Run section on the Nominal platform.
2. Locate and select your run named “my_run_name”.
3. Click on “New Workbook” on the top right-hand side of the page and select “New Empty Workbook”.
4. In the left pane under “Channel Search”, you should see “my_channel_name”.
5. Drag and drop “my_channel_name” into the “Time Series Chart”.
6. Click the play button at the top middle of the page to display the data point.

You should see the data point that written.

Streaming Continuous Data

Now, let’s stream more data points to this channel.

1
import random
2
from time import sleep
3
4
with connection.get_nominal_write_stream(batch_size=1) as stream:
5
    for i in range(1000):
6
        now = datetime.now()
7
        value = random.random()
8
        print(f"Sending data point at {now}: {value}")
9
        stream.enqueue(channel_name="my_channel_name", timestamp=now, value=value)
10
        sleep(1)

After running this code, you can see new data points appearing in your Workbook. To view the latest data points more effectively, select “Last Minute” in the “Time History” dropdown at the top middle of the page.

Streaming Multi-Channel Data to the Connection

The stream allows for streaming multiple channels, with different intervals simultaneously.

1
with connection.get_nominal_write_stream(max_wait_sec=2) as stream:
2
    for i in range(1000):
3
        now = datetime.now()
4
        value1 = random.random()
5
        msg = f"Sending data at {now}: channel1={value1}"
6
7
        stream.enqueue(channel_name="my_channel_name", timestamp=now, value=value1)
8
        if int(now.timestamp()) % 2 == 0:
9
            value2 = random.random()
10
            msg += f", channel2={value2}"
11
            stream.enqueue(channel_name="my_channel_name_2", timestamp=now, value=value2)
12
        print(msg)
13
        sleep(1)

After executing the code above, refresh your Workbook in the browser. In the “Channel Search” section, you should now see “my_channel_name_2”. Drag and drop it into the “Time Series Chart” and click the play button at the top middle of the page to view the data.