How to store a value with its time¶

A reading such as a temperature or a position often means little without the time it was taken. A field of type Timed[T] keeps a value and its timestamp together, so they are written in one go and a reader never sees a new value with an old time. The timestamp counts nanoseconds since the Unix epoch, and it either comes from the box at the moment of the write or is a time you already have, such as one a device reported.

1. Declare a timed field¶

Wrap the field's type in Timed. The type inside can be anything a field could hold on its own, so text still needs its capacity:

from typing import Annotated

from sharedbox import Capacity, SharedBox, SharedStream, Timed


class Thermometer(SharedBox, name="example-thermometer"):
    temperature: Timed[float] = Timed(0.0, 0)
    status: Timed[Annotated[str, Capacity(16)]] = Timed("", 0)
    history: Annotated[list[Timed[float]], Capacity(4)] = []

A field always takes a Timed, so a default is a Timed too, and assigning a plain 21.5 raises TypeError.

2. Let the write stamp the value¶

Timed.now asks the write for the time. The box reads the system clock once the write holds the box's lock, so the stamps of writes from any process follow the order in which the writes happened, as long as nobody sets the system clock back:

thermometer = Thermometer()
thermometer.temperature = Timed.now(21.5)
reading = thermometer.temperature
print(reading.value)  # 21.5
print(reading.timestamp_ns > 0)  # True
print(type(reading.timestamp))  # <class 'float'>

timestamp_ns is the stored integer, and the timestamp property gives the same time in seconds as a float, the form time.time() returns.

3. Keep a time you already have¶

A Timed built with a timestamp of its own is stored as it is, which keeps the time a device or another clock gave the value:

device_ns = 1_762_000_000_123_456_789
thermometer.status = Timed("ready", device_ns)
print(thermometer.status)  # Timed(value='ready', timestamp_ns=1762000000123456789)

4. Put timed values in collections, records and streams¶

A Timed can go wherever a record can: in a list, as a dict's value, inside a record or an optional, and as a stream's item. Every Timed.now in one write gets the same time, because the clock is read once per write:

thermometer.history = [Timed.now(21.4), Timed.now(21.5)]
first, second = thermometer.history
print(first.timestamp_ns == second.timestamp_ns)  # True

A stream stamps each item when it is sent:

with SharedStream.create(Timed[float], "example-temperatures", capacity=4) as stream:
    reader = stream.reader()
    with stream.sender() as sender:
        sender.send(Timed.now(21.6))
    print(reader.receive().value)  # 21.6

A Timed can't be a set's element or a dict's key, because two equal values taken at different times would count as different keys; defining such a field raises TypeError.

"""The script of the guide "How to store a value with its time"."""
# ruff: noqa: RUF012

from typing import Annotated

from sharedbox import Capacity, SharedBox, SharedStream, Timed


class Thermometer(SharedBox, name="example-thermometer"):
    temperature: Timed[float] = Timed(0.0, 0)
    status: Timed[Annotated[str, Capacity(16)]] = Timed("", 0)
    history: Annotated[list[Timed[float]], Capacity(4)] = []



thermometer = Thermometer()
thermometer.temperature = Timed.now(21.5)
reading = thermometer.temperature
print(reading.value)  # 21.5
print(reading.timestamp_ns > 0)  # True
print(type(reading.timestamp))  # <class 'float'>

device_ns = 1_762_000_000_123_456_789
thermometer.status = Timed("ready", device_ns)
print(thermometer.status)  # Timed(value='ready', timestamp_ns=1762000000123456789)

thermometer.history = [Timed.now(21.4), Timed.now(21.5)]
first, second = thermometer.history
print(first.timestamp_ns == second.timestamp_ns)  # True

with SharedStream.create(Timed[float], "example-temperatures", capacity=4) as stream:
    reader = stream.reader()
    with stream.sender() as sender:
        sender.send(Timed.now(21.6))
    print(reader.receive().value)  # 21.6

thermometer.close()
Thermometer.unlink()
SharedStream.unlink("example-temperatures")