Recording

A [[recorder]] keeps a lossless, per-tick history of selected signals, taken by the engine itself. OPC UA subscriptions sample and can miss ticks; a recorder sees every tick it is asked for. The cycle thread only copies the selected values into a preallocated ring buffer: no allocation, lock, I/O or wait. Everything else (files, compression, counters, remote pulls) happens on other threads, and a sink that falls behind loses rows instead of slowing the cycle.

[[recorder]]
id = "main"
signals = ["plant.*", "ctrl.u[*]", "ctrl.qp_flag"]
every = 1
sink = "file"
dir = "records"

Every key is in the project file reference.

cycle thread --copy, never waits--> ring (buffer_s) --drain thread--> run files      (sink = "file")
                                         |
                                         +--read_rows over OPC UA--> taktwerk record --> run files

Two sinks

• sink = "file" (the default): on an edge PC or a workstation. A drain thread per recorder writes gzipped CSV parts to the engine’s own disk (or a mounted stick). The ring is the drain’s slack.
• sink = "buffer": on a PLC-class device with little RAM and flash. The engine keeps the last buffer_s seconds in RAM and writes nothing. taktwerk record on an edge PC beside it pulls the rows over OPC UA and writes the same files a file sink writes.

Both sinks serve their ring to pulls, so a file sink can be pulled as well.

What a row holds

One row every every ticks, at the ticks divisible by every: row n holds cycle n · every. Skipped ticks are recorded too (a stale input skips a tick; the row then holds the held outputs and taktwerk.status is 2). So consecutive rows always differ by exactly every cycles; any other jump is a loss.

A row is what the cycle holds after the tick is stored: outputs, composed inputs, system signals, and inputs and tunables bound whole to an instance, an input or tunable as the instance stepping most often holds it.

Selecting signals

signals lists patterns over signal names, in column order; a signal matched twice is recorded once.

• * matches any run of characters, dots included; ? matches one character.
• A trailing [*] takes every element of a one-dimensional signal, [i] element i (from 0). Without a selector an array is recorded whole.
• An empty list (the default) records every output.
• Each array element is one column, name[i], or name[i,j] with logical indices for a matrix. A text signal is one column.
• A pattern that matches nothing recordable is an error. An external signal read only element by element (a source of a composed input) is not held whole and cannot be recorded; record the composed input <instance>.<input> instead.

Files of one run

<dir>/<project>/<recorder>/<run-id>/
  run.json
  part-0001.csv.gz
  part-0002.csv.gz
  events.csv

• <dir> is the recorder’s dir, relative to the project file or absolute; for taktwerk record it is --dir.
• <project> is the project file’s name without extension, <recorder> the recorder’s id.
• <run-id> is YYYYMMDDTHHMMSSZ-<hash>: the recorder host’s UTC when the run was opened and the first 8 hex digits of the project file’s SHA-256. A second run in the same second gets -2, then -3.

A run opens when the engine starts and closes when it stops; every start is a new run.

run.json

Written when the run opens, rewritten with end when it closes.

Device clocks are not trusted: rows carry the cycle and the engine time, and every UTC timestamp comes from the recorder host.

Parts

A part is a gzip file holding CSV: a header line, then one line per row.

cycle,t_engine,filt.y[0],filt.y[1],filt.y[2],filt.u[0],filt.u[1],filt.u[2],filt.gain,taktwerk.status
0,0.0,0.9090909090909091,1.9607843137254901,2.9702970297029703,0.0,0.0,0.0,1.0,1
1,0.01,0.8264462809917356,1.9223375624759707,2.9408881482207625,0.0,0.0,0.0,1.0,1

• cycle is the tick, from 0; t_engine is start_time + cycle · tick_s, the time the models see.
• Floats are printed shortest round-trip: 0.1, 1e-7, NaN, inf, -inf. Integers are decimal, booleans 0 or 1. Text is a CSV field, quoted when it holds a comma, quote or line break.
• The gzip header carries no file name and no time, and the stream is flushed every 10 s of engine time, never on wall time. The same rows give the same bytes.
• The open part of a running engine ends at its last flush; it is a complete gzip file only once closed.

events.csv

cycle,t_engine,kind,source,detail, one line per event, flushed about once a second.

A tunable’s start value is no change and logs nothing; a write of the same value neither.

Rotation and retention

A part closes, and the next one opens, when its compressed size reaches rotate.size_mb or it covers rotate.hours of engine time, whichever comes first. Rotation counts engine time, not wall time, so the same rows always split into the same parts.

Retention runs when a run opens and at every rotation, over all runs of one recorder (<dir>/<project>/<recorder>/). It deletes closed parts, oldest first, while the recorder’s parts together exceed retain.max_gb, and every closed part older than retain.days (by its modification time). The open part is never deleted; run.json and events.csv are kept.

Decimation and several recorders

every = 10 keeps one row per 10 ticks. A project may have several recorders, each with its own signals, every, sink and ring, for example a per-tick one for a few signals and a decimated one for everything. Their ids keep their directories, counters and pull nodes apart.

Counters

Each recorder publishes system signals under <prefix>.recorder.<id>. (<prefix> is engine.system_prefix, default taktwerk), about once a second, on the OPC UA server and in the TUI:

Sizing the ring

The ring holds buffer_s seconds of engine time: ceil(buffer_s / (tick_s · every)) rows, at least 2. A row takes one 8-byte word per numeric column, one per byte of a text column’s capacity, and two words of its own. taktwerk check prints both per recorder:

  recorder  sink    every  columns  ring rows  ring     dir
  main      file    1      8        60000      4.8 MB   records
  slow      buffer  10     3        600        24.0 kB  -

A ring larger than 4 GiB is refused. For a buffer sink, buffer_s is how long a puller may be away without a gap; for a file sink, how long the disk may stall without a drop.

Gaps and drops

Both show as a jump in the cycle numbers of a run, larger than every. They differ in where the rows were lost:

• Dropped: the engine’s own file sink fell more than a ring behind the cycle (a stalled or slow disk). Counted in rows_dropped, logged as a dropped event with the cycles lost. Raise buffer_s, record fewer signals or a larger every, or use a faster disk.
• Gap: a puller asked for rows the ring had already overwritten (the puller stopped or the network was down for longer than buffer_s). Counted in the engine’s gaps, logged as a gap event in the puller’s events.csv. Raise buffer_s on the engine.

A skipped tick is neither: its row is there, with taktwerk.status 2.

Reading a run

The standard library is enough: gzip and csv over every part in order, json for run.json. Every value arrives as a string; float() reads the float columns, NaN and inf included. This script is examples/recording/read_run.py:

"""Read one recorded run with the Python standard library alone.

Usage: python3 read_run.py <run-dir>
"""

import csv
import gzip
import json
import sys
from pathlib import Path


def rows(run: Path):
    """Yield every row of the run as a dict, parts in order, header lines skipped."""
    for part in sorted(run.glob("part-*.csv.gz")):
        with gzip.open(part, "rt", newline="") as f:
            try:
                yield from csv.DictReader(f)
            except EOFError:
                pass  # the open part of a running engine ends at its last flush


def main() -> None:
    run = Path(sys.argv[1])
    meta = json.loads((run / "run.json").read_text())
    every = meta["every"]
    n, gaps, last, prev = 0, 0, None, None
    for row in rows(run):
        cycle = int(row["cycle"])
        if prev is not None and cycle != prev + every:
            gaps += 1  # rows lost between prev and cycle
        prev, last, n = cycle, row, n + 1
    print(
        f"{meta['project']}/{meta['recorder']}: {n} rows, tick {meta['tick_s']} s, "
        f"every {every}, {gaps} gaps"
    )
    print("last row:", last)
    with open(run / "events.csv", newline="") as f:
        for e in csv.DictReader(f):
            print(f"event {e['cycle']} {e['kind']} {e['source']}: {e['detail']}")


if __name__ == "__main__":
    main()

Reproducible runs

taktwerk run --cycles N stops the engine after N ticks with a stop event cycle limit reached, at any speed; --speed max runs it unpaced and loses no row. Two runs of a deterministic project (no external inputs, no edits while it runs) give byte-identical parts and the same events.csv; run.json differs in run_id, engine_run and the UTC fields. taktwerk.overruns counts missed deadlines, which depend on the machine’s load; leave it out of signals when parts must match.

taktwerk run examples/recording/project.toml --cycles 200 --speed max --no-connectors

Remote recording

taktwerk record pulls one recorder of a running engine over OPC UA and writes the run files on the host it runs on, until SIGINT or SIGTERM. Run it on an edge PC beside a PLC-class device that records with sink = "buffer":

taktwerk record opc.tcp://10.0.0.2:4840 --dir /data/records --recorder main

• --recorder may be left out when the engine has only one; --namespace and --prefix match the server’s namespace and the engine’s system_prefix when they are not the defaults.
• It starts with the oldest row the engine still holds and opens the run at the first row it gets: source in run.json is the endpoint, start.utc the puller host’s clock.
• It pulls again at once while batches come back full, else after a tenth of the engine’s buffer_s (at most 1 s). It stays lossless while it is never away longer than buffer_s.
• A jump in the cycle numbers is written as a gap event, and so are engine events that left the engine’s log before they were pulled.
• When the engine stops, its server stays up (at most 5 s) until every recorder pulled during the run has its last rows and events; the puller writes the stop event, the rows up to the last tick, and closes the run.
• A lost connection is retried. When the engine restarted (engine_run changed), the run is closed and the next one opened.
• Its parts are written the same way as a local sink’s, so the same rows give the same bytes.

Pull interface

For integrators writing their own puller. The engine’s OPC UA server adds per recorder, next to the counters, in the signal namespace:

• <prefix>.recorder.<id>.manifest, a String variable: run.json without the run fields (run_id, host, source, start, end), as JSON.
• <prefix>.recorder.<id>.read_rows(from_cycle: UInt64, max_rows: UInt32) returns engine_run (String), oldest_cycle, next_cycle (UInt64) and rows (ByteString). Rows start at the first row at or after from_cycle still held, 0 meaning the oldest. Each row is its cycle and then one word per column word, all little-endian u64: a float’s bits (f32 in the low 32), an integer sign- or zero-extended, a bool 0 or 1, a text byte per word. A reply holds at most 8 MiB. Ask again from next_cycle; a jump in the cycles is a gap.
• <prefix>.recorder.<id>.read_events(from_seq: UInt64, max: UInt32) returns engine_run, oldest_seq, next_seq, then cycles (UInt64[]), kinds, sources and details (String[]), one entry per event, at most 4096 per call.

The methods are called on the recorder’s folder object, NodeId s=<prefix>.recorder.<id>/. A changed engine_run means the engine restarted and its cycles count from 0 again.