Write data

Write samples to channels with the Python and TypeScript clients: one-call writes, writers, auto-commit, writer modes, auto-index, and the rules of writes.

Write samples that you already have in one call. To write samples as they arrive, open a writer. Every write must obey the rules of writes.

The examples on this page use an index channel time and a data channel temperature that time indexes. To create channels like these, see Create channels.

Write samples in one call

write stores samples that you already have, such as data from a file:

import synnax as sy

start = sy.TimeStamp.now()
client.write(
    start,
    {
        "time": [start + sy.TimeSpan.SECOND * i for i in range(5)],
        "temperature": [22.5, 22.6, 22.8, 23.1, 23.0],
    },
)

import { TimeSpan, TimeStamp } from "@synnaxlabs/client";

const start = TimeStamp.now();
await client.write(start, {
  time: [0, 1, 2, 3, 4].map((i) => start.add(TimeSpan.seconds(i))),
  temperature: [22.5, 22.6, 22.8, 23.1, 23.0],
});

start is the timestamp of the first sample. The write fails when any of the channels already has data in the time range of the samples.

Each call opens a writer, writes, and closes the writer. Streamers do not get these samples. Do not call write in a loop to send live data: each call is a full round trip to the Core. Use a writer instead.

Write with a writer

A writer stays open and writes many frames to its channels. Open it with a start timestamp and the channels it writes to:

with client.open_writer(
    start=sy.TimeStamp.now(),
    channels=["time", "temperature"],
) as writer:
    for i in range(100):
        writer.write({"time": sy.TimeStamp.now(), "temperature": 20 + i * 0.1})
        sy.sleep(0.1)

const writer = await client.openWriter({
  start: TimeStamp.now(),
  channels: ["time", "temperature"],
});
try {
  for (let i = 0; i < 100; i++) {
    await writer.write({ time: TimeStamp.now(), temperature: 20 + i * 0.1 });
    await new Promise((resolve) => setTimeout(resolve, 100));
  }
} finally {
  await writer.close();
}

This writer sends one sample to each channel every 100 milliseconds. A writer can stay open for days.

The writer sends samples to the Core, which stores them and sends them to streamers. The writer mode sets which of the two the Core does:

Write formats

write takes the samples in several shapes. When a write holds samples for one channel of an index, it must hold samples for all the channels of the writer that share that index, as Rule 1 says.

import numpy as np
import pandas as pd

now = sy.TimeStamp.now()

# One sample for each channel.
writer.write({"time": now, "temperature": 22.5})

# Many samples, as lists or NumPy arrays.
writer.write({
    "time": [now, now + sy.TimeSpan.SECOND],
    "temperature": np.array([22.5, 22.6], dtype=np.float32),
})

# A list of channels and a list of samples.
writer.write(["time", "temperature"], [[now], [22.5]])

# A DataFrame, with one column for each channel.
writer.write(pd.DataFrame({"time": [now], "temperature": [22.5]}))

# A frame.
writer.write(sy.Frame({"time": [now], "temperature": [22.5]}))

import { Frame, Series } from "@synnaxlabs/client";

const now = TimeStamp.now();

// One sample for each channel.
await writer.write({ time: now, temperature: 22.5 });

// Many samples.
await writer.write({
  time: [now, now.add(TimeSpan.SECOND)],
  temperature: [22.5, 22.6],
});

// A list of channels and a list of samples.
await writer.write(["time", "temperature"], [[now], [22.5]]);

// A frame. Typed arrays are the fastest way to write many samples. Their type
// must match the data type of the channel.
await writer.write(
  new Frame({
    time: new Series({
      data: new BigInt64Array([now.valueOf()]),
      dataType: "timestamp",
    }),
    temperature: new Series(new Float32Array([22.5])),
  }),
);

Channels with the string or json data type take strings and objects in the same writes. For more about frames, see Series and frames.

Errors

write does not wait for the Core. When the Core rejects a write, the next call to write, commit, or close raises the error. After an error, the writer accepts no more writes: close it, and open a new writer.

Auto-commit

A commit makes written samples available to reads. By default, a writer commits after each write.

To make a group of writes available together, turn auto-commit off and call commit after the group. commit waits for the Core and returns the end of the committed time range. Samples that you did not commit are lost when the writer closes.

with client.open_writer(
    start=sy.TimeStamp.now(),
    channels=["time", "temperature"],
    enable_auto_commit=False,
) as writer:
    for path in files:
        for frame in read_chunks(path):
            writer.write(frame)
        writer.commit()

const writer = await client.openWriter({
  start: TimeStamp.now(),
  channels: ["time", "temperature"],
  enableAutoCommit: false,
});
try {
  for (const path of files) {
    for await (const frame of readChunks(path)) await writer.write(frame);
    await writer.commit();
  }
} finally {
  await writer.close();
}

Here, files and read_chunks stand for your own code. Each file becomes available to reads when its commit returns.

Do not commit after each small write: each commit waits for a round trip to the Core. Commit every few seconds, or keep auto-commit on.

With auto-commit on, the Core saves its index of the samples to disk at most once per interval. The auto_index_persist_interval option sets the interval, which is one second by default. When the Core stops without warning, it can lose the samples of the last interval.

Writer modes

Set mode when you open the writer:

• persist -> the Core stores the samples, but does not send them to streamers. Use it for historical data.
• stream -> the Core sends the samples to streamers, but does not store them. commit has no effect. Use it for data that you need only live.
• persist_stream -> the Core stores the samples and sends them to streamers. This is the default.

writer = client.open_writer(
    start=sy.TimeStamp.now(),
    channels=["time", "temperature"],
    mode="persist",
)

const writer = await client.openWriter({
  start: TimeStamp.now(),
  channels: ["time", "temperature"],
  mode: "persist",
});

Write without timestamps

With auto_index on, the Core makes the timestamps. Each write that leaves out the index channel gets the time at which the Core receives it. The samples in one write are 1 nanosecond apart. When the writer is also open on the index channel, a write that includes it uses your timestamps.

with client.open_writer(
    start=0,
    channels=["temperature"],
    auto_index=True,
) as writer:
    writer.write({"temperature": 22.5})

const writer = await client.openWriter({
  channels: ["temperature"],
  autoIndex: true,
});
try {
  await writer.write("temperature", 22.5);
} finally {
  await writer.close();
}

The writer adds the index channel of temperature by itself. The timestamps include the time that a sample takes to reach the Core. For sensor data, send the time at which you read each sample.

Writer options

The rules of writes

The Core rejects a write that breaks one of these rules. Rules 1 and 2 do not apply to virtual channels, which store no data.

Rule 1

In one write, an index channel and each of its data channels must have the same number of samples, and each channel of the writer that shares the index must be in the write.

Here, temperature and pressure are data channels that time indexes, and a writer is open on all three. This write is valid:

{
  "time": [1791028800000000000, 1791028800100000000, 1791028800200000000],
  "temperature": [19.17, 19.18, 19.19],
  "pressure": [16.22, 17.22, 18.22],
}

These writes are not valid:

{
  "time": [1791028800000000000, 1791028800100000000, 1791028800200000000],
  "temperature": [19.17, 19.18, 19.19],
  // One sample is missing.
  "pressure": [16.22, 17.22],
}

{
  "time": [1791028800000000000, 1791028800100000000, 1791028800200000000],
  "temperature": [19.17, 19.18, 19.19],
  // The pressure channel is missing.
}

Rule 2

Write to an index channel before or together with its data channels.

A writer that is open on time and temperature must write time in each write, unless auto-index is on. To write only temperature, open a writer on temperature alone after the timestamps are in time, with start at the first of those timestamps. Use this to calculate a new channel from stored data with the same index.

What the rules mean for your channels

All data channels that share an index must have samples at the same times, so they must have the same sample rate. Give a channel with a different rate its own index.

Command channels usually need their own index too. You often write one command at a time, such as to open one valve, wait, and open a second valve. Commands that do not have the same timestamps cannot share an index.