API Reference

adwin

CONTEXTS__SPECIAL = {'ADwin_LowInit': -2, 'ADwin_Init': -1, 'ADwin_Finish': 2 ** 31 - 1} module-attribute

Used for passing information to the ADwin controller

CYCLES__RUN = (0, 2 ** 31 - 2) module-attribute

The first and last cycle at which a row outside the special contexts can be played.

The real-time program's cyclecount is a 32-bit long that starts at 0 and is incremented once past the last row, so that row can be at most 2**31 - 2. Everything outside the range belongs to the sentinels of CONTEXTS__SPECIAL, which share the column with the time axis (D19). See validate.cycles for what each way out does.

DATA__FINISH__DIGITAL = 41 module-attribute

Where upload leaves the final state (B11): the number of analogue and digital rows in Par_15 and Par_16, the analogue rows as module, channel and digits in data_31..33, the digital ones as module, channel and value in data_41..43. The sequencer's finish: plays them, and the console reads them back to show what the apparatus holds after a run.

MODULES__DIGITAL__MAX = 16 module-attribute

The digital modules, which the sequencer's lowinit: programs as outputs: their number in Par_19, and the modules in data_44, at most digitalModulesMaxDim of them. upload writes them from the machine specifications, so a second digital module is programmed like the first, and its rows reach it through their own module number, as analogue rows do (D18, #133).

PAR__PROCESSDELAY__EXPECTED = 9 module-attribute

processdelayExpected in the sequencer: the Processdelay upload built the arrays for.

PAR__PROCESSDELAY__REPORTED = 14 module-attribute

processdelayReported in the sequencer: the Processdelay its event loop runs at, written at the end of init:. upload clears it, so a 0 after a run means that the loaded program does not report it. That is a program older than the check, which is refused rather than trusted. Where the two differ, the sequencer has refused to play the run (#128).

PAR__SEQUENCES__FINISHED = 18 module-attribute

sequencesFinished in the sequencer: how many sequences have finished since the machine was booted, incremented at the end of every finish:. The console uses it to tell a restart after a run, when the apparatus holds the run's final state, from a restart by hand.

PAR__SEQUENCE__OWNER = 17 module-attribute

sequenceOwner in the sequencer: the number of the process playing the arrays, from the start of its lowinit: to the end of its finish:, and 0 otherwise. The data arrays are shared by every process on the machine, so a write for one process can land under a run of another; this says which one to wait for. Read as a name, not as proof: whether the named process is still running is asked of the process itself, so that a value left behind by one that did not finish cannot hold anything up.

PROCESSDELAY__RATE = {'T12': 1000000000.0} module-attribute

How fast Processdelay counts on each processor type, in units per second, keyed by what ADwin.ADwin.Processor_Type() returns. A process's cycle period is its Processdelay divided by this.

The one hardware constant the package keeps, and only for the processors it has run on: both laboratories use the T12, at 1 ns per unit. Another type -- including "T12.1", which the driver reports separately -- is refused by name rather than guessed at, and belongs here only once its rate is confirmed. Kept as a rate rather than a tick so that 5000 / 1e9 gives exactly the float 5e-6.

ROWS__MAX = {'analogue': 10000000, 'digital': 10000, 'analogue__finish': 256, 'digital__finish': 256} module-attribute

How many rows each array of the sequencer holds: analogMaxArrayDim, digitalMaxArrayDim and finishMaxArrayDim in both .bas programs, which must agree. upload refuses more rather than letting the transfer fail partway, with some arrays written and others not.

adc

Recording with the ADC variant of the sequencer, resources/ADwin/WignerTimeADwinADC.bas.

That program plays an upload exactly as WignerTimeADwin.bas does, and in addition records one channel of an ADC card in burst mode, over a window of the run. The window is stated here in seconds of the timeline, and written to the program in cycles of the run, at the period core.upload read off the machine. The program therefore assumes no period of its own. It used to convert seconds to cycles at a fixed 5 us, so at a period of 2 us, Lab2's, a window asked for at 1 s would have opened at 0.4 s (#94, step 10).

A window is armed for one run: the program's finish: disarms it. A run nobody armed records nothing, rather than a window left over from an earlier run, and read says so.

The contract with the program, whose #Defines must agree with the numbers below:

• Par_42 and Par_43, written by arm: the cycle at which the burst starts, and the cycle by which it is over. The program starts the burst only if 0 <= Par_42 < Par_43 <= Par_1, and its finish: resets Par_43 to 0, which is what disarms.
• FPar_61, written by arm: the window's duration in seconds. The program turns it into a number of samples at its own sample period, and caps it at its buffer.
• Par_41 and Par_44, reported by the program, and cleared by arm: how many samples it recorded, and their period in whole nanoseconds. Par_41 is 0 unless the burst had its whole window. The period is a Par rather than an FPar because an FPar reaches Python in single precision, which over two million samples would shift the last one by some 30 ns.
• Data_1: the samples, as the card's digits.

FPar_62, the start in seconds, is no longer read.

PROCESS = 4 module-attribute

The process number in the header of WignerTimeADwinADC.bas.

Samples

Bases: NamedTuple

One recording: time in seconds of the timeline, one entry per sample, and digits as the card delivered them, with the window that was armed and the run that played it.

Source code in src/wignertime/adwin/adc.py

class Samples(NamedTuple):
    """
    One recording: `time` in seconds of the timeline, one entry per sample, and `digits` as
    the card delivered them, with the `window` that was armed and the `run` that played it.
    """

    time: np.ndarray
    digits: np.ndarray
    window: Window
    run: core.Run

Window

Bases: NamedTuple

What arm wrote: a window of the run of upload, in cycles of that run.

cycle__start is the start rounded to the nearest cycle, by the same rule as every row of the timeline, and time__start is that instant in seconds, which is when the first sample is taken. cycle__end is the first cycle by which the whole duration has passed.

Source code in src/wignertime/adwin/adc.py

class Window(NamedTuple):
    """
    What `arm` wrote: a window of the run of `upload`, in cycles of that run.

    `cycle__start` is the start rounded to the nearest cycle, by the same rule as every row
    of the timeline, and `time__start` is that instant in seconds, which is when the first
    sample is taken. `cycle__end` is the first cycle by which the whole `duration` has
    passed.
    """

    upload: core.Upload
    cycle__start: int
    cycle__end: int
    duration: float

    @property
    def time__start(self):
        """The start of the window as armed, in seconds of the timeline."""
        return self.cycle__start * self.upload.cycle_period

time__start property

The start of the window as armed, in seconds of the timeline.

arm(upload, t, duration)

Arms the next run of upload to record from t for duration seconds, and returns the Window as written.

upload must be one made for the ADC variant, process 4. The window must lie within the run: the samples are read out in finish:, and a burst still under way there would give a partly stale buffer, so a window that ends after the run is refused here. If process 4 is running, this waits for it first, since its finish: would disarm the new window.

Source code in src/wignertime/adwin/adc.py

def arm(upload, t, duration):
    """
    Arms the next run of `upload` to record from `t` for `duration` seconds, and returns the
    `Window` as written.

    `upload` must be one made for the ADC variant, process 4. The window must lie within the
    run: the samples are read out in `finish:`, and a burst still under way there would give
    a partly stale buffer, so a window that ends after the run is refused here. If process 4
    is running, this waits for it first, since its `finish:` would disarm the new window.
    """
    if upload.process != PROCESS:
        raise ValueError(
            "The ADC records only in `WignerTimeADwinADC.bas`, process {}, but this upload"
            " was made for process {}.".format(PROCESS, upload.process)
        )
    if not duration > 0:
        raise ValueError(
            "The duration of a recording must be a positive number of seconds, not"
            " {!r}.".format(duration)
        )
    if t < 0:
        raise ValueError(
            "A recording cannot start before the run does, at {} s.".format(t)
        )

    period = upload.cycle_period
    cycle__start = int(np.round(t / period))
    # Rounded to a millionth of a cycle before the ceiling, so that float noise cannot add a
    # whole cycle: 10e-6 / 2e-6 is 5.000000000000001.
    cycle__end = cycle__start + math.ceil(round(duration / period, 6))
    if cycle__end > upload.cycle__last:
        raise ValueError(
            "The recording from {:g} s for {:g} s ends after the run, which ends at {:g} s:"
            " the burst would still be under way when `finish:` reads it out. Shorten the"
            " recording, or extend the timeline.".format(t, duration, upload.time__last)
        )

    machine = upload.machine
    core._wait_for_the_arrays(
        machine,
        PROCESS,
        "arming its recording, which its `finish:` would disarm",
        "arming the recording of process {process}",
    )
    machine.Set_Par(PAR__CYCLE__START, cycle__start)
    machine.Set_Par(PAR__CYCLE__END, cycle__end)
    machine.Set_FPar(FPAR__DURATION, duration)
    machine.Set_Par(PAR__SAMPLES, 0)
    machine.Set_Par(PAR__SAMPLE_PERIOD__NS, 0)

    return Window(upload, cycle__start, cycle__end, duration)

read(window, run)

Reads out what the run recorded in the window armed for it, and returns the Samples.

Refuses a run of another upload, and one not yet waited for. Raises if the program recorded nothing, which it does when the run was not armed or did not last until the window closed, and if it recorded less than the window, which it does when the window is longer than its buffer. A program that reports no sample period is older than this contract, and is refused, since it would have placed the window by its own assumed period.

Source code in src/wignertime/adwin/adc.py

def read(window, run):
    """
    Reads out what the `run` recorded in the `window` armed for it, and returns the `Samples`.

    Refuses a run of another upload, and one not yet waited for. Raises if the program
    recorded nothing, which it does when the run was not armed or did not last until the
    window closed, and if it recorded less than the window, which it does when the window is
    longer than its buffer. A program that reports no sample period is older than this
    contract, and is refused, since it would have placed the window by its own assumed period.
    """
    if run.upload is not window.upload:
        raise ValueError(
            "This window was armed for another upload than the one this run played."
        )
    if run.duration is None:
        raise ValueError(
            "The run has not been waited for, so its recording may be incomplete. Read it"
            " after `core.wait`, or use `adc.run`."
        )

    machine = window.upload.machine
    sample_period = machine.Get_Par(PAR__SAMPLE_PERIOD__NS) * 1e-9
    if sample_period == 0:
        raise RuntimeError(
            "Process {} did not report its sample period, so the program loaded there is"
            " older than the one that takes its window in cycles, and it placed the window"
            " by an assumed period. Load the current `WignerTimeADwinADC.bas`.".format(
                PROCESS
            )
        )

    samples = machine.Get_Par(PAR__SAMPLES)
    if samples == 0:
        raise RuntimeError(
            "Process {} recorded nothing: either the run was not armed (a window is armed"
            " for one run only, so arm it again, or use `adc.run`), or the run ended before"
            " cycle {}, when the window closed.".format(PROCESS, window.cycle__end)
        )
    if samples * sample_period < window.duration - sample_period:
        raise RuntimeError(
            "Process {} recorded {} samples, {:g} s, of the {:g} s asked for: the window is"
            " longer than its buffer.".format(
                PROCESS, samples, samples * sample_period, window.duration
            )
        )

    # As the lab has always read it off the rig.
    digits = np.asarray(machine.GetData_Int64(DATA__SAMPLES, 1, samples))
    time = window.time__start + np.arange(samples) * sample_period
    return Samples(time, digits, window, run)

run(upload, t, duration)

Records from t for duration seconds in one run of upload, and returns the Samples: arm, then core.run, then read. Arms afresh each time, so it can be called repeatedly on the same upload, to average over runs.

Source code in src/wignertime/adwin/adc.py

def run(upload, t, duration):
    """
    Records from `t` for `duration` seconds in one run of `upload`, and returns the
    `Samples`: `arm`, then `core.run`, then `read`. Arms afresh each time, so it can be
    called repeatedly on the same upload, to average over runs.
    """
    window = arm(upload, t, duration)
    return read(window, core.run(upload))

connection

The choice made here is to model our ADwin connections as little more than tables of data (DataFrames). This allows for the least amount of coupling with other implementations an convenience of interaction with timelines etc.

What follows are simply convenience functions to make the creation easier.

new(*variable_module_channel)

Convenience for creating a table with 'variable', 'module' and 'channel' columns.

'variable's have the form 'context_equipment__unit' or 'context_equipment'. In the latter case, the 'variable' is taken to be digital (unitless).

vmcs: e.g. "AOM__MOT__V", 1, 1 or ["shutter__MOT", 1, 11], ["shutter__repump", 1, 12], ["shutter__imaging", 1, 13],

Source code in src/wignertime/adwin/connection.py

def new(*variable_module_channel) -> pd.DataFrame:
    """
    Convenience for creating a table with 'variable', 'module' and 'channel' columns.

    'variable's have the form 'context_equipment__unit' or 'context_equipment'. In the latter case, the 'variable' is taken to be digital (unitless).

    vmcs:
    e.g.
        "AOM__MOT__V", 1, 1
    or
        ["shutter__MOT", 1, 11],
        ["shutter__repump", 1, 12],
        ["shutter__imaging", 1, 13],
    """

    try:
        frame = wt_frame.new_schema(np.atleast_2d(variable_module_channel), _SCHEMA)
    except:
        raise ValueError("=== Input to 'connection' not well formatted ===")

    # NOTE: deliberately outside the `try`. A badly shaped table and a badly named
    # variable are different mistakes, and the second one already says which name
    # offends -- information a blanket re-raise would throw away.
    return _ensure_valid_names(frame)

remove_unconnected_variables(timeline, connections)

Purges the given timeline of any variables that do not have a matching connection.

NOTE: Assumes timeline and connections are both pd.DataFrame-like things

Source code in src/wignertime/adwin/connection.py

def remove_unconnected_variables(timeline, connections):
    """
    Purges the given timeline of any `variable`s that do not have a matching `connection`.

    NOTE: Assumes timeline and connections are both pd.DataFrame-like things
    """
    timeline = deepcopy(timeline)
    _disconnections = [
        v
        for v in timeline["variable"].unique()
        if v not in connections["variable"].unique()
    ]

    for v in _disconnections:
        timeline.drop(timeline[timeline.variable == v].index, inplace=True)

    return timeline

console

The manual console: direct access to the apparatus's channels from a notebook, for alignment and debugging, through resources/ADwin/WignerTimeConsole.bas loaded as process 10.

It works from the tables the timelines use. panel joins connections to devices, so a channel is named, converted and bounded in one place only (D22). An analogue value goes through its device's to_V, a function where it is one, and then through conversion.to_digits with its module's range, width and gain from the machine specifications, the conversion the sequencer's arrays are made with (D11). The console has no notion of a timeline beyond the one it takes its starting values from.

Shadow state, not a mailbox. configure writes the panel into the program as a list of entries, one per variable. set_value writes one word, the digits wanted for one entry, and returns without waiting. The program's sweep writes to the hardware wherever the wanted digits differ from the ones last written, so repeated values coalesce and nothing on either side waits on the other. The mailbox this replaced made Python poll Par_73 until the program had served the previous request, and both explanations kept for the fault in which the processor saturated for seconds at a time involved that loop (D22).

The contract with the program, whose #defines must agree with the numbers below:

• data_51[i] / data_52[i]: the digits wanted for entry i, written here, and the digits last written to the hardware, written by the program. -1 in data_52 means write again; -2 in either means unknown.
• data_53[i] / data_54[i]: the module and channel of entry i, in the panel's row order.
• data_56[i]: 1 where entry i is a digital line, which takes 0 or 1, and 0 where it is analogue. The program programs the module of every digital entry as outputs (D18).
• data_55[i]: 1 where the program has written entry i since it started, which is what makes its value one set on the console rather than one a run left there. jumps reads it.
• Par_75: the number of entries. configure sets it to 0 while it rewrites them.
• Par_78: a token configure draws, so that a panel configured again elsewhere makes the old one refuse, rather than write its values into another panel's entries.
• Par_74, Par_77: hardware writes and sweeps since the program started, for health.
• Par_76: the value of adwin.PAR__SEQUENCES__FINISHED the program last saw.

After a run. A sequence stops the console for its run, and starts it again afterwards if it was running. The program then adopts the run's final state (maintainer's policy, 2026-09-27): an entry the final state names takes its digits, and one it does not name becomes unknown, because it holds whatever the run left it at. Nothing is written to either, and anything wanted before or during the run is discarded. set_value refuses while a sequence owns the outputs. A start by hand, with no run since, writes every entry again from what was wanted.

Before a run. adwin.core.upload warns about each analogue channel set on the console that the run will jump from its value, whether in the initial state or at the first row that commands it (jumps). The MOT coils are the case in point: the lab leaves them out of the initial state, so that the MOT stays in its steady state between runs, and the jump from a value set by hand comes with the first stage that sets them.

ENTRIES__MAX = 512 module-attribute

consoleMaxEntries in the program: 16 modules of 32 channels.

PROCESS = 10 module-attribute

The process number in the header of WignerTimeConsole.bas.

Console

Bases: NamedTuple

A console configured with a panel: the machine, the panel table whose rows are its entries, and the token that configure drew for it.

Source code in src/wignertime/adwin/console.py

class Console(NamedTuple):
    """
    A console configured with a panel: the machine, the panel `table` whose rows are its
    entries, and the `token` that `configure` drew for it.
    """

    machine: object
    table: wt_frame.CLASS
    token: int

Health

Bases: NamedTuple

A snapshot of the machine as the console sees it. workload is ADwin's processor workload in percent, sweeps the program's heartbeat and writes its hardware writes since it started. They are for the fault reported in the lab, the processor saturating for seconds at a time while the console is in use: sampled over a session, they date an episode, and show whether the console was writing while it lasted.

Source code in src/wignertime/adwin/console.py

class Health(NamedTuple):
    """
    A snapshot of the machine as the console sees it. `workload` is ADwin's processor workload
    in percent, `sweeps` the program's heartbeat and `writes` its hardware writes since it
    started. They are for the fault reported in the lab, the processor saturating for seconds
    at a time while the console is in use: sampled over a session, they date an episode, and
    show whether the console was writing while it lasted.
    """

    workload: int
    console: int
    owner: int
    entries: int
    sweeps: int
    writes: int

Jump

Bases: NamedTuple

An analogue channel that a run will move, at once, from a value set on the console: the variable, the value it held, the value the run first commanded, both in its own unit, and the cycle at which the run does so, negative for the initial state.

Source code in src/wignertime/adwin/console.py

class Jump(NamedTuple):
    """
    An analogue channel that a run will move, at once, from a value set on the console: the
    `variable`, the value it `held`, the value the run first `commanded`, both in its own unit,
    and the `cycle` at which the run does so, negative for the initial state.
    """

    variable: str
    held: float
    commanded: float
    cycle: int

OutputsOwned

Bases: RuntimeError

A sequence owns the outputs, so a value is refused: after the run the apparatus holds the run's final state, and the console adopts it.

Source code in src/wignertime/adwin/console.py

class OutputsOwned(RuntimeError):
    """
    A sequence owns the outputs, so a value is refused: after the run the apparatus holds the
    run's final state, and the console adopts it.
    """

    def __init__(self, owner):
        self.owner = owner
        super().__init__(
            "A sequence, process {}, owns the outputs, so the value is refused. After the run"
            " the apparatus holds its final state, which the console adopts.".format(
                owner
            )
        )

configure(machine, table)

Writes the panel into the console program as its entries, each wanting its default_value, and returns the Console. The program writes them to the hardware, except the variables without a default, which are unknown until they are set.

The entries are rewritten with their count at 0, so that the program never sweeps a half-written list, and a fresh token is drawn. A run finished before this does not count as one the program has yet to adopt, since these values are newer.

Source code in src/wignertime/adwin/console.py

def configure(machine, table):
    """
    Writes the panel into the console program as its entries, each wanting its
    `default_value`, and returns the `Console`. The program writes them to the hardware, except
    the variables without a default, which are unknown until they are set.

    The entries are rewritten with their count at 0, so that the program never sweeps a
    half-written list, and a fresh token is drawn. A run finished before this does not count
    as one the program has yet to adopt, since these values are newer.
    """
    owner = machine.Get_Par(wt_adwin.PAR__SEQUENCE__OWNER)
    if owner:
        raise OutputsOwned(owner)
    if len(table) > ENTRIES__MAX:
        raise ValueError(
            "The console holds at most {} entries, and this panel has {}.".format(
                ENTRIES__MAX, len(table)
            )
        )

    wanted = [
        DIGITS__UNKNOWN if wt_frame.isnull(value) else _digits(table, name, value)[1]
        for name, value in zip(table["variable"], table["default_value"])
    ]
    count = len(table)
    machine.Set_Par(PAR__ENTRIES, 0)
    for number, values in (
        (DATA__MODULE, [int(m) for m in table["module"]]),
        (DATA__CHANNEL, [int(c) for c in table["channel"]]),
        (DATA__WANTED, wanted),
        (DATA__WRITTEN, [DIGITS__REASSERT] * count),
        (DATA__TOUCHED, [0] * count),
        (DATA__DIGITAL, [int(d) for d in table["digital"]]),
    ):
        machine.SetData_Long(values, number, 1, count)
    machine.Set_Par(
        PAR__SEQUENCES__SEEN, machine.Get_Par(wt_adwin.PAR__SEQUENCES__FINISHED)
    )
    token = random.randrange(1, 2**31 - 1)
    machine.Set_Par(PAR__TOKEN, token)
    machine.Set_Par(PAR__ENTRIES, count)
    return Console(machine, table, token)

create_UI(machine, table, continuous_update=True)

Configures the console with the panel and returns its widgets: a vertical slider per analogue variable, a toggle per digital line, a button that shows what the program holds, and a line for messages. Configuring writes every default to the apparatus; a variable without one is marked as unknown and left as it is.

A slider sends its value while it is dragged, since a write costs one driver call and repeated values coalesce in the program. continuous_update=False sends it on release.

A value that is refused, because a sequence owns the outputs or the console is not running, puts the widget back and says why, so that the widgets show what was asked for. After a run the widgets are brought up to the adopted final state at the next move, or with the button. A variable the final state does not name is marked, since what it holds is unknown until it is set.

Source code in src/wignertime/adwin/console.py

def create_UI(machine, table, continuous_update=True):
    """
    Configures the console with the `panel` and returns its widgets: a vertical slider per
    analogue variable, a toggle per digital line, a button that shows what the program holds,
    and a line for messages. Configuring writes every default to the apparatus; a variable
    without one is marked as unknown and left as it is.

    A slider sends its value while it is dragged, since a write costs one driver call and
    repeated values coalesce in the program. `continuous_update=False` sends it on release.

    A value that is refused, because a sequence owns the outputs or the console is not
    running, puts the widget back and says why, so that the widgets show what was asked for.
    After a run the widgets are brought up to the adopted final state at the next move, or
    with the button. A variable the final state does not name is marked, since what it
    holds is unknown until it is set.
    """
    if not importlib.util.find_spec("ipywidgets"):
        raise ImportError("The console's widgets require `ipywidgets` to be installed.")
    import ipywidgets as widgets

    console = configure(machine, table)
    messages = widgets.Output()
    quiet = {"on": False}  # set while widgets are moved from here rather than by hand
    adopted = {"seen": machine.Get_Par(PAR__SEQUENCES__SEEN)}
    controls = {}

    def mark(control, unknown):
        if isinstance(control, widgets.ToggleButton):
            control.button_style = "warning" if unknown else ""
        else:
            control.style.handle_color = "orange" if unknown else None

    def refresh(_=None):
        adopted["seen"] = machine.Get_Par(PAR__SEQUENCES__SEEN)
        held = readback(console)
        unknown = []
        quiet["on"] = True
        try:
            for name, value in zip(held["variable"], held["value"]):
                control = controls[name]
                if np.isnan(value):
                    unknown.append(name)
                else:
                    is_toggle = isinstance(control, widgets.ToggleButton)
                    control.value = bool(value) if is_toggle else value
                mark(control, np.isnan(value))
        finally:
            quiet["on"] = False
        messages.append_stdout(
            "Showing what the console holds. Unknown after the last run: {}.\n".format(
                ", ".join(unknown) or "nothing"
            )
        )

    def observe(control, name):
        def on_change(change):
            if quiet["on"]:
                return
            if machine.Get_Par(PAR__SEQUENCES__SEEN) != adopted["seen"]:
                refresh()
            try:
                set_value(console, name, change["new"])
            except RuntimeError as refused:
                quiet["on"] = True
                try:
                    control.value = change["old"]
                finally:
                    quiet["on"] = False
                messages.append_stdout("{}: {}\n".format(name, refused))
                return
            mark(control, False)

        control.observe(on_change, names="value")

    sliders, toggles = [], [[], [], []]
    for row in table.itertuples():
        unknown = wt_frame.isnull(row.default_value)
        if wt_frame.isnull(row.to_V):
            control = widgets.ToggleButton(
                value=False if unknown else bool(row.default_value),
                description=row.variable,
                layout=widgets.Layout(margin="20px 5px"),
            )
            group = {"shutter": 0, "AOM": 1}.get(row.variable.split("_", 1)[0], 2)
            toggles[group].append(control)
        else:
            control = widgets.FloatSlider(
                value=(
                    min(max(0.0, row.value__min), row.value__max)
                    if unknown
                    else row.default_value
                ),
                min=row.value__min,
                max=row.value__max,
                step=0.01,
                description="{}\n({})".format(
                    wt_variable.without_unit(row.variable),
                    wt_variable.unit(row.variable),
                ),
                orientation="vertical",
                continuous_update=continuous_update,
                layout=widgets.Layout(margin="20px 40px", height="300px"),
            )
            sliders.append(control)
        observe(control, row.variable)
        controls[row.variable] = control
        mark(control, unknown)

    button = widgets.Button(description="Refresh")
    button.on_click(refresh)

    return widgets.VBox(
        [widgets.HBox(sliders), *[widgets.HBox(t) for t in toggles], button, messages]
    )

final_state(machine, table)

The final state of the last upload, as {variable: value} for the variables of the panel it sets, read from the arrays upload writes for the sequencer's finish:.

Source code in src/wignertime/adwin/console.py

def final_state(machine, table):
    """
    The final state of the last upload, as `{variable: value}` for the variables of the
    `panel` it sets, read from the arrays `upload` writes for the sequencer's `finish:`.
    """
    by_port = {(row.module, row.channel): row for row in table.itertuples()}
    state = {}

    count = machine.Get_Par(wt_adwin.PAR__FINISH__ANALOGUE)
    if count:
        first = wt_adwin.DATA__FINISH__ANALOGUE
        modules, channels, digits = (
            np.asarray(machine.GetData_Long(first + i, 1, count)) for i in range(3)
        )
        for module, channel, digit in zip(modules, channels, digits):
            row = by_port.get((int(module), int(channel)))
            if row is not None:
                state[row.variable] = from_digits(row, digit)

    count = machine.Get_Par(wt_adwin.PAR__FINISH__DIGITAL)
    if count:
        first = wt_adwin.DATA__FINISH__DIGITAL
        modules, channels, values = (
            np.asarray(machine.GetData_Long(first + i, 1, count)) for i in range(3)
        )
        for module, channel, value in zip(modules, channels, values):
            row = by_port.get((int(module), int(channel)))
            if row is not None:
                state[row.variable] = int(value)

    return state

from_digits(row, digits)

The value of the panel's row whose digits are digits. The digits give the voltage exactly, as conversion.to_digits is linear. A linear to_V then gives the value; a calibration function, which cannot in general be inverted, is inverted by interpolation over the row's range, which assumes that it is monotonic there.

Source code in src/wignertime/adwin/console.py

def from_digits(row, digits):
    """
    The value of the panel's `row` whose digits are `digits`. The digits give the voltage
    exactly, as `conversion.to_digits` is linear. A linear `to_V` then gives the value; a
    calibration function, which cannot in general be inverted, is inverted by interpolation
    over the row's range, which assumes that it is monotonic there.
    """
    v_min, v_max = row.voltage__min, row.voltage__max
    step = (v_max - v_min) / (2 ** int(row.bits) - 1)
    voltage = v_min + digits * step
    # 0 V falls on the boundary between two codes, and is read back as 0 rather than as the
    # middle of the code it was rounded to, half a step away.
    if abs(voltage) <= step / 2:
        voltage = 0.0
    if not callable(row.to_V):
        return float(voltage / row.to_V)

    grid = np.linspace(row.value__min, row.value__max, 4097)
    voltages = np.vectorize(row.to_V)(grid)
    order = np.argsort(voltages)
    return float(np.interp(voltage, voltages[order], grid[order]))

health(machine)

The console's Health now.

Source code in src/wignertime/adwin/console.py

def health(machine):
    """The console's `Health` now."""
    return Health(
        workload=machine.Workload(),
        console=machine.Process_Status(PROCESS),
        owner=machine.Get_Par(wt_adwin.PAR__SEQUENCE__OWNER),
        entries=machine.Get_Par(PAR__ENTRIES),
        sweeps=machine.Get_Par(PAR__SWEEPS),
        writes=machine.Get_Par(PAR__WRITES),
    )

jumps(machine, analogue, connections, devices, machine_specifications=None)

The Jumps of a run whose analogue rows are analogue, as adwin.core.convert gives them: each analogue channel that holds a value written by the console since it last started, and whose first value in the run differs from it by more than one DAC step.

Only values the console wrote count. After a run the console holds the run's final state, and a jump from that is part of the sequence, not a surprise. Nothing is reported when the console has not yet seen the last run that finished, because its record is then older than what the apparatus holds.

Source code in src/wignertime/adwin/console.py

def jumps(machine, analogue, connections, devices, machine_specifications=None):
    """
    The `Jump`s of a run whose analogue rows are `analogue`, as `adwin.core.convert` gives them:
    each analogue channel that holds a value written by the console since it last started, and
    whose first value in the run differs from it by more than one DAC step.

    Only values the console wrote count. After a run the console holds the run's final state,
    and a jump from that is part of the sequence, not a surprise. Nothing is reported when the
    console has not yet seen the last run that finished, because its record is then older than
    what the apparatus holds.
    """
    count = machine.Get_Par(PAR__ENTRIES)
    seen = machine.Get_Par(PAR__SEQUENCES__SEEN)
    if not count or seen != machine.Get_Par(wt_adwin.PAR__SEQUENCES__FINISHED):
        return []

    first = {}
    for cycle, module, channel, digits in analogue:
        if cycle != wt_adwin.CONTEXTS__SPECIAL["ADwin_Finish"]:
            first.setdefault((int(module), int(channel)), (int(cycle), int(digits)))

    table = _conversions(wt_frame.join(connections, devices), machine_specifications)
    by_port = {(int(r.module), int(r.channel)): r for r in table.itertuples()}
    modules, channels, written, touched = (
        np.asarray(machine.GetData_Long(number, 1, count))
        for number in (DATA__MODULE, DATA__CHANNEL, DATA__WRITTEN, DATA__TOUCHED)
    )

    found = []
    for module, channel, held, mark in zip(modules, channels, written, touched):
        port = (int(module), int(channel))
        if not mark or held < 0 or port not in first or port not in by_port:
            continue
        cycle, digits = first[port]
        if abs(digits - held) > 1:
            row = by_port[port]
            found.append(
                Jump(
                    row.variable,
                    from_digits(row, held),
                    from_digits(row, digits),
                    cycle,
                )
            )
    return found

panel(connections, devices, timeline__defaults, machine_specifications=None)

The table the console works from: connections joined to devices, with a default_value per variable, the last value timeline__defaults gives it (normally the lab's experiment.init()), and what _conversions adds from machine_specifications (adwin.internal.SPECIFICATIONS__DEFAULT when not given), as the pipeline reads them.

The two tables must describe the same apparatus (device.check_correspondence). Here that matters more than anywhere: a variable with no device is taken for a digital line, so an analogue channel whose device name was mistyped would be switched between digits 0 and 1, that is, to -10 V. Every analogue variable needs finite bounds, which are its slider's range.

A variable the defaults do not mention keeps NaN as its default_value, is reported, and is left as it is: configure writes nothing to it until it is set. The lab leaves the MOT coils and the dispenser out of its initial state so that the MOT stays in its steady state between runs, and building the console should not undo that.

Source code in src/wignertime/adwin/console.py

def panel(connections, devices, timeline__defaults, machine_specifications=None):
    """
    The table the console works from: `connections` joined to `devices`, with a
    `default_value` per variable, the last value `timeline__defaults` gives it (normally the
    lab's `experiment.init()`), and what `_conversions` adds from `machine_specifications`
    (`adwin.internal.SPECIFICATIONS__DEFAULT` when not given), as the pipeline reads them.

    The two tables must describe the same apparatus (`device.check_correspondence`). Here
    that matters more than anywhere: a variable with no device is taken for a digital line,
    so an analogue channel whose device name was mistyped would be switched between digits
    0 and 1, that is, to -10 V. Every analogue variable needs finite bounds, which are its
    slider's range.

    A variable the defaults do not mention keeps NaN as its `default_value`, is reported, and is
    left as it is: `configure` writes nothing to it until it is set. The lab leaves the MOT coils
    and the dispenser out of its initial state so that the MOT stays in its steady state between
    runs, and building the console should not undo that.
    """
    device.check_correspondence(connections, devices)
    table = _conversions(wt_frame.join(connections, devices), machine_specifications)

    analogue = table["to_V"].notna()
    unbounded = table.loc[
        analogue
        & ~(np.isfinite(table["value__min"]) & np.isfinite(table["value__max"])),
        "variable",
    ]
    if len(unbounded):
        raise ValueError(
            "The console bounds each analogue slider by its device's `value__min` and"
            " `value__max`, and these have none, or an infinite one: {}.".format(
                sorted(unbounded)
            )
        )

    defaults = (
        wt_frame.sort(timeline__defaults, "time")  # stably, so the last written wins
        .groupby("variable")["value"]
        .last()
        .rename("default_value")
        .reset_index()
    )
    table = wt_frame.join(table, defaults)

    missing = sorted(table.loc[table["default_value"].isna(), "variable"])
    if missing:
        print(
            "console: the defaults give no value for {}; the console leaves them as they are"
            " until they are set.".format(missing)
        )

    return table

readback(console)

What the console program holds for each variable of the panel: a frame of variable, value (NaN where unknown), and pending, true where the value wanted has not yet been written to the hardware. That should last a sweep or two at most.

After a run, this is the final state the program adopted, and NaN for the variables the final state does not name.

Source code in src/wignertime/adwin/console.py

def readback(console):
    """
    What the console program holds for each variable of the panel: a frame of `variable`,
    `value` (NaN where unknown), and `pending`, true where the value wanted has not yet been
    written to the hardware. That should last a sweep or two at most.

    After a run, this is the final state the program adopted, and NaN for the variables the
    final state does not name.
    """
    machine, table = console.machine, console.table
    count = len(table)
    wanted = np.asarray(machine.GetData_Long(DATA__WANTED, 1, count))
    written = np.asarray(machine.GetData_Long(DATA__WRITTEN, 1, count))

    values = []
    for row, digits in zip(table.itertuples(), wanted):
        if digits == DIGITS__UNKNOWN:
            values.append(np.nan)
        elif wt_frame.isnull(row.to_V):
            values.append(float(digits))
        else:
            values.append(from_digits(row, digits))

    return wt_frame.new(
        {
            "variable": list(table["variable"]),
            "value": values,
            "pending": list((wanted >= 0) & (wanted != written)),
        }
    )

set_value(console, name, value)

Sets the variable name to value, in its own unit, and returns without waiting: the program writes it to the hardware within a sweep or two.

Refuses while a sequence owns the outputs (OutputsOwned), when the console process is not running, and when the console has been configured again since console was, from this notebook or another, since its entries are then another panel's.

Source code in src/wignertime/adwin/console.py

def set_value(console, name, value):
    """
    Sets the variable `name` to `value`, in its own unit, and returns without waiting: the
    program writes it to the hardware within a sweep or two.

    Refuses while a sequence owns the outputs (`OutputsOwned`), when the console process is
    not running, and when the console has been configured again since `console` was, from this
    notebook or another, since its entries are then another panel's.
    """
    machine = console.machine
    owner = machine.Get_Par(wt_adwin.PAR__SEQUENCE__OWNER)
    if owner:
        raise OutputsOwned(owner)
    if machine.Get_Par(PAR__TOKEN) != console.token:
        raise RuntimeError(
            "The console has been configured again since this panel was, so its entries are"
            " another panel's. Configure it again, or rebuild the widgets."
        )
    if machine.Process_Status(PROCESS) == 0:
        raise RuntimeError(
            "The console, process {}, is not running, so nothing would write the value. Load"
            " `WignerTimeConsole.bas` and start it.".format(PROCESS)
        )

    index, digits = _digits(console.table, name, value)
    machine.SetData_Long([digits], DATA__WANTED, index + 1, 1)

to_digits(row, value)

The digits for value of the panel's analogue row, as the sequencer gets them.

Source code in src/wignertime/adwin/console.py

def to_digits(row, value):
    """The digits for `value` of the panel's analogue `row`, as the sequencer gets them."""
    voltage = row.to_V(value) if callable(row.to_V) else value * row.to_V
    return int(
        conversion.to_digits(
            voltage,
            voltage_range=[row.voltage__min, row.voltage__max],
            num_bits=int(row.bits),
        )
    )

core

POLL__PERIOD = 0.1 module-attribute

How often a process is asked whether it has stopped, in seconds: the lab's own rate.

LostEvents

Bases: RuntimeError

A run lost events: ADwin's term for event cycles that came due while the previous one was still executing.

Nothing is skipped when that happens -- cyclecount advances once per executed event, so every row is still played, in order, at its cycle -- but the cycles themselves stretch, and the run takes longer than its timeline says by run.slip. The shot is not the experiment that was described, so it is refused rather than reported. run carries the record.

Source code in src/wignertime/adwin/core.py

class LostEvents(RuntimeError):
    """
    A run lost events: ADwin's term for event cycles that came due while the previous one
    was still executing.

    Nothing is skipped when that happens -- `cyclecount` advances once per *executed*
    event, so every row is still played, in order, at its cycle -- but the cycles
    themselves stretch, and the run takes longer than its timeline says by `run.slip`.
    The shot is not the experiment that was described, so it is refused rather than
    reported. `run` carries the record.
    """

    def __init__(self, run):
        self.run = run
        super().__init__(
            "Process {} lost {} events during the run, so its sequence ran {:.1f} us"
            " longer than described: some cycle had more rows to play than one cycle"
            " period allows.".format(
                run.upload.process, run.lost_events, run.slip * 1e6
            )
        )

PeriodRefused

Bases: RuntimeError

The sequencer refused to play a run, because the Processdelay its event loop runs at is not the one the arrays were built for (#128). It played the initial state and nothing after.

upload reads the period off the machine, but before the start. A program may still set its own Processdelay once started, and a different process from the one the upload was made for may have been started. Either way, playing the arrays would rescale every time in them.

Source code in src/wignertime/adwin/core.py

class PeriodRefused(RuntimeError):
    """
    The sequencer refused to play a run, because the Processdelay its event loop runs at is not
    the one the arrays were built for (#128). It played the initial state and nothing after.

    `upload` reads the period off the machine, but before the start. A program may still set its
    own Processdelay once started, and a different process from the one the upload was made for
    may have been started. Either way, playing the arrays would rescale every time in them.
    """

    def __init__(self, run):
        self.run = run
        rate = wt_adwin.PROCESSDELAY__RATE[run.upload.processor]
        super().__init__(
            "Process {} refused to play the run: it runs at a Processdelay of {} ({:g} us),"
            " but the arrays were built for {} ({:g} us). Nothing after the initial state"
            " was played.".format(
                run.upload.process,
                run.processdelay__reported,
                run.processdelay__reported / rate * 1e6,
                run.upload.processdelay,
                run.upload.cycle_period * 1e6,
            )
        )

Run dataclass

The record of one run of an upload, filled in as it goes: start sets the first two fields and wait the rest.

time__start is wall-clock time (time.time()), so the record can be filed with the data the shot produced; duration is how long the run took, as seen from here, in seconds; lost_events counts ADwin's lost events during it (see LostEvents); processdelay__reported is the Processdelay the sequencer reported running at (see PeriodRefused).

Source code in src/wignertime/adwin/core.py

@dataclasses.dataclass
class Run:
    """
    The record of one run of an upload, filled in as it goes: `start` sets the first two
    fields and `wait` the rest.

    `time__start` is wall-clock time (`time.time()`), so the record can be filed with the
    data the shot produced; `duration` is how long the run took, as seen from here, in
    seconds; `lost_events` counts ADwin's lost events during it (see `LostEvents`);
    `processdelay__reported` is the Processdelay the sequencer reported running at (see
    `PeriodRefused`).
    """

    upload: Upload
    time__start: float
    lost_events: int | None = None
    duration: float | None = None
    processdelay__reported: int | None = None

    @property
    def slip(self):
        """How much longer the run took than described, in seconds."""
        return self.lost_events * self.upload.cycle_period

slip property

How much longer the run took than described, in seconds.

Upload

Bases: NamedTuple

The log of one upload: where the timeline went, at what period, and what was written.

The first two fields are the machine and the process, in the order of the (machine, process) pair that routines starting the controller take -- the camera routines of sec:parameter_scan among them -- so an Upload can be passed wherever such a pair is expected.

cycle__last is what Par_1 was set to, the cycle of the last row outside the special contexts, and time__last is the same instant in seconds. analogue and digital are the rows played, (cycle, module, channel, digits); analogue__finish and digital__finish are the final state, (module, channel, digits), which the sequencer applies however the run ends.

Source code in src/wignertime/adwin/core.py

class Upload(NamedTuple):
    """
    The log of one `upload`: where the timeline went, at what period, and what was written.

    The first two fields are the machine and the process, in the order of the
    `(machine, process)` pair that routines starting the controller take -- the camera
    routines of `sec:parameter_scan` among them -- so an `Upload` can be passed wherever
    such a pair is expected.

    `cycle__last` is what `Par_1` was set to, the cycle of the last row outside the
    special contexts, and `time__last` is the same instant in seconds. `analogue` and
    `digital` are the rows played, `(cycle, module, channel, digits)`; `analogue__finish` and
    `digital__finish` are the final state, `(module, channel, digits)`, which the sequencer
    applies however the run ends.
    """

    machine: ADwin.ADwin
    process: int
    processor: str
    processdelay: int
    cycle_period: float
    cycle__last: int
    analogue: list
    digital: list
    analogue__finish: list
    digital__finish: list

    @property
    def time__last(self):
        return self.cycle__last * self.cycle_period

    def __repr__(self):
        # The arrays run to hundreds of thousands of rows, so they are counted, not shown.
        return (
            "Upload(process={}, processor={!r}, processdelay={}, cycle_period={!r},"
            " cycle__last={}, time__last={!r}, rows={} analogue + {} digital,"
            " finish={} analogue + {} digital)".format(
                self.process,
                self.processor,
                self.processdelay,
                self.cycle_period,
                self.cycle__last,
                self.time__last,
                len(self.analogue),
                len(self.digital),
                len(self.analogue__finish),
                len(self.digital__finish),
            )
        )

convert(timeline, connections, devices, cycle_period, machine_specifications=None, time_resolution=None)

Convenience for converting a Wigner timeline (DataFrame) to an ADbasic-compatible list of tuples.

This takes an operation-layer timeline, adds the columns necessary for an ADwin conversion, based on the supplied or default specifications, and then converts the relevant columns according to adwin.to_tuples, i.e. [[(cycle, module, channel, value), ...], [(cycle, module, channel, value), ...]].

cycle_period is the period of the controller's event loop, in seconds, and has no default: it belongs to the program loaded on the machine, not to the package, and the two laboratories this was written for run at 5 us and 2 us. A wrong value is a uniform rescaling of every time in the experiment, which reads as physics rather than as an error, so it is asked for rather than assumed.

time_resolution is the step at which the ramps that bind none are sampled, the cycle period when not given -- the finest step the hardware can act on. A coarser one thins them without moving anything in time, since cycles are computed from the times themselves. A ramp that binds its own keeps it, e.g. function=functools.partial(tanh, time_resolution=1e-4) (#65): this is a default, never an override (timeline.expand).

machine_specifications describes the installed modules, and is internal.SPECIFICATIONS__DEFAULT when not given, read at call time.

Source code in src/wignertime/adwin/core.py

def convert(
    timeline,
    connections,
    devices,
    cycle_period,
    machine_specifications=None,
    time_resolution=None,
) -> list[list[tuple]]:
    """
    Convenience for converting a Wigner timeline (DataFrame) to an ADbasic-compatible list of tuples.

    This takes an operation-layer timeline, adds the columns necessary for an ADwin conversion, based on the supplied or default specifications, and then converts the relevant columns according to `adwin.to_tuples`, i.e.  [[(cycle, module, channel, value), ...],
    [(cycle, module, channel, value), ...]].

    `cycle_period` is the period of the controller's event loop, in seconds, and has no
    default: it belongs to the program loaded on the machine, not to the package, and
    the two laboratories this was written for run at 5 us and 2 us. A wrong value is a
    uniform rescaling of every time in the experiment, which reads as physics rather
    than as an error, so it is asked for rather than assumed.

    `time_resolution` is the step at which the ramps that bind none are sampled, the
    cycle period when not given -- the finest step the hardware can act on. A coarser
    one thins them without moving anything in time, since cycles are computed from the
    times themselves. A ramp that binds its own keeps it, e.g.
    `function=functools.partial(tanh, time_resolution=1e-4)` (#65): this is a default,
    never an override (`timeline.expand`).

    `machine_specifications` describes the installed modules, and is
    `internal.SPECIFICATIONS__DEFAULT` when not given, read at call time.
    """
    if time_resolution is None:
        time_resolution = cycle_period

    machine_specifications = ad.specifications(machine_specifications)

    return funcy.compose(
        wt_validate.ascending,
        lambda tline: ad.to_tuples(
            tline,
            machine_specifications=machine_specifications,
        ),
        lambda tline: ad.add(
            tline,
            connections,
            devices,
            cycle_period,
            machine_specifications=machine_specifications,
        ),
        lambda tline: tl.expand(
            tline,
            time_resolution=time_resolution,
        ),
        lambda tline: connection.remove_unconnected_variables(tline, connections),
    )(timeline)

link_device(DeviceNo=1, raiseExceptions=1, useNumpyArrays=0)

A Wrapper around ADwin.ADwin. Returns a new (stateful) ADwin machine object that is digitally connected to a physical ADwin machine.

Source code in src/wignertime/adwin/core.py

def link_device(DeviceNo=1, raiseExceptions=1, useNumpyArrays=0):
    """
    A Wrapper around ADwin.ADwin. Returns a new (stateful) ADwin machine object that is digitally connected to a physical ADwin machine.
    """
    return ADwin.ADwin(
        DeviceNo=DeviceNo,
        raiseExceptions=raiseExceptions,
        useNumpyArrays=useNumpyArrays,
    )

read_cycle_period(machine, process)

The cycle period of process on machine, in seconds: its Processdelay, read off the machine, divided by the rate at which the processor counts it.

Read, never set. An ADbasic program can overwrite its own Processdelay, so a value written from here is not one the machine is bound to keep. For the same reason this shows the value before any the program sets for itself once started.

Source code in src/wignertime/adwin/core.py

def read_cycle_period(machine, process):
    """
    The cycle period of `process` on `machine`, in seconds: its `Processdelay`, read off the
    machine, divided by the rate at which the processor counts it.

    Read, never set. An ADbasic program can overwrite its own `Processdelay`, so a value
    written from here is not one the machine is bound to keep. For the same reason this
    shows the value before any the program sets for itself once started.
    """
    return _timing(machine, process)[2]

run(upload)

Runs an upload with nothing to collect alongside, and returns its Run: the plain case, as for a sequence whose results are read off the machine afterwards.

Source code in src/wignertime/adwin/core.py

def run(upload):
    """
    Runs an `upload` with nothing to collect alongside, and returns its `Run`: the plain
    case, as for a sequence whose results are read off the machine afterwards.
    """
    with running(upload) as record:
        pass
    return record

running(upload)

Brackets one run of an upload: starts it on entry, and on leaving waits for it to end and checks it (wait). The block is where whatever the run triggers is collected:

camera.arm()                          # before the start
with adwin.running(log) as run:
    frames = camera.capture(4)        # triggered by the timeline
# here the run is over, and lost no events

If the block raises, the run is still waited out before the error goes on, but not checked, and not stopped. A stop plays the final state since B11 was fixed, but that a stop from the PC reaches finish: is not yet confirmed on the rig.

Source code in src/wignertime/adwin/core.py

@contextlib.contextmanager
def running(upload):
    """
    Brackets one run of an `upload`: starts it on entry, and on leaving waits for it to end
    and checks it (`wait`). The block is where whatever the run triggers is collected:

        camera.arm()                          # before the start
        with adwin.running(log) as run:
            frames = camera.capture(4)        # triggered by the timeline
        # here the run is over, and lost no events

    If the block raises, the run is still waited out before the error goes on, but not
    checked, and not stopped. A stop plays the final state since B11 was fixed, but that a
    stop from the PC reaches `finish:` is not yet confirmed on the rig.
    """
    run = start(upload)
    try:
        yield run
    except BaseException:
        _wait_until_stopped(upload.machine, upload.process)
        raise
    wait(run)

start(upload)

Starts the process an upload was made for, and returns its Run.

If the process is still running -- a replay of the same upload, which does not go through upload and its wait -- this waits for it first, saying so once. Everything that must be ready before the sequence begins, such as a camera waiting for its trigger, has to be armed before this is called.

Source code in src/wignertime/adwin/core.py

def start(upload):
    """
    Starts the process an `upload` was made for, and returns its `Run`.

    If the process is still running -- a replay of the same upload, which does not go
    through `upload` and its wait -- this waits for it first, saying so once. Everything
    that must be ready before the sequence begins, such as a camera waiting for its
    trigger, has to be armed before this is called.
    """
    machine, process = upload.machine, upload.process
    _wait_for_the_arrays(
        machine,
        process,
        "starting it again",
        "starting process {process}, which would play the same arrays",
    )

    run = Run(upload=upload, time__start=time.time())
    machine.Start_Process(process)
    return run

upload(timeline, connections, devices, machine, process, machine_specifications=None, time_resolution=None)

Converts a timeline for process on machine and writes it into the machine's memory, ready for the process to be started. Returns the Upload log of what was written.

The cycle period is read off the machine (read_cycle_period), never given: a wrong one rescales every time in the experiment, which reads as physics rather than as an error. That is why the machine and the process are both required. The process is the one whose Processdelay sets the period, so it must be the one started afterwards; the arrays themselves are shared by every process on the machine.

machine_specifications and time_resolution are passed on to convert.

If the process is still running, upload converts first and then waits for it to stop before writing, saying so once. The machine accepts writes mid-run and the running sequence reads them, so writing at once would rewrite the arrays under a run that is still playing (A15/#151). A parameter scan that uploads shot N+1 while shot N plays its tail is the case in point. The process that owns the arrays (Par_17) is waited for too, when it is another one, such as the ADC variant.

Then it warns about each analogue channel set on the console that the run will jump from its value (console.jumps), once the previous run is over and the console has seen it. A jump from a value set by hand is often intended, so it is reported, not refused.

NOTE: Stateful. It writes Par_1..3 and the data arrays, and starts nothing.

Source code in src/wignertime/adwin/core.py

def upload(
    timeline,
    connections,
    devices,
    machine: ADwin.ADwin,
    process: int,
    machine_specifications=None,
    time_resolution=None,
) -> Upload:
    """
    Converts a timeline for `process` on `machine` and writes it into the machine's memory,
    ready for the process to be started. Returns the `Upload` log of what was written.

    The cycle period is read off the machine (`read_cycle_period`), never given: a wrong
    one rescales every time in the experiment, which reads as physics rather than as an
    error. That is why the machine and the process are both required. The process is the
    one whose `Processdelay` sets the period, so it must be the one started afterwards;
    the arrays themselves are shared by every process on the machine.

    `machine_specifications` and `time_resolution` are passed on to `convert`.

    If the process is still running, `upload` converts first and then waits for it to stop
    before writing, saying so once. The machine accepts writes mid-run and the running
    sequence reads them, so writing at once would rewrite the arrays under a run that is
    still playing (A15/#151). A parameter scan that uploads shot N+1 while shot N plays its
    tail is the case in point. The process that owns the arrays (`Par_17`) is waited for
    too, when it is another one, such as the ADC variant.

    Then it warns about each analogue channel set on the console that the run will jump from
    its value (`console.jumps`), once the previous run is over and the console has seen it. A
    jump from a value set by hand is often intended, so it is reported, not refused.

    NOTE: Stateful. It writes `Par_1..3` and the data arrays, and starts nothing.
    """
    processor, processdelay, cycle_period = _timing(machine, process)

    output = convert(
        timeline,
        connections,
        devices,
        cycle_period,
        machine_specifications=machine_specifications,
        time_resolution=time_resolution,
    )

    # Either set may legitimately be empty -- a digital-only apparatus has no analogue
    # updates -- so the two are concatenated rather than stacked. Stacking them assumed
    # both were non-empty and raised `IndexError: too many indices` on an empty one,
    # before any `Set_Par` had run, leaving the machine holding the whole of the previous
    # experiment (#73).
    cycles = np.concatenate(
        [np.array(rows)[:, 0] for rows in output if len(rows)]
        or [np.array([], dtype=int)]
    )

    # Finds the maximum cycle value, discounting special contexts
    cycles__run = cycles[~np.isin(cycles, list(wt_adwin.CONTEXTS__SPECIAL.values()))]
    if not len(cycles__run):
        raise ValueError(
            "There is nothing to run: no update falls outside the special contexts "
            "{}, so the experiment has no duration. Every variable may have been "
            "dropped for want of a `connection`, or the timeline may describe only "
            "initial and final states.".format(list(wt_adwin.CONTEXTS__SPECIAL))
        )
    cycle__last = int(cycles__run.max())

    # The final state leaves the playback arrays for arrays of its own, which `finish:`
    # plays from the start however the run ended (B11). Its rows need no cycle.
    analogue, analogue__finish = _split_finish(output[0])
    digital, digital__finish = _split_finish(output[1])
    rows = {
        "analogue": analogue,
        "digital": digital,
        "analogue__finish": analogue__finish,
        "digital__finish": digital__finish,
    }
    too_many = {k: len(v) for k, v in rows.items() if len(v) > wt_adwin.ROWS__MAX[k]}
    if too_many:
        raise ValueError(
            "The timeline does not fit the sequencer's arrays: {} rows against room for"
            " {}. Nothing was written.".format(
                too_many, {k: wt_adwin.ROWS__MAX[k] for k in too_many}
            )
        )

    # Only now, so that the conversion overlaps whatever is left of the previous run.
    _wait_for_the_arrays(
        machine,
        process,
        "uploading, so as not to rewrite the arrays it is playing",
        "uploading for process {process}, so as not to rewrite the arrays it is playing",
    )
    _warn_of_jumps(
        wt_console.jumps(
            machine, analogue, connections, devices, machine_specifications
        ),
        cycle_period,
    )

    # `endCC`, `analogArrayDim` and `digitalArrayDim` in `WignerTimeADwin.bas`: the event
    # loop ends after the last cycle, and the counts say how far into each array to read.
    # `analogFinishDim` and `digitalFinishDim` do the same for the final state.
    machine.Set_Par(1, cycle__last)
    machine.Set_Par(2, len(analogue))
    machine.Set_Par(3, len(digital))
    machine.Set_Par(wt_adwin.PAR__FINISH__ANALOGUE, len(analogue__finish))
    machine.Set_Par(wt_adwin.PAR__FINISH__DIGITAL, len(digital__finish))

    # The period check (#128): the sequencer compares its own Processdelay with this at the
    # end of `init:`, and plays nothing past the initial state if they differ. The report is
    # cleared so that a program which does not make one cannot pass off an old one.
    machine.Set_Par(wt_adwin.PAR__PROCESSDELAY__EXPECTED, processdelay)
    machine.Set_Par(wt_adwin.PAR__PROCESSDELAY__REPORTED, 0)

    # The digital modules, which `lowinit:` programs as outputs (D18).
    modules__digital = ad.modules__digital(ad.specifications(machine_specifications))
    if len(modules__digital) > wt_adwin.MODULES__DIGITAL__MAX:
        raise ValueError(
            "The machine specifications describe {} digital modules, and the sequencer"
            " programs at most {}.".format(
                len(modules__digital), wt_adwin.MODULES__DIGITAL__MAX
            )
        )
    machine.Set_Par(wt_adwin.PAR__MODULES__DIGITAL, len(modules__digital))
    if modules__digital:
        machine.SetData_Long(
            modules__digital,
            wt_adwin.DATA__MODULES__DIGITAL,
            1,
            len(modules__digital),
        )

    # `cycle, module, channel, digits` go to data_10..13 for the analogue set and
    # data_20..23 for the digital one. The final state is `module, channel, digits` in
    # data_31..33 and data_41..43: the same numbers plus 20, without the cycle.
    #
    # An empty set is communicated by its count alone, set just above: there is nothing
    # to write, and a zero-length transfer is not meaningful. The count is what stops the
    # real-time program reading the array -- which matters, because the arrays are never
    # cleared, so a previous and longer run's contents are still sitting in them (#8).
    for data__first, columns, rows__set in (
        (10, range(4), analogue),
        (20, range(4), digital),
        (wt_adwin.DATA__FINISH__ANALOGUE, range(3), analogue__finish),
        (wt_adwin.DATA__FINISH__DIGITAL, range(3), digital__finish),
    ):
        if rows__set:
            for number, column in enumerate(columns, start=data__first):
                machine.SetData_Long(
                    [row[column] for row in rows__set], number, 1, len(rows__set)
                )

    return Upload(
        machine=machine,
        process=process,
        processor=processor,
        processdelay=processdelay,
        cycle_period=cycle_period,
        cycle__last=cycle__last,
        analogue=analogue,
        digital=digital,
        analogue__finish=analogue__finish,
        digital__finish=digital__finish,
    )

wait(run)

Waits for a Run to end, fills in its record, and returns it.

Raises PeriodRefused if the sequencer refused to play the run, and refuses a program that did not report its Processdelay at all: one older than the check, whose run nothing vouches for. Then raises LostEvents if the run lost any.

The count is ADwin's counter as the run ends, which the ADbasic manual (6.00, p. 79) describes as the cycles lost "since process start": the run's own. UNVERIFIED on the rig. If the counter instead accumulated from the moment the program was loaded, every run after one that lost events would be refused as well, which is loud. The difference across the run, read before this, was right only under that reading, and would have passed, without a word, a run that lost as many events as the one before it.

Source code in src/wignertime/adwin/core.py

def wait(run):
    """
    Waits for a `Run` to end, fills in its record, and returns it.

    Raises `PeriodRefused` if the sequencer refused to play the run, and refuses a program that
    did not report its Processdelay at all: one older than the check, whose run nothing vouches
    for. Then raises `LostEvents` if the run lost any.

    The count is ADwin's counter as the run ends, which the ADbasic manual (6.00, p. 79)
    describes as the cycles lost "since process start": the run's own. UNVERIFIED on the rig. If the counter
    instead accumulated from the moment the program was loaded, every run after one that lost
    events would be refused as well, which is loud. The difference across the run, read
    before this, was right only under that reading, and would have passed, without a word, a
    run that lost as many events as the one before it.
    """
    machine, process = run.upload.machine, run.upload.process
    _wait_until_stopped(machine, process)

    run.duration = time.time() - run.time__start

    run.processdelay__reported = machine.Get_Par(wt_adwin.PAR__PROCESSDELAY__REPORTED)
    if run.processdelay__reported == 0:
        raise RuntimeError(
            "Process {} did not report the Processdelay it ran at, so the program loaded"
            " there is older than the period check (#128), and nothing confirms that the run"
            " kept the timeline's times. Load the current `WignerTimeADwin.bas`, or"
            " `WignerTimeADwinADC.bas` as process 4.".format(process)
        )
    if run.processdelay__reported != run.upload.processdelay:
        raise PeriodRefused(run)

    run.lost_events = machine.Get_Lost_Events(process)
    if run.lost_events:
        raise LostEvents(run)

    return run

display

MARGIN = 0.05 module-attribute

The width of each margin, as a fraction of the run's duration.

quantities(timeline, variables=None, do_context=True, do_show=True, symbol_quantities=SYMBOL_QUANTITY, cmap__context='magma', range__x=None)

Displays the given timeline, filtered by variable, in terms of different quantites, i.e. by common unit. The mapping between unit and 'quantity' can be provided as a dictionary.

NOTE: Unit and quantity terminology taken from SI conventions.

Source code in src/wignertime/adwin/display.py

def quantities(
    timeline: wt_frame.CLASS,
    variables=None,
    do_context: bool = True,
    do_show: bool = True,
    symbol_quantities: dict = SYMBOL_QUANTITY,
    cmap__context="magma",
    range__x=None,
):
    """
    Displays the given `timeline`, filtered by `variable`, in terms of different quantites, i.e. by common `unit`. The mapping between `unit` and 'quantity' can be provided as a dictionary.

    NOTE: Unit and quantity terminology taken from SI conventions.
    """
    if variables:
        tline = wt_frame.subframe(timeline, "variable", variables)
    else:
        tline = timeline
    # A sorted copy, and a stable one (#153). This sorted the caller's own frame in place,
    # and not stably, so a value superseded at the same instant could be drawn as the one
    # in force.
    tline = wt_frame.sort(tline, "time")
    tline, margin = _into_margins(tline)

    # =====================================================================
    # ADwin
    # =====================================================================
    # To make the special contexts (where there is no time) visible, each spans its
    # margin: before the run for those whose sentinel is negative, after it otherwise.
    if do_context:
        info__context = tl.context_info(tline)
        for label, sentinel in adwin.CONTEXTS__SPECIAL.items():
            if label in info__context.keys():
                d = deepcopy(info__context[label]["times"])
                if sentinel < 0:
                    info__context[label]["times"] = [d[0], d[1] + margin]
                else:
                    info__context[label]["times"] = [d[0] - margin, d[1]]
    # =====================================================================

    if variables is None:
        variables = tline["variable"].unique()

    if variables is None:
        return None

    units = wt_variable.units(tline)
    unit_variables__analog = {
        u: [v for v in variables if wt_variable.unit(v) == u]
        for u in units
        if u not in ["digital", LABEL__ANCHOR]
    }
    variables__digital = [v for v in variables if wt_variable.unit(v) == "digital"]

    tline__anchors = tline[anchor.mask(tline)]

    prop_cycle = plt.rcParams["axes.prop_cycle"]
    colors = prop_cycle.by_key()["color"]

    num_analog_panels = len(unit_variables__analog)
    num_digital_panels = 1 if variables__digital else 0
    num_panels = num_analog_panels + num_digital_panels

    fig, axes = plt.subplots(
        num_analog_panels + num_digital_panels,
        sharex=True,
        figsize=(7.5, 7.5),
        height_ratios=[1] * num_analog_panels + num_digital_panels * [2],
    )
    if num_panels == 1:
        axes = [axes]
    fig.tight_layout()

    if range__x is not None:
        axes[0].set_xlim(*range__x)
    # =====================================================================
    # ANALOGUE
    # =====================================================================

    analogLabels = []
    if num_analog_panels > 0:
        axes__analogue = axes[:-num_digital_panels] if num_digital_panels != 0 else axes
        for key, axis in zip(unit_variables__analog.keys(), axes__analogue):
            if key in symbol_quantities:
                axis.set_ylabel(f"{symbol_quantities[key]} [{key}]")
            else:
                axis.set_ylabel(f"[{key}]")

            for variable, color in zip(unit_variables__analog[key], colors):
                array = tline[tline["variable"] == variable]
                # axis.plot(array["time"], array["value"], marker="o", ms=3)
                axis.step(
                    array["time"],
                    array["value"],
                    where="post",
                    # color=color,
                    marker="o",
                    ms=3,
                )

                analogLabels.append(
                    axis.text(0, array.iat[0, 2], variable, color=color)
                )
            if do_context:
                _draw_context(axis, info__context, cmap__context=cmap__context)
    #
    # =====================================================================
    # DIGITAL
    # =====================================================================
    digitalLabels = []
    if num_digital_panels > 0:
        divider = 1.5 * len(variables__digital)
        axes[-1].set_ylabel("Digital")

        for variable, offset, color in zip(
            variables__digital, range(len(list(variables__digital))), colors
        ):
            baseline = offset / divider
            array = tline[tline["variable"] == variable]
            axes[-1].axhline(baseline, color=color, linestyle=":", alpha=0.5)
            axes[-1].axhline(baseline + 1, color=color, linestyle=":", alpha=0.5)
            axes[-1].step(
                array["time"],
                array["value"] + baseline,
                where="post",
                color=color,
                marker="o",
                ms=3,
            )
            digitalLabels.append(
                axes[-1].text(0, baseline, variable + "_OFF", color=color)
            )
            digitalLabels.append(
                axes[-1].text(0, baseline + 1, variable + "_ON", color=color)
            )
        axes[-1].set_yticks([i / divider for i in range(len(list(variables__digital)))])
        axes[-1].set_yticklabels([])
        if do_context:
            _draw_context(axes[-1], info__context, cmap__context=cmap__context)

    for anchorTime in tline__anchors["time"]:
        for axis in axes:
            axis.axvline(anchorTime, color="0.5", linestyle="--")

    ax2 = axes[0].twiny()
    ax2.set_xlim(axes[0].get_xlim())
    ax2.set_xticks(list(tline__anchors["time"]))  # Set ticks at the specified x-values
    ax2.set_xticklabels(list(tline__anchors["context"]))

    def _sync_axes(event):
        xlim = axes[0].get_xlim()
        ax2.set_xlim(xlim)
        for label in analogLabels + digitalLabels:
            label.set_position((0.9 * xlim[0] + 0.1 * xlim[1], label.get_position()[1]))

    # Connect the sync function to the 'xlim_changed' event
    axes[0].callbacks.connect("xlim_changed", _sync_axes)

    if do_show:
        plt.show()

    return fig, axes

internal

For 'lower-level' manipulations of ADwin-specifc timeline informaton.

In general, the user shouldn't need to use these functions and there is no guarantee that the API will not change.

add(timeline, connections, devices, cycle_period, machine_specifications=None)

Takes an 'operational' layer timeline and inserts ADwin-specific columns, e.g. cycles and numbers for the module and channel etc.

Which modules are digital, and how each analogue module converts to digits, are read from machine_specifications (D11, D18).

cycle_period is the period of the controller's event loop, in seconds; see core.convert.

Source code in src/wignertime/adwin/internal.py

def add(timeline, connections, devices, cycle_period, machine_specifications=None):
    """
    Takes an 'operational' layer timeline and inserts ADwin-specific columns, e.g. cycles and numbers for the module and channel etc.

    Which modules are digital, and how each analogue module converts to digits, are read
    from `machine_specifications` (D11, D18).

    `cycle_period` is the period of the controller's event loop, in seconds; see
    `core.convert`.
    """

    wtl.debug("Got to `adwin.core.add`")
    machine_specifications = specifications(machine_specifications)

    # Here because this is where the two tables meet, and because conversion is the one
    # point at which hardware enters. Neither `connection.new` nor `device.new` can do it
    # alone: each sees only its own vocabulary (A14).
    device.check_correspondence(connections, devices)
    check_modules_described(connections, machine_specifications)
    check_module_kinds(connections, modules__digital(machine_specifications))

    dff = wt_frame.join(timeline, connections)
    dff = wt_frame.join(dff, devices)
    # Stably (A18): among a variable's rows at one instant the last written is in effect,
    # and `drop_duplicates` keeps the last of each cycle. The quicksort this used could
    # put a ramp's end after the `update` superseding it, and send the ramp's end.
    dff = wt_frame.sort(dff, "time")

    mask__digital = dff["module"].isin(modules__digital(machine_specifications))

    # Each analogue module with its own range, width and gain (D11).
    dff["value__digits"] = np.nan
    for module in sorted(set(dff.loc[~mask__digital, "module"])):
        mask = dff["module"] == module
        dff.loc[mask, "value__digits"] = conv.add(
            dff.loc[mask],
            specifications=conversion__module(machine_specifications, int(module)),
        )["value__digits"].to_numpy()

    dff.loc[mask__digital, "value__digits"] = round(dff["value"])

    device.check_within_range(dff)
    dcycle = add_cycle(dff, cycle_period)

    return wt_validate.all(dcycle)

add_cycle(timeline, cycle_period, special_contexts=None)

Inserts a new cycle column into the timeline as a conversion of the time column into 'number of cycles'.

Parameters: - timeline: DataFrame containing the experimental data. - cycle_period: The period of the controller's event loop, in seconds. - special_contexts: Dictionary with context-specific overrides for cycle values; wt_adwin.CONTEXTS__SPECIAL when not given.

The column is 64-bit here, although the machine's counter is 32-bit: a row too late for the counter has to survive long enough for validate.cycles to refuse it, and casting first would wrap it into an ordinary-looking number instead. validate.types narrows it once that check has run.

Raises: - ValueError if the time column is missing, or the cycle period is not a positive number of seconds.

Source code in src/wignertime/adwin/internal.py

def add_cycle(timeline, cycle_period, special_contexts=None):
    """
    Inserts a new `cycle` column into the timeline as a conversion of the `time` column into 'number of cycles'.

    Parameters:
    - timeline: DataFrame containing the experimental data.
    - cycle_period: The period of the controller's event loop, in seconds.
    - special_contexts: Dictionary with context-specific overrides for cycle values;
      `wt_adwin.CONTEXTS__SPECIAL` when not given.

    The column is 64-bit here, although the machine's counter is 32-bit: a row too late
    for the counter has to survive long enough for `validate.cycles` to refuse it, and
    casting first would wrap it into an ordinary-looking number instead. `validate.types`
    narrows it once that check has run.

    Raises:
    - ValueError if the `time` column is missing, or the cycle period is not a positive
      number of seconds.
    """
    if special_contexts is None:
        special_contexts = wt_adwin.CONTEXTS__SPECIAL

    if "time" not in timeline.columns:
        raise ValueError(
            f"`time` column not found. Columns present: {list(timeline.columns)}"
        )

    if not cycle_period > 0:
        raise ValueError(
            "`cycle_period` must be a positive number of seconds, not {!r}.".format(
                cycle_period
            )
        )

    # Rows before the run are at −∞ and rows after it at +∞ (#154), in the special
    # contexts, whose cycles are the sentinels below. Anywhere else a time that is not
    # finite has no cycle, and casting it gave an arbitrary integer with a numpy warning.
    times = timeline["time"].to_numpy(dtype=float)
    mask__finite = np.isfinite(times)
    mask__special = timeline["context"].isin(list(special_contexts)).to_numpy()
    if (~mask__finite & ~mask__special).any():
        raise ValueError(
            "Rows outside the special contexts {} must be at an instant of the run; ±∞"
            " is before or after it (#154). Offending rows:\n{}".format(
                list(special_contexts),
                timeline.loc[
                    ~mask__finite & ~mask__special, ["variable", "time", "context"]
                ],
            )
        )

    cycles = np.zeros(len(times), dtype=np.int64)
    cycles[mask__finite] = np.round(times[mask__finite] / cycle_period)
    timeline["cycle"] = cycles

    # Apply special context cycles
    timeline = wt_frame.replace_column__filtered(
        timeline,
        special_contexts,
        column__change="cycle",
    )

    return timeline

check_module_kinds(connections, modules__digital)

Checks that each connected variable is of the kind of the module it is connected to.

A variable's kind is stated twice: by its name, since a variable without a __unit suffix is a digital line, and by its module, whose width says what the port is. Nothing checked that the two agree (A16). An analogue variable on the digital module was rounded to an integer and switched as a digital line: 1.5 A on a coil became a 2 written to a digital output, without a word. A digital line on an analogue module did fail, but later, in a cast, with a message that named neither the variable nor the cause.

Source code in src/wignertime/adwin/internal.py

def check_module_kinds(connections, modules__digital):
    """
    Checks that each connected variable is of the kind of the module it is connected to.

    A variable's kind is stated twice: by its name, since a variable without a `__unit`
    suffix is a digital line, and by its module, whose width says what the port is. Nothing
    checked that the two agree (A16). An analogue variable on the digital module was rounded
    to an integer and switched as a digital line: 1.5 A on a coil became a 2 written to a
    digital output, without a word. A digital line on an analogue module did fail, but later,
    in a cast, with a message that named neither the variable nor the cause.
    """
    kinds = {True: "digital", False: "analogue"}
    wrong = [
        "  {} on module {}: {} by its name, but the module is {}".format(
            name,
            module,
            kinds[wt_variable.unit(name) == "digital"],
            kinds[module in modules__digital],
        )
        for name, module in zip(connections["variable"], connections["module"])
        if (wt_variable.unit(name) == "digital") != (module in modules__digital)
    ]
    if wrong:
        raise ValueError(
            "\n".join(
                [
                    "A variable and the module it is connected to disagree on whether it"
                    " is digital:",
                    "",
                    *wrong,
                    "",
                    "Connect it to a module of its kind, or rename it: a digital line has no"
                    " `__unit` suffix.",
                ]
            )
        )

check_modules_described(connections, machine_specifications)

Checks that every module a connection names is one the specifications describe.

Everything about a module – whether it is digital, and how its values become digits – is read from its entry in the specifications, so a connection to a module without one has no right answer. It used to be taken for analogue and converted as ±10 V, 16 bits.

Source code in src/wignertime/adwin/internal.py

def check_modules_described(connections, machine_specifications):
    """
    Checks that every module a connection names is one the specifications describe.

    Everything about a module – whether it is digital, and how its values become digits – is
    read from its entry in the specifications, so a connection to a module without one has no
    right answer. It used to be taken for analogue and converted as ±10 V, 16 bits.
    """
    count = len(machine_specifications["modules"])
    undescribed = sorted(
        {int(m) for m in connections["module"] if not 1 <= int(m) <= count}
    )
    if undescribed:
        raise ValueError(
            "Connections name module(s) {}, but the machine specifications describe modules"
            " 1 to {} only. Describe every installed module in"
            " `machine_specifications['modules']`, or correct the connections.".format(
                undescribed, count
            )
        )

conversion__module(machine_specifications, module)

How the values of an analogue module become digits, as the keyword arguments of conversion.to_digits: the module's voltage_range, its width in bits, and its gain, from its entry in the specifications.

Each module is converted with its own entry (D11, #125). Every analogue module the package had run on is ±10 V and 16 bits, which is why converting all of them with conversion.SPECIFICATIONS__DEFAULT went unnoticed: a module of any other range was converted wrongly, without a word. An entry missing one of the three raises, rather than being completed with those values.

Source code in src/wignertime/adwin/internal.py

def conversion__module(machine_specifications, module):
    """
    How the values of an analogue `module` become digits, as the keyword arguments of
    `conversion.to_digits`: the module's `voltage_range`, its width in `bits`, and its `gain`,
    from its entry in the specifications.

    Each module is converted with its own entry (D11, #125). Every analogue module the
    package had run on is ±10 V and 16 bits, which is why converting all of them with
    `conversion.SPECIFICATIONS__DEFAULT` went unnoticed: a module of any other range was
    converted wrongly, without a word. An entry missing one of the three raises, rather
    than being completed with those values.
    """
    entry = machine_specifications["modules"][module - 1]
    missing = [key for key in ("voltage_range", "bits", "gain") if key not in entry]
    if missing:
        raise ValueError(
            "Module {} is analogue, but its entry in the machine specifications gives no {},"
            " so its values cannot be converted to digits.".format(module, missing)
        )
    return {
        "voltage_range": entry["voltage_range"],
        "num_bits": entry["bits"],
        "gain": entry["gain"],
    }

modules__digital(machine_specifications)

The module numbers carrying digital connections, derived from the specifications.

A digital line is one bit wide, so a module of one-bit channels is a digital module. This is a derivation rather than a declaration, deliberately: a kind field beside bits would be a second source of truth, and a module declared {"kind": "digital", "bits": 16} would have no right answer. The same reasoning keeps module and channel numbers out of the device conversions.

Until 2026-09-22 the test read m.get("bits", False) == True, which picks out the digital module only because 1 == True in Python (D12/#126). Note what that did and did not cost: x == True and x == 1 agree for every number, so the answer was never wrong -- what was wrong was that "is one bit wide" was written as a comparison against a boolean, leaving the intent unrecoverable from the code.

A module that declares no bits at all now raises rather than being taken for analogue, which is the one behaviour that changed. Silently reading an under-specified module as analogue is a guess about hardware, and it would put a 16-bit conversion on a digital line.

NOTE: Modules are numbered from 1 (unlike Python lists).

Any number of modules may be digital, and the real-time program honours each: upload sends it the list, which its lowinit: programs as outputs, and the module of every digital row, as it does for the analogue ones (D18, #133).

Source code in src/wignertime/adwin/internal.py

def modules__digital(machine_specifications):
    """
    The module numbers carrying digital connections, derived from the specifications.

    A digital line is one bit wide, so a module of one-bit channels is a digital module.
    This is a *derivation* rather than a declaration, deliberately: a `kind` field beside
    `bits` would be a second source of truth, and a module declared
    `{"kind": "digital", "bits": 16}` would have no right answer. The same reasoning
    keeps module and channel numbers out of the device conversions.

    Until 2026-09-22 the test read `m.get("bits", False) == True`, which picks out the
    digital module only because `1 == True` in Python (D12/#126). Note what that did and
    did not cost: `x == True` and `x == 1` agree for every number, so the answer was
    never wrong -- what was wrong was that "is one bit wide" was written as a comparison
    against a boolean, leaving the intent unrecoverable from the code.

    A module that declares no `bits` at all now raises rather than being taken for
    analogue, which is the one behaviour that changed. Silently reading an
    under-specified module as analogue is a guess about hardware, and it would put a
    16-bit conversion on a digital line.

    NOTE: Modules are numbered from 1 (unlike Python lists).

    Any number of modules may be digital, and the real-time program honours each: `upload`
    sends it the list, which its `lowinit:` programs as outputs, and the module of every
    digital row, as it does for the analogue ones (D18, #133).
    """
    modules = machine_specifications["modules"]

    modules__unspecified = [i + 1 for i, m in enumerate(modules) if "bits" not in m]
    if modules__unspecified:
        raise ValueError(
            "Module(s) {} declare no `bits`, so whether they are digital cannot be"
            " determined. Every entry of `machine_specifications['modules']` needs its"
            " width: 1 for a digital module, 16 for the usual analogue one.".format(
                modules__unspecified
            )
        )

    return [i + 1 for i, m in enumerate(modules) if m["bits"] == 1]

specifications(machine_specifications=None)

The machine specifications to convert against: those given, or else SPECIFICATIONS__DEFAULT as it stands when this is called.

Read here rather than bound as a default argument, so that rebinding the global reaches every function that uses it (D23).

An entry for cycle_period is refused rather than ignored. The period used to live here, and a specification written then would otherwise be accepted with its period silently unused -- the failure D15 describes, in a new place.

Source code in src/wignertime/adwin/internal.py

def specifications(machine_specifications=None):
    """
    The machine specifications to convert against: those given, or else
    `SPECIFICATIONS__DEFAULT` as it stands when this is called.

    Read here rather than bound as a default argument, so that rebinding the global
    reaches every function that uses it (D23).

    An entry for `cycle_period` is refused rather than ignored. The period used to live
    here, and a specification written then would otherwise be accepted with its period
    silently unused -- the failure D15 describes, in a new place.
    """
    if machine_specifications is None:
        machine_specifications = SPECIFICATIONS__DEFAULT

    if "cycle_period" in machine_specifications:
        raise ValueError(
            "The machine specifications carry a `cycle_period`, which is no longer read"
            " from them: it belongs to the program loaded on the machine, not to a"
            " description of its modules. Pass it to `adwin.core.convert` as"
            " `cycle_period` instead, and remove it from the specifications."
        )

    return machine_specifications

to_tuples(timeline, machine_specifications=None)

Takes a full, ADwin-compatible, dataframe of the experimental run and converts the result to an output format that can be processed by ADwin (tuples), separating analogue and digital values.

return [[(cycle, module, channel, value), ...], [(cycle, module, channel, value), ...]]

Source code in src/wignertime/adwin/internal.py

def to_tuples(timeline, machine_specifications=None):
    """
    Takes a full, ADwin-compatible, dataframe of the experimental run and converts the result to an output format that can be processed by ADwin (tuples), separating analogue and digital values.

    return [[(cycle, module, channel, value), ...],
    [(cycle, module, channel, value), ...]]
    """
    wtl.debug("Got to `output`")
    machine_specifications = specifications(machine_specifications)

    if not timeline["cycle"].is_monotonic_increasing:
        timeline = timeline.sort_values(by=["cycle"], ignore_index=True)

    if not ("module" in timeline.columns):
        raise ValueError(
            "No `module` listed in timeline. Remember to add ADwin specifications before ADwin export."
        )

    mods_digital = modules__digital(machine_specifications)
    mods_analogue = [x for x in timeline["module"].unique() if x not in mods_digital]

    # NOTE: filtered by value rather than by a formatted query string. `module` is an
    # int64 column, so `unique()` yields numpy scalars, and building a query out of them
    # produced `module in [np.int64(3), np.int64(4)]` -- which pandas parses and then
    # rejects with `UndefinedVariableError: name 'np' is not defined`, an error that
    # looks like it comes from inside pandas and says nothing about modules (#41). A
    # bare `int()` on each element used to hold that off; comparing values removes the
    # failure mode instead of guarding it.
    return [
        to_tuples__raw(wt_frame.subframe(timeline, "module", mods_analogue)),
        to_tuples__raw(wt_frame.subframe(timeline, "module", mods_digital)),
    ]

to_tuples__raw(timeline, cols=['cycle', 'module', 'channel', 'value__digits'])

A raw extraction of ADwin-relevant values from a timeline, regardless of whether or not the module is digital or not.

NOTE: No validation is done here.

Source code in src/wignertime/adwin/internal.py

def to_tuples__raw(timeline, cols=["cycle", "module", "channel", "value__digits"]):
    """
    A raw extraction of ADwin-relevant values from a `timeline`, regardless of whether or not the module is digital or not.

    NOTE: No validation is done here.
    """
    return [tuple([np.int32(i) for i in x]) for x in timeline[cols].values]

validate

all(timeline, do_drop_repeats=True)

Includes ADwin-specific methods ontop of the basic timeline sanitization for removing unnecessary points and raising errors on illogical input.

do_drop_repeats can be turned off for debugging, or to inspect the unfiltered cost of an expansion.

Source code in src/wignertime/adwin/validate.py

def all(timeline, do_drop_repeats=True):
    """
    Includes ADwin-specific methods ontop of the basic timeline sanitization for removing unnecessary points and raising errors on illogical input.

    `do_drop_repeats` can be turned off for debugging, or to inspect the unfiltered cost of an expansion.
    """
    return funcy.compose(
        drop_repeats if do_drop_repeats else funcy.identity,
        drop_duplicates,
        special_contexts,
        types,
        cycles,
    )(timeline)

ascending(output)

Refuses converted arrays whose cycles are not in ascending order (D20).

output is what internal.to_tuples returns, [analogue, digital], each a list of (cycle, module, channel, digits). The real-time program walks each array with an index that only moves forward, so a row out of order is not skipped but played late: at the cycle of the row before it. Nothing on the machine can tell.

to_tuples sorts, so this holds by construction today. It is checked on the arrays themselves because they are the contract with the machine, and nothing between here and the upload would notice if it stopped holding.

Source code in src/wignertime/adwin/validate.py

def ascending(output):
    """
    Refuses converted arrays whose cycles are not in ascending order (D20).

    `output` is what `internal.to_tuples` returns, `[analogue, digital]`, each a list of
    `(cycle, module, channel, digits)`. The real-time program walks each array with an
    index that only moves forward, so a row out of order is not skipped but played
    *late*: at the cycle of the row before it. Nothing on the machine can tell.

    `to_tuples` sorts, so this holds by construction today. It is checked on the arrays
    themselves because they are the contract with the machine, and nothing between here
    and the upload would notice if it stopped holding.
    """
    for label, rows in zip(["analogue", "digital"], output):
        cycle = np.array([row[0] for row in rows], dtype=np.int64)
        backwards = np.flatnonzero(np.diff(cycle) < 0)
        if len(backwards):
            # Rows are counted from 1, as the controller's arrays are.
            i = int(backwards[0])
            raise ValueError(
                "The {} array is not in ascending cycle order: row {} is at cycle {},"
                " after one at cycle {}. The controller would play it late, at"
                " cycle {}.".format(label, i + 2, cycle[i + 1], cycle[i], cycle[i])
            )

    return output

cycles(timeline, special_contexts=None)

Refuses a row, outside the special contexts, at a cycle the real-time program cannot play.

Such a row belongs in wt_adwin.CYCLES__RUN. Everything outside that range is the territory of the special contexts' sentinels, and each way of leaving it fails silently on the machine rather than here (D19):

• at cycle -1 or -2, the row is played with the init or lowinit rows;
• below -2, it sorts ahead of the lowinit rows, and since the backend plays only the row its index points at, and only when the count reaches it, the index never moves: no row of that array is played at all -- not the run, not the initial state, not the final one;
• past the last cycle, the machine's 32-bit counter wraps negative, into the sentinels, so the run replays its initialisation and does not end.

A row earlier than half a cycle before zero is therefore an error, however it came about, rather than something to shift or drop. Must run before types, which narrows the column to 32 bits and would wrap the late rows into plausible ones.

Source code in src/wignertime/adwin/validate.py

def cycles(timeline, special_contexts=None):
    """
    Refuses a row, outside the special contexts, at a cycle the real-time program cannot play.

    Such a row belongs in `wt_adwin.CYCLES__RUN`. Everything outside that range is the
    territory of the special contexts' sentinels, and each way of leaving it fails
    silently on the machine rather than here (D19):

    - at cycle -1 or -2, the row is played with the init or lowinit rows;
    - below -2, it sorts ahead of the lowinit rows, and since the backend plays only the
      row its index points at, and only when the count reaches it, the index never moves:
      **no row of that array is played at all** -- not the run, not the initial state,
      not the final one;
    - past the last cycle, the machine's 32-bit counter wraps negative, into the
      sentinels, so the run replays its initialisation and does not end.

    A row earlier than half a cycle before zero is therefore an error, however it came
    about, rather than something to shift or drop. Must run before `types`, which narrows
    the column to 32 bits and would wrap the late rows into plausible ones.
    """
    if special_contexts is None:
        special_contexts = wt_adwin.CONTEXTS__SPECIAL

    first, last = wt_adwin.CYCLES__RUN
    mask__run = ~timeline["context"].isin(list(special_contexts))
    mask__outside = mask__run & ~timeline["cycle"].between(first, last)

    if mask__outside.any():
        raise ValueError(
            "Rows outside the special contexts {} must fall within cycles {}..{} of the"
            " run. A row before the start stops every row of its array from being"
            " played, the initial and final states included; one past the end wraps the"
            " controller's cycle counter. Offending rows:\n{}".format(
                list(special_contexts),
                first,
                last,
                timeline.loc[mask__outside, ["variable", "time", "context", "cycle"]],
            )
        )

    return timeline

drop_duplicates(timeline, subset=['variable', 'cycle'], unless_context=list(wt_adwin.CONTEXTS__SPECIAL.keys()))

An alternative to that in timeline, to deal with ADwin-specific cases.

Drop rows where the columns specified in subset are both duplicated, except for in the specific contexts listed.

This removes temporal collisions: two entries for the same variable that round to the same cycle. The later entry wins, as wt_frame.duplicated keeps the last occurrence.

Source code in src/wignertime/adwin/validate.py

def drop_duplicates(
    timeline,
    subset=["variable", "cycle"],
    unless_context=list(wt_adwin.CONTEXTS__SPECIAL.keys()),
):
    """
    An alternative to that in timeline, to deal with ADwin-specific cases.

    Drop rows where the columns specified in `subset` are both duplicated, except for in the specific `context`s listed.

    This removes *temporal* collisions: two entries for the same variable that round to the same cycle. The later entry wins, as `wt_frame.duplicated` keeps the last occurrence.
    """
    mask__duplicates = wt_frame.duplicated(timeline, subset=subset)

    return timeline[~mask__duplicates | (timeline["context"].isin(unless_context))]

drop_repeats(timeline, subset=['module', 'channel'], column__value='value__digits', column__order='cycle', unless_context=list(wt_adwin.CONTEXTS__SPECIAL.keys()))

Drops rows that would command a channel to the value it already holds.

Where drop_duplicates removes temporal collisions, this removes value redundancy: rows whose digitized value is unchanged from the previous row on the same physical channel. The ADwin program only acts when a value differs from the preceding one, so these rows occupy memory and transfer bandwidth without ever producing an output change.

The effect is largest on expanded ramps. A ramp sampled at the cycle period produces one row per cycle, but a device can only express as many distinct outputs as it has digits across the ramp's span; the surplus rows repeat. Since the sampling grid is the hardware's own time grid and the value grid is the converter's least significant bit, the rows that survive are exactly the transitions the hardware is capable of making. Filtering here is therefore not an approximation to bit-flip-timed expansion – on this grid it is equivalent to it, and it needs no inverse of the ramp function, which would be unavailable for interpolated calibrations in any case.

Grouping is by physical channel rather than by variable, since the channel is the state the real-time program compares against.

Rows in the special contexts are excluded from the comparison entirely and always kept: they carry the safe-state guarantee and have no position in time.

The first and last row of every channel are always kept. Retaining the last matters beyond making the final commanded state explicit: adwin.core.upload takes the run length from the highest non-special cycle, and the tail of a hyperbolic-tangent ramp is flat, so dropping trailing repeats would silently shorten the experiment.

Source code in src/wignertime/adwin/validate.py

def drop_repeats(
    timeline,
    subset=["module", "channel"],
    column__value="value__digits",
    column__order="cycle",
    unless_context=list(wt_adwin.CONTEXTS__SPECIAL.keys()),
):
    """
    Drops rows that would command a channel to the value it already holds.

    Where `drop_duplicates` removes temporal collisions, this removes *value* redundancy: rows whose digitized value is unchanged from the previous row on the same physical channel. The ADwin program only acts when a value differs from the preceding one, so these rows occupy memory and transfer bandwidth without ever producing an output change.

    The effect is largest on expanded `ramp`s. A ramp sampled at the cycle period produces one row per cycle, but a device can only express as many distinct outputs as it has digits across the ramp's span; the surplus rows repeat. Since the sampling grid is the hardware's own time grid and the value grid is the converter's least significant bit, the rows that survive are exactly the transitions the hardware is capable of making. Filtering here is therefore not an approximation to bit-flip-timed expansion – on this grid it is equivalent to it, and it needs no inverse of the ramp function, which would be unavailable for interpolated calibrations in any case.

    Grouping is by physical channel rather than by `variable`, since the channel is the state the real-time program compares against.

    Rows in the special `context`s are excluded from the comparison entirely and always kept: they carry the safe-state guarantee and have no position in time.

    The first and last row of every channel are always kept. Retaining the last matters beyond making the final commanded state explicit: `adwin.core.upload` takes the run length from the highest non-special cycle, and the tail of a hyperbolic-tangent ramp is flat, so dropping trailing repeats would silently shorten the experiment.
    """
    columns__needed = set(subset) | {column__value, column__order, "context"}
    if not columns__needed.issubset(timeline.columns):
        return timeline

    mask__special = timeline["context"].isin(unless_context)
    timeline__run = timeline[~mask__special]

    if timeline__run.empty:
        return timeline

    mask__changed = wt_frame.mask__changed(
        timeline__run,
        subset=subset,
        column__value=column__value,
        column__order=column__order,
    )

    # `mask__changed` covers only the non-special rows, so it is widened back to the
    # full index. `fill_value` is what keeps this a boolean mask: reindexing without
    # one introduces NaN, which bool cannot hold, silently upcasting to object dtype.
    mask__keep = mask__special | mask__changed.reindex(timeline.index, fill_value=False)

    return timeline[mask__keep]

special_contexts(timeline, special_contexts=wt_adwin.CONTEXTS__SPECIAL)

Ensures that a variable has at most one entry before the run, across ADwin_LowInit and ADwin_Init together, and at most one after it, in ADwin_Finish. There is no concept of 'time' inside the special contexts.

The contexts before the run are counted together (#154, (b1), settled by the maintainer 2026-09-29): both are at −∞, so nothing in the timeline orders them, and a value lookup would otherwise see two candidates there. They used to be counted per context. What is lost is changing a variable between lowinit: and init:, which nothing did. A context's side is the sign of its sentinel.

Similarly, the time values are adjusted to avoid automatic removal later on.

Source code in src/wignertime/adwin/validate.py

def special_contexts(timeline, special_contexts=wt_adwin.CONTEXTS__SPECIAL):
    """
    Ensures that a variable has at most one entry **before the run**, across
    `ADwin_LowInit` and `ADwin_Init` together, and at most one after it, in
    `ADwin_Finish`. There is no concept of 'time' inside the special contexts.

    The contexts before the run are counted together (#154, (b1), settled by the
    maintainer 2026-09-29): both are at −∞, so nothing in the timeline orders them, and a
    value lookup would otherwise see two candidates there. They used to be counted per
    context. What is lost is changing a variable between `lowinit:` and `init:`, which
    nothing did. A context's side is the sign of its sentinel.

    Similarly, the time values are adjusted to avoid automatic removal later on.
    """
    df = timeline[timeline["context"].isin(special_contexts)]
    side = df["context"].map(
        lambda c: "before the run" if special_contexts[c] < 0 else "after the run"
    )
    df_N = df.assign(side=side).groupby(["variable", "side"])["context"].agg(list)
    duplicates = df_N[df_N.map(len) > 1].reset_index()
    duplicates.columns = ["variable", "side", "contexts"]

    # Replace time values with those specified in wt_adwin.CONTEXTS__SPECIAL
    timeline = wt_frame.replace_column__filtered(timeline, wt_adwin.CONTEXTS__SPECIAL)

    if duplicates.empty:
        return timeline
    else:
        raise ValueError(
            "The same variable has more than one value before the run (ADwin_LowInit and"
            " ADwin_Init together) or after it (ADwin_Finish). These special contexts have"
            " no concept of time, so nothing orders the values. For details, see the"
            " duplicate information:\n" + str(duplicates)
        )

config

CONTEXT__INFER = INFER module-attribute

INFER, under the name the context signatures use.

INFER = _Infer() module-attribute

The default of origin in update, anchor and ramp, and of context in create, update, anchor and ramp (#142). It is there so that a signature shows that the default does something: the origin is inferred from the timeline being extended (see ORIGIN__DEFAULTS and ORIGIN__DEFAULTS__RAMP), and an unstated context is inherited from it, and help(tl.update) reads origin=INFER.

None means exactly the same, both as the whole argument and in either slot of an origin pair. That is deliberate. A stage that takes origin=None or context=None and passes it on -- the usual way of writing "no opinion of my own" -- then gets the library's default without having to know that this object exists, and a pair such as [None, 0.0] reads the same as [INFER, 0.0]. Absolute placement is a number, origin=0.0, and reads the same in update, anchor and ramp: a ramp always starts where its variable is (#142), so its value slot is never an origin question. There is no way to switch context inheritance off: every row has a context, stated or inherited (#156).

It is an object rather than a string so that it cannot be mistaken for a name: context="INFER" is an ordinary context.

ORIGIN__DEFAULTS = [[ANCHOR, None], [LAST, None]] module-attribute

For update and anchor, whose values are absolute -- hence None in every value slot.

ORIGIN__DEFAULTS__RAMP = [[ANCHOR, VARIABLE], [LAST, VARIABLE]] module-attribute

For ramp, which is the one core function that needs a value origin: a ramp runs from wherever the variable currently sits to the target, so its start value has to be looked up. Hence VARIABLE in the value slot -- the variable's own previous value, bounded by the time origin. It is the only thing that slot can hold: a ramp always starts where its variable is (#142), and ramp refuses anything else written there.

The LAST step is not decoration. Without it (this was a single-entry list until 2026-09-18) a ramp onto a timeline holding no anchor fell off the end of the chain and landed in absolute time, so it could be placed before the rows it was appended to, with no warning. See KNOWN_ISSUES A4.

ORIGIN__INFER = INFER module-attribute

INFER, under the name the origin signatures use.

VARIABLE__REGEX = re.compile('^([^_]+)__([^_]+(?:_[^_]+)*)(?:__([^_]+))?$') module-attribute

The naming convention for a variable, as <device>__<UID>(__<unit>), e.g. coil__MOT_lower__A and shutter__MOT; the three groups are what variable.parse returns. A missing __<unit> is what marks a line as digital. <device> and <unit> contain no _; the <UID> may contain single ones (D7, #121, settled by the maintainer 2026-09-29).

Keeping _ out of <device> is what makes the grammar safe to migrate to: a name in the old <device>_<UID>(__<unit>) form has a _ before its first __, so it is refused rather than read with its unit taken for a UID, i.e. as digital. It also refuses a device of several words, power_supply__X__V, which the old grammar split silently; such a device is supply__power_X__V.

This lives here, rather than in variable, because the convention is a default rather than a law: a site that consistently applies a different one can rebind this name before building its connections. variable.parse and adwin.connection.is_valid_name read it at call time, so an override takes effect without reimporting.

Origin

Bases: Enum

The reserved origin words, as tags rather than strings (#158, 2026-09-29).

A string in an origin is always a name -- of a variable or of a context -- and a tag is always a rule. While the rules were the strings "anchor", "last" and "variable", a string was sometimes one and sometimes the other, so a context or a variable named after one of the words could not be referred to, and had to be refused where it was written (A9). Tags are compared with is, print as their bare names, as INFER does, and never reach the data, since an origin is resolved while the timeline is built.

• ANCHOR: the most recent anchor;
• LAST: the latest entry of the timeline;
• VARIABLE: each variable relative to its own most recent entry, in time or in value -- the one tag admissible in either slot.

Source code in src/wignertime/config.py

class Origin(enum.Enum):
    """
    The reserved origin words, as tags rather than strings (#158, 2026-09-29).

    A string in an `origin` is always a *name* -- of a variable or of a context -- and a
    tag is always a *rule*. While the rules were the strings `"anchor"`, `"last"` and
    `"variable"`, a string was sometimes one and sometimes the other, so a context or a
    variable named after one of the words could not be referred to, and had to be refused
    where it was written (A9). Tags are compared with `is`, print as their bare names, as
    `INFER` does, and never reach the data, since an origin is resolved while the
    timeline is built.

    - `ANCHOR`: the most recent anchor;
    - `LAST`: the latest entry of the timeline;
    - `VARIABLE`: each variable relative to its own most recent entry, in time or in
      value -- the one tag admissible in either slot.
    """

    ANCHOR = "anchor"
    LAST = "last"
    VARIABLE = "variable"

    def __repr__(self):
        return self.name

    __str__ = __repr__

conversion

add(timeline, specifications=SPECIFICATIONS__DEFAULT, column__conversion='to_V', column__new='value__digits')

Performs a conversion, according to the associated factor or function, adds the resulting values as another column, value__digits, and returns the result.

Source code in src/wignertime/conversion.py

def add(
    timeline: wt_frame.CLASS,
    specifications=SPECIFICATIONS__DEFAULT,
    column__conversion: str = "to_V",
    column__new: str = "value__digits",
) -> wt_frame.CLASS:
    """
    Performs a conversion, according to the associated factor or function, adds the resulting values as another column, `value__digits`, and returns the result.
    """
    if column__conversion in timeline.columns:
        dff = _add_linear(
            timeline,
            column__conversion=column__conversion,
            column__new=column__new,
            specifications=specifications,
        )
        return _add_function(
            dff,
            specifications=specifications,
            column__conversion=column__conversion,
            column__new=column__new,
        )

    else:
        raise ValueError(
            f"Cannot convert values because {column__conversion} column does not exist. "
        )

function_from_file(path, method='cubic', fill_value='extrapolate', indices__column=[0, 1], **read_csv__args)

An interpolation function drawn from two columns of a CSV-like calibration file.

NOTE: If you would like to invert the interpolation then just specify the columns backwards, e.g. indices__column=[1,0]

e.g. function_from_file( "resources/calibration/aom_calibration.dat", names=["voltage", "transparency"], sep=r"\s+", ),

Source code in src/wignertime/conversion.py

def function_from_file(
    path,
    method="cubic",
    fill_value="extrapolate",
    indices__column=[0, 1],
    **read_csv__args,
):
    r"""
    An interpolation function drawn from *two columns* of a CSV-like calibration file.

    NOTE: If you would like to invert the interpolation then just specify the columns backwards, e.g. indices__column=[1,0]

    e.g.
    function_from_file(
        "resources/calibration/aom_calibration.dat",
        names=["voltage", "transparency"],
        `sep=r"\s+"`,
    ),
    """
    df = pd.read_csv(path, **read_csv__args).dropna()

    # Deal with possible x-duplicates
    columns = df.columns
    df_avg = df.groupby(columns[indices__column[0]], as_index=False).mean()

    return interp1d(
        df_avg.iloc[:, 0],
        df_avg.iloc[:, indices__column[1]],
        kind=method,
        fill_value=fill_value,
    )

to_digits(voltage, voltage_range=[-10.0, 10.0], num_bits=16, gain=1)

Transforms any voltage linearly to analogue-digital-converter(ADC) digits.

Source code in src/wignertime/conversion.py

def to_digits(voltage, voltage_range=[-10.0, 10.0], num_bits: int = 16, gain: int = 1):
    """
    Transforms any voltage linearly to analogue-digital-converter(ADC) digits.
    """

    v = np.asarray(voltage, dtype=float)
    v_min, v_max = np.asarray(voltage_range) / gain

    result = np.round(((v - v_min) / (v_max - v_min)) * (2**num_bits - 1)).astype(int)
    return result.item() if result.ndim == 0 else result

demo

full_experiment

An example implementation of a real experiment, using 'Wigner Time' timelines.

As well as providing conveniences, the functions can be used to document the intention and meaning of each stage.

connections = adcon.new(['shutter__MOT', 1, 11], ['shutter__repump', 1, 12], ['shutter__OP1', 1, 14], ['shutter__OP2', 1, 15], ['shutter__science', 1, 10], ['shutter__transverse_pump', 1, 9], ['AOM__MOT', 1, 1], ['AOM__repump', 1, 2], ['AOM__OP_aux', 1, 30], ['AOM__OP', 1, 31], ['coil__compensation_X__A', 4, 7], ['coil__compensation_Y__A', 3, 2], ['coil__MOT_lower__A', 4, 1], ['coil__MOT_upper__A', 4, 3], ['coil__MOT_lower_plus__A', 4, 2], ['coil__MOT_upper_plus__A', 4, 4], ['lockbox__MOT__MHz', 3, 8], ['trigger__TC__V', 3, 1], ['AOM__science', 1, 4], ['AOM__science__trans', 4, 8], ['trigger__camera', 1, 0]) module-attribute

devices stores how to map our physical quantities to an implementation voltage, as well as specifying the range of values that should be allowed for this variable. A linear conversion is the factor taking the unit to volts, so a device whose permitted range spans the controller's full output is 10 / limit — 10 / 5.0 for a coil driven over +/-5 A by a +/-10 V DAC.

These specifications are deliberately separated from connections because they represent physical properties and conversions that are independent of the particular DAC wiring.

devices = device.new(['coil__compensation_X__A', 10 / 3.0, -3, 3], ['coil__compensation_Y__A', 10 / 3.0, -3, 3], ['coil__MOT_lower__A', 10 / 5.0, -5, 5], ['coil__MOT_upper__A', 10 / 5.0, -5, 5], ['coil__MOT_lower_plus__A', 10 / 5.0, -5, 5], ['coil__MOT_upper_plus__A', 10 / 5.0, -5, 5], ['lockbox__MOT__MHz', 0.05, -200, 200], ['trigger__TC__V', 1.0, -10, 10], ['AOM__science__trans', conv.function_from_file('resources/calibration/aom_calibration.dat', sep='\\s+'), 0.0, 1.0]) module-attribute

'constants' allow us to store site-specific details that help define our exeriment.

MOT(duration=15, lower_current=-1.0, upper_current=-0.98)

Creates a Magneto-Optical Trap.

Source code in src/wignertime/demo/full_experiment.py

def MOT(duration=15, lower_current=-1.0, upper_current=-0.98):
    """
    Creates a Magneto-Optical Trap.
    """
    return tl.stack(
        tl.update(
            shutter__MOT=1,
            shutter__repump=1,
            coil__MOT_lower__A=lower_current,
            coil__MOT_upper__A=upper_current,
        ),
        tl.anchor(duration),
        context="MOT",
    )

MOT_detuned_growth(duration=0.1, duration_ramp=0.01, detuning__MHz=-5)

Final stage of MOT collection with detuned MOT beams for increased capture range.

Source code in src/wignertime/demo/full_experiment.py

def MOT_detuned_growth(duration=100e-3, duration_ramp=10e-3, detuning__MHz=-5):  # pt=3,
    """
    Final stage of MOT collection with detuned MOT beams for increased capture range.
    """
    return tl.stack(
        tl.ramp(
            lockbox__MOT__MHz=detuning__MHz,
            duration=duration_ramp,
            #            fargs={"ti": pt},
        ),
        tl.anchor(duration),
        context="MOT",
    )

default_state(MOT_ON=True, **kwargs)

Starts/leaves the system in a sane state that is appropriate for creating a new timeline

As a general rule, AOMs are kept on as long as possible to keep them in thermal equilibrium. When needed, we turn them off before the opening of the shutter. The same holds for the coils.

The same stage at both ends of the experiment: init and finish differ only in the time, the context and MOT_ON.

Source code in src/wignertime/demo/full_experiment.py

def default_state(MOT_ON=True, **kwargs):
    """
    Starts/leaves the system in a sane state that is appropriate for creating a new timeline

    As a general rule, AOMs are kept on as long as possible to keep them in thermal equilibrium. When needed, we turn them off before the opening of the shutter. The same holds for the coils.

    The same stage at both ends of the experiment: `init` and `finish` differ only in the
    time, the context and `MOT_ON`.
    """
    return tl.stack(
        tl.update(
            lockbox__MOT__MHz=0.0,
            coil__compensation_X__A=constants.Compensation.X__A,
            coil__compensation_Y__A=constants.Compensation.Y__A,
            coil__MOT_lower_plus__A=-constants.Compensation.Z__A,
            coil__MOT_upper_plus__A=constants.Compensation.Z__A,
            AOM__MOT=1,
            AOM__repump=1,
            AOM__OP_aux=0,
            AOM__OP=1,
            AOM__science=1,
            shutter__MOT=MOT_ON,
            shutter__repump=MOT_ON,
            shutter__OP1=0,
            shutter__OP2=1,
            shutter__science=0,
            shutter__transverse_pump=0,
            AOM__science__trans=1.0,
            trigger__TC__V=0.0,
            **kwargs,
        )
    )

finish(wait=1, lower_current=-1.0, upper_current=-0.98, MOT_ON=True, **kwargs)

Safely winds down the system, 'ramping' the analog variables to the “default state” in a given duration by the default ramp_function.

The anchor function is used to specify a key time instant, around which other times can be specified.

The ADwin_Finish environment means that the “default state” will be actuated even when the process is interrupted.

Source code in src/wignertime/demo/full_experiment.py

def finish(wait=1, lower_current=-1.0, upper_current=-0.98, MOT_ON=True, **kwargs):
    """
    Safely winds down the system, 'ramping' the analog variables to the “default state” in a given duration by the default `ramp_function`.

    The `anchor` function is used to specify a key time instant, around which other times can be specified.

    The ADwin_Finish environment means that the “default state” will be actuated even when the process is interrupted.
    """
    duration = 1e-2
    return tl.stack(
        tl.anchor(wait, context="finalRamps"),
        tl.ramp(
            lockbox__MOT__MHz=0.0,
            coil__MOT_lower__A=lower_current,
            coil__MOT_upper__A=upper_current,
            coil__compensation_X__A=constants.Compensation.X__A,
            coil__compensation_Y__A=constants.Compensation.Y__A,
            coil__MOT_lower_plus__A=-constants.Compensation.Z__A,
            coil__MOT_upper_plus__A=constants.Compensation.Z__A,
            duration=duration,
            context="finalRamps",
        ),
        default_state(
            time=math.inf,  # after the run: no instant, and 'ADwin_Finish' is played when it ends, or is stopped (#154)
            context="ADwin_Finish",
            MOT_ON=MOT_ON,
            **kwargs,
        ),
    )

magnetic_trapping(duration_initial=5e-05, lower_current_initial=-1.8, upper_current_initial=-1.7, duration_strengthen=0.003, lower_current_strengthen=-4.8, upper_current_strengthen=-4.7)

Does what it says on the tin.

Source code in src/wignertime/demo/full_experiment.py

def magnetic_trapping(
    duration_initial=50e-6,
    lower_current_initial=-1.8,
    upper_current_initial=-1.7,
    duration_strengthen=3e-3,
    lower_current_strengthen=-4.8,
    upper_current_strengthen=-4.7,
):
    """
    Does what it says on the tin.
    """
    return tl.stack(
        pull_coils(
            duration_initial,
            lower_current_initial,
            upper_current_initial,
        ),
        pull_coils(
            duration_strengthen,
            lower_current_strengthen,
            upper_current_strengthen,
            time=duration_initial,
        ),
        tl.anchor(duration_initial + duration_strengthen),
        context="magnetic_trapping",
    )

molasses(duration=0.005, duration_coil_ramp=0.0009, duration_lockbox_ramp=0.001, to__MHz=-90, delay=0)

For slowing down the atoms by creating an optical density.

Source code in src/wignertime/demo/full_experiment.py

def molasses(
    duration=5e-3,
    duration_coil_ramp=9e-4,
    duration_lockbox_ramp=1e-3,
    to__MHz=-90,  # coil_pt=3, lockbox_pt=3,
    delay=0,  # arbitrary delay to shutter for ad hoc compensation of small drifts
):
    """
    For slowing down the atoms by creating an optical density.
    """

    return tl.stack(
        tl.ramp(
            coil__MOT_lower__A=0,
            coil__MOT_upper__A=0,
            duration=duration_coil_ramp,
            #            fargs={"ti": coil_pt},
        ),
        tl.ramp(
            lockbox__MOT__MHz=to__MHz,
            duration=duration_lockbox_ramp,
            #            fargs={"ti": lockbox_pt},
        ),
        tl.update(
            shutter__MOT=[duration - constants.lag_MOT_shutter + delay, 0],
            AOM__MOT=[duration, 0],
        ),
        tl.anchor(duration),
        context="molasses",
    )

optical_pumping(duration_exposition=8e-05, duration_coil_ramp=5e-05, i=-0.12, delay1=0, delay2=0, delay_repump=0, delay_shutter_reinitialization=0.1)

Creates an experimental timeline for optical pumping.

NOTE: The AOM is switched off close to, but before, the opening of the first shutter

WARNING: Shutters are reinitialized so that additional optical pumping stages can be added later.

Source code in src/wignertime/demo/full_experiment.py

def optical_pumping(
    duration_exposition=80e-6,
    duration_coil_ramp=50e-6,
    i=-0.12,  # pt=3,
    delay1=0,
    delay2=0,
    delay_repump=0,  # arbitrary delays to shutters for ad hoc compensation of small drifts
    delay_shutter_reinitialization=0.1,
):
    """
    Creates an experimental timeline for optical pumping.

    NOTE:
    The AOM is switched off close to, but before, the opening of the first shutter

    WARNING:
    Shutters are reinitialized so that additional optical pumping stages can be added later.
    """

    duration_full = duration_exposition + duration_coil_ramp
    return tl.stack(
        tl.ramp(
            coil__MOT_lower__A=i,
            coil__MOT_upper__A=-i,
            duration=duration_coil_ramp,
            #            fargs={"ti": pt},
        ),
        tl.update(AOM__OP=[[-0.1, 0], [duration_coil_ramp, 1], [duration_full, 0]]),
        tl.update(
            shutter__OP1=[
                [duration_coil_ramp - constants.OP.lag_shutter_on + delay1, 1],
                [delay_shutter_reinitialization, 0],
            ]
        ),
        tl.update(
            shutter__OP2=[
                [duration_full - constants.OP.lag_shutter_off + delay2, 0],
                [delay_shutter_reinitialization, 1],
            ]
        ),
        tl.update(
            shutter__repump=0,
            time=duration_full - constants.lag_repump_shutter + delay_repump,
        ),
        tl.update(AOM__repump=0, time=duration_full),
        tl.anchor(duration_full),
        context="optical_pumping",
    )

pull_coils(duration, lower_current, upper_current, lower_plus_current=0, upper_plus_current=0, pt=3, time=None)

Controls the concentric coil pairs responsible for 'pulling' the atoms.

It names no context: it inherits the one of the stage it is part of, as magnetic_trapping's two calls do.

Source code in src/wignertime/demo/full_experiment.py

def pull_coils(
    duration,
    lower_current,
    upper_current,
    lower_plus_current=0,
    upper_plus_current=0,
    pt=3,
    time=None,
):
    """
    Controls the concentric coil pairs responsible for 'pulling' the atoms.

    It names no context: it inherits the one of the stage it is part of, as
    `magnetic_trapping`'s two calls do.
    """
    return tl.ramp(
        coil__MOT_lower__A=lower_current,
        coil__MOT_upper__A=upper_current,
        coil__MOT_lower_plus__A=lower_plus_current - constants.Compensation.Z__A,
        coil__MOT_upper_plus__A=upper_plus_current + constants.Compensation.Z__A,
        function=lambda origin, terminus, time_resolution: ramp_function.tanh(
            origin, terminus, time_resolution, pt
        ),
        duration=duration,
        time=time,
    )

trigger_camera(time, exposure, context, origin)

Opens the camera for exposure, starting time after origin.

origin has no default, because this stage is placed rather than chained: with one, a call that forgot it would fall back on the latest anchor and land wherever the timeline happened to end, without an error (maintainer, 2026-10-06).

context is required rather than inherited: a trigger placed into a finished timeline would otherwise adopt the context of its last row at an instant, finalRamps, and a camera trigger is not part of the final ramps.

The camera is not part of the default state, so the timeline it is placed into should set it initially and finally, e.g. init(trigger__camera=0) and finish(trigger__camera=0).

Source code in src/wignertime/demo/full_experiment.py

def trigger_camera(time, exposure, context, origin):
    """
    Opens the camera for `exposure`, starting `time` after `origin`.

    `origin` has no default, because this stage is placed rather than chained: with one,
    a call that forgot it would fall back on the latest anchor and land wherever the
    timeline happened to end, without an error (maintainer, 2026-10-06).

    `context` is required rather than inherited: a trigger placed into a finished timeline would otherwise adopt the context of its last row at an instant, `finalRamps`, and a camera trigger is not part of the final ramps.

    The camera is not part of the default state, so the timeline it is placed into should set it initially and finally, e.g. `init(trigger__camera=0)` and `finish(trigger__camera=0)`.
    """
    return tl.update(
        trigger__camera=[[time, 1], [time + exposure, 0]],
        context=context,
        origin=origin,
    )

device

A device is represented by a dataframe that contains a variable (with a given unit) and a means, scalar or function, to convert this quantity to a voltage. It also specifies a minimum and maximum value for the variable, to be used in validation.

The unit range is used for conversion and the saftey range is for sanity checking the output.

add(timeline, devices)

For connecting device information to a timeline.

Source code in src/wignertime/device.py

def add(timeline, devices):
    """
    For connecting device information to a `timeline`.
    """
    return wt_frame.join(timeline, devices)

check_correspondence(connections, devices)

Checks that the device and connection tables describe the same apparatus.

The two are deliberately separate — recalibrating a device and rewiring the apparatus are independent operations — but they are not independent vocabularies, and nothing used to notice when they drifted apart. A single transposed letter in a device name left the variable with no bounds at all, and check_within_range reads absent bounds as "a digital line" and skips it: 500 A passed on a coil declared +/-5 A (A14).

Both directions raise, because each catches a different half of that:

• an analogue variable with a connection but no device will be driven with neither a calibration nor limits;
• a device with no connection is the other end of the same typo. It is harmless in itself, since nothing reads it, but tolerating it is what let the first case happen quietly.

A digital line has no unit and needs no device, so it is not expected to have one.

Source code in src/wignertime/device.py

def check_correspondence(connections, devices):
    """
    Checks that the `device` and `connection` tables describe the same apparatus.

    The two are deliberately separate — recalibrating a device and rewiring the apparatus
    are independent operations — but they are not independent *vocabularies*, and nothing
    used to notice when they drifted apart. A single transposed letter in a device name
    left the variable with no bounds at all, and `check_within_range` reads absent bounds
    as "a digital line" and skips it: 500 A passed on a coil declared +/-5 A (A14).

    Both directions raise, because each catches a different half of that:

    - an **analogue variable with a connection but no device** will be driven with neither
      a calibration nor limits;
    - a **device with no connection** is the other end of the same typo. It is harmless in
      itself, since nothing reads it, but tolerating it is what let the first case happen
      quietly.

    A digital line has no unit and needs no device, so it is not expected to have one.
    """
    names__connected = set(connections["variable"])
    names__calibrated = set(devices["variable"])

    analogue = {
        v for v in names__connected if wt_variable.unit(v) != "digital"
    } - names__calibrated
    orphaned = names__calibrated - names__connected

    if analogue or orphaned:
        raise ValueError(
            "\n".join(
                [
                    "The `device` and `connection` tables do not describe the same apparatus.",
                    "",
                    "  connected, analogue, but no device : {}".format(
                        sorted(analogue) or "none"
                    ),
                    "  a device, but nothing connected    : {}".format(
                        sorted(orphaned) or "none"
                    ),
                    "",
                    "An analogue channel without a device has neither a conversion nor"
                    " safety limits, and `check_within_range` cannot tell that from a"
                    " digital line. A device naming nothing connected is usually the"
                    " other end of the same misspelling.",
                ]
            )
        )
    return True

check_within_range(timeline, columns__bounds=['value__min', 'value__max'])

Considers whether the timeline values fall inside device safety ranges (see SCHEMA). Raises an error if not, naming every variable that offends rather than only the first.

A variable with no bounds at all is one that has no entry in devices – a digital line – and is skipped. A variable with only one bound is checked against that bound alone.

"Typically" until 2026-09-21: an analogue variable could reach here unbounded through a mistyped device name, and this function could not tell that from a digital line, so it passed it. check_correspondence, called from adwin.internal.add, now rules that out on the path to hardware (A14).

ASSUMES: That a value column is present, and that the timeline has already been joined to devices.

Source code in src/wignertime/device.py

def check_within_range(timeline, columns__bounds=["value__min", "value__max"]):
    """
    Considers whether the `timeline` `value`s fall inside device safety ranges (see SCHEMA). Raises an error if not, naming every variable that offends rather than only the first.

    A variable with no bounds at all is one that has no entry in `device`s – a digital line – and is skipped. A variable with only one bound is checked against that bound alone.

    "Typically" until 2026-09-21: an analogue variable could reach here unbounded through a mistyped device name, and this function could not tell that from a digital line, so it passed it. `check_correspondence`, called from `adwin.internal.add`, now rules that out on the path to hardware (A14).

    ASSUMES: That a `value` column is present, and that the timeline has already been joined to `device`s.
    """

    if not wt_frame.is_column_float(timeline["value"]):
        raise ValueError("Value column might not contain floats.")

    columns__missing = [c for c in columns__bounds if c not in timeline.columns]
    if columns__missing:
        raise ValueError(
            "Safety limits cannot be checked because the column(s) {} are absent. `device`s should be joined to the timeline before validation. Columns present: {}".format(
                columns__missing, list(timeline.columns)
            )
        )

    column__min, column__max = columns__bounds
    violations = []

    for variable, group in timeline.groupby("variable"):
        bound__min = group[column__min].iloc[0]
        bound__max = group[column__max].iloc[0]

        if wt_frame.isnull(bound__min) and wt_frame.isnull(bound__max):
            # No device entry for this variable, so there is nothing to check.
            continue

        if not wt_frame.isnull(bound__max):
            value__max = group["value"].max()
            if value__max > bound__max:
                violations.append((variable, value__max, "above", bound__max))

        if not wt_frame.isnull(bound__min):
            value__min = group["value"].min()
            if value__min < bound__min:
                violations.append((variable, value__min, "below", bound__min))

    if violations:
        raise ValueError(
            "The following variables were given values outside their device safety range. Please provide values only inside it:\n"
            + "\n".join("  {}: {} is {} the limit of {}".format(*v) for v in violations)
        )

    return True

new(*variable_toV_min_max)

Convenience for creating a table that specifies the conversion and range of device values ('variable', 'to_V', 'value__min' and 'value__max' columns). 'to_V' specifies the conversion and can either be a floating point factor or a function that takes the output from the givin units to a Voltage. The min-max limits outline the range that a user is allowed to specify (and will be validated before output). This allows for error-checking before passing values to real devices.

variables are used as the means by which connections and devices can be later connected.

If the 'value__min/max' columns are not specified, then they will be taken as +/- infinity.

vfmm: e.g. "coil__compensation_X__A", 3/10., -3.0, 3.0, or ["coil__compensation_Y__A", 0.333, -3.0], ["coil__MOT_lower__A", , -5, 5], ["coil__MOT_upper__A", lambda x: x - 100,-5, 5],

Source code in src/wignertime/device.py

def new(*variable_toV_min_max) -> wt_frame.CLASS:
    """
    Convenience for creating a table that specifies the conversion and range of device values ('variable', 'to_V', 'value__min' and 'value__max' columns). 'to_V' specifies the conversion and can either be a floating point factor or a function that takes the output from the givin units to a Voltage. The min-max limits outline the range that a user is allowed to specify (and will be validated before output). This allows for error-checking before passing values to real devices.

    `variable`s are used as the means by which `connection`s and `device`s can be later connected.

    If the 'value__min/max' columns are not specified, then they will be taken as +/- infinity.

    vfmm:
    e.g.
        "coil__compensation_X__A", 3/10., -3.0, 3.0,
    or
        ["coil__compensation_Y__A", 0.333, -3.0],
        ["coil__MOT_lower__A", <function>, -5, 5],
        ["coil__MOT_upper__A", lambda x: x - 100,-5, 5],
    """

    def process_input(args):
        l = len(args)
        if l == 2:
            return np.concatenate([args, [-np.inf, np.inf]])
        elif l == 3:
            return np.concatenate([args, [np.inf]])
        elif l == 4:
            return args  # No changes
        else:
            raise ValueError(
                f"Invalid list of devices {args}: the number of arguments should be less than 5."
            )

    # A purely digital apparatus has no devices at all, and that is a legitimate table
    # rather than a mistake -- `check_correspondence` expects exactly this where nothing
    # analogue is connected. Without the early return, `ensure_2d(())` yields one empty
    # row and the arity check rejects it.
    if not variable_toV_min_max:
        return wt_frame.new_schema([], SCHEMA)

    input4 = [process_input(args) for args in wt_util.ensure_2d(variable_toV_min_max)]

    # Narrowed from a bare `except:` that discarded the cause and reported only
    # "=== Input to 'device' not well formatted ===". It would also have swallowed the
    # name check below (A14). The same fault was fixed in `adwin/connection.py` on
    # 2026-09-11.
    try:
        new = wt_frame.new_schema(input4, SCHEMA)
    except (TypeError, ValueError, KeyError) as e:
        raise ValueError(
            "Input to `device.new` is not well formatted: {}. Each device is"
            " `[variable, to_V]`, `[variable, to_V, value__min]` or"
            " `[variable, to_V, value__min, value__max]`.".format(e)
        ) from e

    # Parity with `connection.new`, which has always refused a malformed name. Catches a
    # name that could not denote a variable at all; it cannot catch one that is well
    # formed but denotes nothing, which is what `check_correspondence` is for (A14).
    _ensure_valid_names(new)

    # convert dtype to float if possible (i.e. no functions)
    if pd.to_numeric(new["to_V"], errors="coerce").notna().all():
        new["to_V"] = new["to_V"].astype(float)

    return new

file

For PC-level loading and saving of stored timelines.

load(path)

Reads the given file into memory.

Source code in src/wignertime/file.py

def load(path: str | Path) -> wt_frame.CLASS:
    """
    Reads the given file into memory.
    """
    path = Path(path).expanduser()

    if not path.exists():
        raise FileNotFoundError(path)

    suffix = path.suffix.lower()

    match suffix:
        case ".pkl" | ".pickle":
            df = wt_frame.read_pickle(path)

        case ".csv":
            df = wt_frame.read_csv(path)

        case ".json":
            df = wt_frame.read_json(path)
            if "time" in df.columns and df["time"].dtype == object:
                df["time"] = df["time"].astype(float)  # "-inf" and "inf" (#154)

        case ".parquet":
            if not (_has_module("pyarrow") or _has_module("fastparquet")):
                raise ImportError(
                    "Reading parquet requires 'pyarrow' or 'fastparquet'."
                )
            df = wt_frame.read_parquet(path)

        case ".feather":
            if not _has_module("pyarrow"):
                raise ImportError("Reading feather requires 'pyarrow'.")
            df = wt_frame.read_feather(path)

        case _:
            raise ValueError(f"Unsupported file suffix: {suffix}")

    # Parquet, JSON and feather return a missing value as `None` where the in-memory
    # timeline holds `nan`; CSV and pickle keep `nan`. Settle on one, so that a
    # round-trip is faithful whichever format was chosen.
    return wt_frame.normalise_nulls(df)

save(df, path=None)

Writes the given timeline to file, according to convenient features.

--------
- If `path` has a recognized suffix, save in that format if the required
  dependency is available.
- If `path` has no suffix:
    - save as parquet if possible
    - otherwise save as pickle
- If `path` is a directory (or looks like a directory), generate a filename
  from the caller's variable name for `df`, falling back to "timeline".
- If the target already exists, append __002, __003, ... before the suffix.

Returns

Path
    The actual path written.

Supported suffixes

.parquet, .feather, .pickle, .csv, .json

Source code in src/wignertime/file.py

def save(df: wt_frame.CLASS, path: str | Path | None = None) -> Path:
    """
    Writes the given timeline to file, according to convenient features.

        --------
        - If `path` has a recognized suffix, save in that format if the required
          dependency is available.
        - If `path` has no suffix:
            - save as parquet if possible
            - otherwise save as pickle
        - If `path` is a directory (or looks like a directory), generate a filename
          from the caller's variable name for `df`, falling back to "timeline".
        - If the target already exists, append __002, __003, ... before the suffix.

        Returns
        -------
        Path
            The actual path written.

        Supported suffixes
        ------------------
        .parquet, .feather, .pickle, .csv, .json
    """
    if path is None:
        path = Path.cwd()
    path = Path(path).expanduser()

    writers = _available_writers()

    is_existing_dir = path.exists() and path.is_dir()
    looks_like_dir = str(path).endswith(("/", "\\")) or (
        not path.suffix and not path.exists()
    )

    if is_existing_dir or looks_like_dir:
        path.mkdir(parents=True, exist_ok=True)
        stem = _infer_caller_name(df) or "timeline"

        if ".parquet" in writers:
            ext = ".parquet"
        elif ".pkl" in writers:
            ext = ".pkl"
        elif ".pickle" in writers:
            ext = ".pickle"
        else:
            # Fallback to the first available writer if needed
            ext = next(iter(writers))

        path = path / f"{_sanitize_stem(stem)}{ext}"

    elif not path.suffix:
        if ".parquet" in writers:
            ext = ".parquet"
        elif ".pkl" in writers:
            ext = ".pkl"
        elif ".pickle" in writers:
            ext = ".pickle"
        else:
            ext = next(iter(writers))

        path = path.with_suffix(ext)

    suffix = path.suffix.lower()
    if suffix not in writers:
        supported = ", ".join(sorted(writers))
        raise ValueError(
            f"Unsupported file suffix {suffix!r}. Supported writable formats: {supported}"
        )

    path.parent.mkdir(parents=True, exist_ok=True)
    path = _next_available_path(path)

    writer = writers[suffix]
    export_df = _stringify_callables_for_export(df, suffix)
    writer(export_df, path)

    return path

national_instruments

This is currently not implemented, but if it is of interest to you then please don't hesitate to get in touch: thomas.clark@wigner.hun-ren.hu .

ramp_function

ATTRIBUTE__POINTS = '__wigner_time_points__' module-attribute

How many time-value pairs a ramp function interpolates between.

This belongs to the function, not to the caller expanding it: tanh is defined by two points, and an interpolation wanting interior control points would be defined by more. expand used to take the number as an argument instead (num__bounds), which meant it could be given a value the data did not match, and could not be told apart from the caller's other keywords. See KNOWN_ISSUES B6.

The name it replaces was wrong as well as misplaced: for two points "bounds" is exact, since start and end are the boundaries -- but that is the one case where the number need not be stated at all. A third point is an interior control point, not a bound.

POINTS__DEFAULT = 2 module-attribute

Assumed of a ramp function that does not say, which is every hand-written one.

linear(origin, terminus, time_resolution=None)

A series of [time, value] pairs according to the line defined by two points and the time resolution.

time_resolution left None is filled by whoever expands the ramp; see _require_resolution.

Source code in src/wignertime/ramp_function.py

@with_points(2)
def linear(
    origin: list[float],
    terminus: list[float],
    time_resolution: float | None = None,
):
    """
    A series of [time, value] pairs according to the line defined by two points and the time resolution.

    `time_resolution` left `None` is filled by whoever expands the ramp; see
    `_require_resolution`.
    """
    _require_resolution("linear", time_resolution)
    t1, v1 = origin
    t2, v2 = terminus
    m = (v2 - v1) / (t2 - t1)
    times = np.arange(t1, t2, time_resolution)

    return np.array([times, m * (times - t1) + v1]).transpose()

points(f)

How many time-value pairs f interpolates between. See ATTRIBUTE__POINTS.

Source code in src/wignertime/ramp_function.py

def points(f) -> int:
    """
    How many time-value pairs `f` interpolates between. See `ATTRIBUTE__POINTS`.
    """
    return getattr(f, ATTRIBUTE__POINTS, POINTS__DEFAULT)

tanh(origin, terminus, time_resolution=None, sharpness=3)

Hyperbolic tan, with a call signature adapted for practical timeline population.

origin/terminus are time-value pairs time_resolution left None is filled by whoever expands the ramp: bind one with functools.partial(tanh, time_resolution=...) for a ramp of its own resolution (#65). See _require_resolution.

sharpness is a measure of how linear the 'slope' of the function is around the halfway point and in practice is used for easing transitions between the end-points. For example, sharpness ~0 (!=0) gives a linear ramp between origin and terminus, whereas large values approximate a step-function at the half-way point. In-between these values, the ramps returned will start and end gradually, with a linear movement in the middle.

Source code in src/wignertime/ramp_function.py

@with_points(2)
def tanh(
    origin: list[float],
    terminus: list[float],
    time_resolution: float | None = None,
    sharpness: float = 3,
):
    """
    Hyperbolic tan, with a call signature adapted for practical timeline population.

    origin/terminus are time-value pairs
    `time_resolution` left `None` is filled by whoever expands the ramp: bind one with
    `functools.partial(tanh, time_resolution=...)` for a ramp of its own resolution (#65).
    See `_require_resolution`.

    `sharpness` is a measure of how linear the 'slope' of the function is around the halfway point and in practice is used for easing transitions between the end-points. For example, sharpness ~0 (!=0) gives a linear ramp between `origin` and `terminus`, whereas large values approximate a step-function at the half-way point. In-between these values, the ramps returned will start and end gradually, with a linear movement in the middle.
    """

    _require_resolution("tanh", time_resolution)
    t1, v1 = origin
    t2, v2 = terminus
    times = wt_util.range__inclusive(t1, t2, time_resolution)

    return np.array(
        [times, _normalize(v1, v2, _tanh__scaled(times, sharpness))]
    ).transpose()

with_points(number)

Declare how many time-value pairs a ramp function interpolates between.

Decorating is optional: an undeclared function is taken to want POINTS__DEFAULT, which keeps a user's lambda origin, terminus, time_resolution: ... working without ceremony.

Source code in src/wignertime/ramp_function.py

def with_points(number):
    """
    Declare how many time-value pairs a ramp function interpolates between.

    Decorating is optional: an undeclared function is taken to want `POINTS__DEFAULT`,
    which keeps a user's `lambda origin, terminus, time_resolution: ...` working without
    ceremony.
    """

    def decorate(f):
        setattr(f, ATTRIBUTE__POINTS, number)
        return f

    return decorate

timeline

Multiple layers of abstraction: - operational (time sequence: probe-on, probe-off etc.) – this should go to notebooks / experiment specific packages - variable (time sequence of independent degrees of freedom: AOM_probe_power 5V) - ADwin (ADwin-specific details)

It is a goal to be able to go up and down through the layers of abstraction.

ANCHOR = wt_config.ANCHOR module-attribute

An origin at the most recent anchor. See config.Origin.

LAST = wt_config.LAST module-attribute

An origin at the latest entry of the timeline. See config.Origin.

VARIABLE = wt_config.VARIABLE module-attribute

An origin at each variable's own most recent entry, in time or value. See config.Origin.

as_deferred = wt_util.mark_deferred module-attribute

Mark a user-written function as a deferred timeline function, so that stack will accept it as a constituent. See stack.

noop = wt_util.mark_deferred(lambda timeline, **kwargs: timeline) module-attribute

A stack constituent that contributes nothing, for the branch of a conditional that should add no rows.

Not funcy.identity: it has to carry the deferred tag, and tagging a shared library function would mark it for every other user of funcy. Taking **kwargs also means it survives a stack that forwards keywords -- identity did not, and raised TypeError: identity() got an unexpected keyword argument 'context'.

__getattr__(name)

Say what replaced a public name that is gone, rather than only that it is missing.

create initialised a timeline, and was the one core function a stack could not begin with anything but. Since #85 a stack composes stages only, and the first rows of a timeline are an update applied to an empty timeline by to_timeline.

Source code in src/wignertime/timeline.py

def __getattr__(name):
    """
    Say what replaced a public name that is gone, rather than only that it is missing.

    `create` initialised a timeline, and was the one core function a `stack` could not
    begin with anything but. Since #85 a `stack` composes stages only, and the first rows
    of a timeline are an `update` applied to an empty timeline by `to_timeline`.
    """
    if name == "create":
        raise AttributeError(
            "\n".join(
                [
                    "`create` is gone (#85). The first rows of a timeline are an `update`"
                    " like any other, applied to an empty timeline by `to_timeline`:",
                    "",
                    '    to_timeline(update(AOM__MOT=1, shutter__MOT=0, time=0.0, context="init"))',
                    "",
                    "On an empty timeline the origin is absolute zero, and the rows must"
                    " name their context (#156). In a `stack`, write the `update` itself.",
                ]
            )
        )
    raise AttributeError("module {!r} has no attribute {!r}".format(__name__, name))

anchor(time, *, context=wt_config.CONTEXT__INFER, origin=wt_config.ORIGIN__INFER)

Creates a special, non-physical variable (will never have a matching connection), that can be used for time references, particularly within individual contexts.

This can be very convenient in the context of ramps, where the starting and ending times are often built around a hypothetical point in time, due to physical switching speeds.

time is required. There is no sensible default: it is a displacement from whatever the origin resolves to, and the two readings a default would have to choose between are genuinely different instants (see below).

Where the anchor lands

By default the origin is the most recent anchor where one exists and the last entry otherwise, so time is normally a duration measured from the end of the preceding stage. That is what makes stages chain: a stage may write rows past its own closing anchor -- optical_pumping reinitialises shutters 0.1 s later -- without dragging the next stage along with them.

To mark the end of everything written so far instead, ask for it explicitly:

anchor(0.0, origin=LAST)       # here, at the last entry in the timeline
anchor(0.0)                    # here, at the most recent anchor

The two coincide until some stage writes past its own anchor, and then they do not: in the shipped demo they differ by ~0.1 s from optical_pumping onwards. Which one is meant is therefore worth stating at the call site rather than defaulting.

NB. - Anchors are automatically numbered, for 'global' referencing, but these numbers are not necessary in normal use.

Source code in src/wignertime/timeline.py

def anchor(
    time,
    *,
    context=wt_config.CONTEXT__INFER,
    origin=wt_config.ORIGIN__INFER,
) -> Callable:
    """
    Creates a special, non-physical `variable` (will never have a matching `connection`), that can be used for time references, particularly within individual `context`s.

    This can be very convenient in the context of `ramp`s, where the starting and ending times are often built around a hypothetical point in time, due to physical switching speeds.

    `time` is required. There is no sensible default: it is a displacement from whatever
    the `origin` resolves to, and the two readings a default would have to choose
    between are genuinely different instants (see below).

    *Where the anchor lands*

    By default the `origin` is the most recent anchor where one exists and the last
    entry otherwise, so `time` is normally a duration measured **from the end of the
    preceding stage**. That is what makes stages chain: a stage may write rows past its
    own closing anchor -- `optical_pumping` reinitialises shutters 0.1 s later -- without
    dragging the next stage along with them.

    To mark the end of everything written so far instead, ask for it explicitly:

        anchor(0.0, origin=LAST)       # here, at the last entry in the timeline
        anchor(0.0)                    # here, at the most recent anchor

    The two coincide until some stage writes past its own anchor, and then they do not:
    in the shipped demo they differ by ~0.1 s from `optical_pumping` onwards. Which one
    is meant is therefore worth stating at the call site rather than defaulting.

    NB.
    - Anchors are automatically numbered, for 'global' referencing, but these numbers are not necessary in normal use.
    """
    # NOTE: Makes use of a global variable (LABEL__ANCHOR).

    if time is None:
        raise TypeError(
            "\n".join(
                [
                    "`anchor` requires `time`, a displacement from whatever `origin`"
                    " resolves to.",
                    "",
                    "    anchor(0.0)                    # at the most recent anchor",
                    "    anchor(0.0, origin=LAST)       # at the last entry so far",
                    "    anchor(duration)               # `duration` after the"
                    " preceding stage",
                ]
            )
        )
    if not np.isfinite(time):
        # An anchor marks an instant, and ±∞ is none: it is before or after the run
        # (#154), where nothing can be placed relative to anything.
        raise ValueError(
            "`anchor` was given `time={}`. An anchor marks an instant to place stages"
            " relative to, and ±∞ is before or after the run, not an instant"
            " (#154).".format(time)
        )

    inherit.resolve(context)  # refused where it is written, not where it is applied

    return wt_util.stage(
        _anchor, anchor, dict(time=time, context=context, origin=origin)
    )

cascade(*fs, **kws)

Similarly to stack, a convenience that combines an arbitrary chain of functions with an arbitrary selection of associated keywords.

kws are routed to the associated functions by prefixing, e.g. cascade(MOT, molasses, MOT_duration=1.0) creates a stack of MOT and molasses, with duration=1.0 passed into the MOT function before evaluation.

The motivation for this feature is that different experimental contexts should be built modularly, but, at final composition, the user often just wants a single point of contact to add/change nested variables.

How this differs from stack, which is easy to get wrong

stack takes stages that have already been called; cascade takes the stage functions themselves and calls them::

stack(MOT(duration=15), molasses())               # called here
cascade(MOT, molasses, MOT_duration=15)           # called by cascade

In the first, MOT(duration=15) has had every argument but timeline supplied, and what it returns is a function of the timeline alone -- partial application, not currying, since the remaining argument is supplied in one call rather than one at a time. stack then threads the timeline through that chain.

In the second, nothing has been called: cascade routes MOT_duration=15 to MOT, calls it, and hands the results to stack. So a stage reaches cascade bare and reaches stack applied, and the two are not interchangeable -- writing stack(MOT) raises (see _ensure_stackable), and cascade(MOT(...)) fails because the result takes no keywords to route.

What it returns

A stage, like stack: to_timeline(cascade(...)) is the experiment's timeline, and a cascade is itself stackable, so one beginning mid-experiment composes like any other stage.

Routing is strict: a keyword that names no stage, or that names one but is not a parameter of it, raises rather than being dropped. The alternative -- letting an unrouted keyword broadcast to every stage -- was considered and rejected, because every stage would turn it into rows, and into different kinds of row depending on whether the stage ends in an update or a ramp.

Strictness is not configured but derived, from whether the target has **kwargs. That is only safe because the operation layer confines the open namespace to default_state and the two functions that wrap it; a stage that collects **kwargs is, correctly, still permissive here.

Source code in src/wignertime/timeline.py

def cascade(*fs: Callable, **kws) -> Callable | wt_frame.CLASS:
    """
    Similarly to `stack`, a convenience that combines an arbitrary chain of functions with an arbitrary selection of associated keywords.

    `kws` are routed to the associated functions by prefixing, e.g. `cascade(MOT, molasses, MOT_duration=1.0)` creates a `stack` of `MOT` and `molasses`, with `duration=1.0` passed into the `MOT` function before evaluation.

    The motivation for this feature is that different experimental contexts should be built modularly, but, at final composition, the user often just wants a single point of contact to add/change nested variables.

    *How this differs from `stack`, which is easy to get wrong*

    `stack` takes stages that have **already been called**; `cascade` takes the stage
    functions **themselves** and calls them::

        stack(MOT(duration=15), molasses())               # called here
        cascade(MOT, molasses, MOT_duration=15)           # called by cascade

    In the first, `MOT(duration=15)` has had every argument but `timeline` supplied, and
    what it returns is a function of the timeline alone -- partial application, not
    currying, since the remaining argument is supplied in one call rather than one at a
    time. `stack` then threads the timeline through that chain.

    In the second, nothing has been called: `cascade` routes `MOT_duration=15` to `MOT`,
    calls it, and hands the results to `stack`. So a stage reaches `cascade` bare and
    reaches `stack` applied, and the two are not interchangeable -- writing
    `stack(MOT)` raises (see `_ensure_stackable`), and `cascade(MOT(...))` fails because
    the result takes no keywords to route.

    *What it returns*

    A stage, like `stack`: `to_timeline(cascade(...))` is the experiment's timeline, and
    a `cascade` is itself stackable, so one beginning mid-experiment composes like any
    other stage.

    Routing is **strict**: a keyword that names no stage, or that names one but is not a
    parameter of it, raises rather than being dropped. The alternative -- letting an
    unrouted keyword broadcast to every stage -- was considered and rejected, because
    every stage would turn it into rows, and into different *kinds* of row depending on
    whether the stage ends in an `update` or a `ramp`.

    Strictness is not configured but derived, from whether the target has `**kwargs`.
    That is only safe because the operation layer confines the open namespace to
    `default_state` and the two functions that wrap it; a stage that collects `**kwargs`
    is, correctly, still permissive here.
    """
    # NOTE: keyed by `__name__`, so a stage appearing twice receives the same keywords
    # both times. That is relied upon; see `KNOWN_ISSUES.md` §C.
    for f in fs:
        _ensure_cascadable(f)

    stages__by_name = {f.__name__: f for f in fs}
    names__by_length = sorted(stages__by_name, key=len, reverse=True)

    args__dict = {}
    unroutable = {}
    for key, value in kws.items():
        routed, near_misses = _route_keyword(key, names__by_length, stages__by_name)
        if routed is None:
            unroutable[key] = near_misses
        else:
            name, parameter = routed
            args__dict.setdefault(name, {})[parameter] = value

    if unroutable:
        raise TypeError(_message__unroutable(unroutable, names__by_length))

    # Apply keywords to function stack
    return stack(*[f(**args__dict.get(f.__name__, {})) for f in fs])

context_info(timeline)

Useful data (currently 'variables' and 'times') concerning every context. The result is a dictionary, indexed by context.

e.g. To get the start and end times of the 'MOT' context, call context_info(timeline)['MOT']['times].

Source code in src/wignertime/timeline.py

def context_info(timeline):
    """
    Useful data (currently 'variables' and 'times') concerning every context. The result is a dictionary, indexed by context.

    e.g. To get the start and end times of the 'MOT' context, call `context_info(timeline)['MOT']['times]`.
    """
    if {"context", "time", "variable"}.issubset(timeline.columns):
        tlg = timeline.groupby("context")
        return {
            k: {
                "variables": tlg["variable"].agg(set).to_dict()[k],
                "times": tlg["time"]
                .agg(["first", "last"])
                .apply(list, axis=1)
                .to_dict()[k],
            }
            for k in tlg.groups.keys()
        }

    else:
        return None

expand(timeline, **function_args)

Converts the functions marked in the timeline into individual rows, i.e. applies the functions to the given data.

This is generally a 'one-way' operation and so should only be carried out before the timeline is implemented on a device.

It takes a timeline and returns one, and is not a stage (#85, C7 item 6). Unlike update, ramp and anchor, which add rows, it transforms the whole timeline it is given, so inside a stack it expanded every ramp built so far and dropped function, and the expand in adwin.core.convert then did nothing.

A keyword given to expand is a default, never an override (#65, C7 item 7). It reaches only the ramp functions that leave it unstated (util.function__defaults), so expand(timeline, time_resolution=...) samples the ramps that bind no resolution, and a ramp that binds one keeps it: function=functools.partial(tanh, time_resolution=1e-3) is a coarse ramp in a timeline expanded at the cycle period. That binding used to be overridden silently, since a partial still declares the keyword. A ramp binding none, expanded with none given, raises.

How many rows make up one ramp is read from the ramp function itself (ramp_function.points), not passed in. It used to be the num__bounds argument, which could be given a number the data did not match and was named for the two-point case in which it need not have been given at all -- start and end are the bounds only while there is nothing between them (B6).

Source code in src/wignertime/timeline.py

def expand(timeline, **function_args) -> wt_frame.CLASS:
    """
    Converts the functions marked in the timeline into individual rows, i.e. applies the functions to the given data.

    This is generally a 'one-way' operation and so should only be carried out before the timeline is implemented on a device.

    It takes a timeline and returns one, and is not a stage (#85, C7 item 6). Unlike
    `update`, `ramp` and `anchor`, which add rows, it transforms the whole timeline it is
    given, so inside a `stack` it expanded every ramp built so far and dropped `function`,
    and the `expand` in `adwin.core.convert` then did nothing.

    **A keyword given to `expand` is a default, never an override** (#65, C7 item 7). It
    reaches only the ramp functions that leave it unstated (`util.function__defaults`),
    so `expand(timeline, time_resolution=...)` samples the ramps that bind no resolution,
    and a ramp that binds one keeps it: `function=functools.partial(tanh,
    time_resolution=1e-3)` is a coarse ramp in a timeline expanded at the cycle period.
    That binding used to be overridden silently, since a `partial` still declares the
    keyword. A ramp binding none, expanded with none given, raises.

    How many rows make up one ramp is read from the ramp function itself
    (`ramp_function.points`), not passed in. It used to be the `num__bounds` argument,
    which could be given a number the data did not match and was named for the two-point
    case in which it need not have been given at all -- start and end are the *bounds*
    only while there is nothing between them (B6).
    """
    timeline = _given(timeline, "expand")

    if "num__bounds" in function_args:
        raise TypeError(
            "`expand` no longer takes `num__bounds`: how many points a ramp is made of"
            " is a property of its interpolating function, and is read from it. Declare"
            " it with `ramp_function.with_points(n)` if you are writing one."
        )

    if "function" not in timeline.columns:
        return timeline

    # Labels are the written positions from here on, which is what each ramp's rows are
    # put back at, and what makes the label-based `drop` below exact.
    timeline = timeline.reset_index(drop=True)
    _mask_fs = timeline["function"].notna()
    _dff = timeline[_mask_fs].sort_values(by=["variable", "time"])
    _indices_drop = _dff.index

    # For adding back in the value of other columns, based on the first row, like `context` etc. Written this way to allow for more, unknown columns to continue.
    _columns__keep = _dff.columns.drop(["time", "value", "variable", "function"])

    # Grouped per variable rather than by striding the whole frame. The old global stride
    # meant one variable with an odd number of rows silently misaligned the pairing of
    # *every* variable after it, and the failure surfaced as a bare
    # `ValueError: not enough values to unpack` from the tuple assignment, naming
    # nothing (B6). Per variable, the arithmetic is local and the offender has a name.
    _inds__start = []
    _dfs = []
    for variable, rows in _dff.groupby("variable", sort=False):
        points__required = wt_ramp_function.points(rows["function"].iloc[0])

        if len(rows) % points__required:
            raise ValueError(
                "\n".join(
                    [
                        "{} has {} ramp row(s), which is not a whole number of ramps:"
                        " {} makes each one out of {}.".format(
                            variable,
                            len(rows),
                            getattr(
                                rows["function"].iloc[0],
                                "__name__",
                                repr(rows["function"].iloc[0]),
                            ),
                            points__required,
                        ),
                        "",
                        "A ramp row is one carrying a `function`. Rows written by hand"
                        " into that column, or a `function` left on a row that was"
                        " meant to be a plain entry, are the usual causes.",
                    ]
                )
            )

        for i in range(0, len(rows), points__required):
            group = rows.iloc[i : i + points__required]
            _inds__start.append(group.index[0])

            # A default, never an override: each keyword reaches only the functions that
            # leave it unstated, so a ramp keeps a resolution it binds (#65, C7 item 7).
            function = group["function"].iloc[0]
            if "time_resolution" not in function_args and (
                "time_resolution" in wt_util.parameters__unstated(function)
            ):
                raise ValueError(
                    "{}'s ramp at {} s binds no `time_resolution`, and `expand` was given"
                    " none. Give one, `expand(timeline, time_resolution=1e-4)`, or bind"
                    " one into the ramp: `function=functools.partial(tanh,"
                    " time_resolution=1e-4)`. `adwin.core.convert` gives the cycle"
                    " period.".format(variable, group["time"].iloc[0])
                )
            func = wt_util.function__defaults(function, **function_args)

            # The internal constructor, because this is the one caller that assembles
            # rows rather than being handed them: `update` takes keywords only.
            _dfs.append(
                _populate_timeline(
                    [variable, func(*group[["time", "value"]].values)],
                    context=group["context"].iloc[0],
                ).assign(**group.iloc[0][_columns__keep].to_dict())
            )

    # Dropped into a new frame rather than in place. Every other function here returns a
    # new timeline and leaves its argument alone, and the "description is data" story
    # depends on a frame not changing under whoever is holding it -- `expand` was the one
    # exception, and it took both the ramp rows and the `function` column with it (B5).
    #
    # `adwin.core.convert` was unharmed only by accident of pipeline order:
    # `remove_unconnected_variables` runs first and hands `expand` a fresh frame.
    timeline = timeline.drop(index=_indices_drop).drop(columns=["function"])

    # Each ramp's rows go back where the ramp was written: before the first row kept from
    # after its start, in the order the ramps were written. Among a variable's rows at
    # one instant the last written is in effect, so this is not cosmetic. The ramps used
    # to land after rows written after them, and a ramp's end then superseded the
    # `update` written to supersede it (A18).
    ramps = sorted(zip(_inds__start, _dfs), key=lambda ramp: ramp[0])
    return wt_frame.insert_dataframes(
        timeline,
        list(timeline.index.searchsorted([start for start, _ in ramps])),
        [rows for _, rows in ramps],
    )

ramp(*, duration=None, time=None, time2=None, context=wt_config.CONTEXT__INFER, origin=wt_config.ORIGIN__INFER, origin2=[wt_config.VARIABLE, 0.0], function=wt_ramp_function.tanh, **vtvc_dict)

Convenient ways of defining pairs of points and a function!

A ramp defines ranges of values for each variable across time from a beginning time-value pair to an ending time-value pair. Primarily, for the sake of switching analogue devices on and off in a controllable way. Although a ramp can be as simple as a linear value progression from start to end, the default function for a timeline is hyperbolic tan. This allows the user to soften the value gradient at the beginning and end of the function.

ramp has a slightly different interface to update. **vtvc_dict follows that of update; to specify the points manually (in big lists), use update.

A ramp is two points and a function between them: its start, which is always where the variable is at that instant, and its end, which is written.

Examples of calling ramp Normally, it will look something like ` tl.stack( tl.ramp( coil__compensation_X__A=0.0, coil__compensation_Y__A=0.0, coil__MOT_lower_plus__A=0.0, coil__MOT_upper_plus__A=0.0,

    duration=duration,
    context="final_ramps"))

` The variables are given end values independently and other options collectively. By default, the starting time is also inferred from the previous timeline and so chains of operations can be built up conveniently.

For simpler ramps, it can still be easier, like in update, to supply everything in a list, e.g. tl.ramp(lockbox__MOT__MHz=[500e-3,0.0]) or tl.ramp(lockbox__MOT__MHz=[500e-3, 0.0, "final_ramps"]) - if you want a new context. This works because by default the ending time is relative to the starting time (see the origin keyword argument), such that 't_end' and 'duration' are the same.

A ramp always starts where its variable is (#142, 2026-09-29). A start value that differs from the current one is a step hidden inside a ramp, the discontinuity a ramp exists to avoid. A jump that is wanted is an update before the ramp, where it shows in the table::

stack(update(coil__A=1.0), ramp(coil__A=3.0, duration=0.1))

So a start is never written: the 2-D form that stated one is refused, and so is a value in the value slot of origin. A variable not yet set has to be set first. Nor may a ramp start or end inside another ramp of the same variable (#157).

origin places the start in time. It defaults to wt_config.INFER, the anchor-then-last chain, and None means the same; origin=0.0 is absolute time, as for update.

context defaults to wt_config.INFER, and None means the same: an unstated row inherits the latest context of timeline. Every row has a context (#156), so context="" is refused.

NOTE: duration is a human-readable convenience for normal API usage. This is because the temporal origin of the second point is almost always in reference to the first point. Where there is a conflict, time2 will have supremacy. It is time2 because origin2 places the same point, the end.

Source code in src/wignertime/timeline.py

def ramp(
    *,
    duration=None,
    time=None,
    time2=None,
    context=wt_config.CONTEXT__INFER,
    origin=wt_config.ORIGIN__INFER,
    origin2=[wt_config.VARIABLE, 0.0],
    function=wt_ramp_function.tanh,
    **vtvc_dict,
) -> Callable:
    """
    Convenient ways of defining pairs of points and a function!

    A `ramp` defines ranges of values for each variable across time from a beginning time-value pair to an ending time-value pair. Primarily, for the sake of switching analogue devices on and off in a controllable way. Although a `ramp` can be as simple as a linear `value` progression from start to end, the default function for a `timeline` is hyperbolic tan. This allows the user to soften the value gradient at the beginning and end of the function.

    `ramp` has a slightly different interface to `update`. `**vtvc_dict` follows that of `update`; to specify the points manually (in big lists), use `update`.

    A ramp is two points and a function between them: its start, which is always where the
    variable is at that instant, and its end, which is written.

    *Examples of calling ramp*
    Normally, it will look something like
    `
    tl.stack(
        tl.ramp(
            coil__compensation_X__A=0.0,
            coil__compensation_Y__A=0.0,
            coil__MOT_lower_plus__A=0.0,
            coil__MOT_upper_plus__A=0.0,

            duration=duration,
            context="final_ramps"))
    `
    The variables are given end values independently and other options collectively. By default, the starting time is also inferred from the previous timeline and so chains of operations can be built up conveniently.

    For simpler ramps, it can still be easier, like in `update`, to supply everything in a list, e.g.
    `tl.ramp(lockbox__MOT__MHz=[500e-3,0.0])`
    or
    `tl.ramp(lockbox__MOT__MHz=[500e-3, 0.0, "final_ramps"])` - if you want a new `context`.
    This works because by default the ending time is relative to the starting time (see the `origin` keyword argument), such that 't_end' and 'duration' are the same.

    **A ramp always starts where its variable is** (#142, 2026-09-29). A start value that
    differs from the current one is a step hidden inside a ramp, the discontinuity a ramp
    exists to avoid. A jump that is wanted is an `update` before the ramp, where it shows
    in the table::

        stack(update(coil__A=1.0), ramp(coil__A=3.0, duration=0.1))

    So a start is never written: the 2-D form that stated one is refused, and so is a
    value in the value slot of `origin`. A variable not yet set has to be set first.
    Nor may a ramp start or end inside another ramp of the same variable (#157).

    `origin` places the start in time. It defaults to `wt_config.INFER`, the
    anchor-then-last chain, and `None` means the same; `origin=0.0` is absolute time, as
    for `update`.

    `context` defaults to `wt_config.INFER`, and `None` means the same: an unstated row
    inherits the latest context of `timeline`. Every row has a context (#156), so
    `context=""` is refused.

    NOTE: `duration` is a human-readable convenience for normal API usage. This is because the temporal origin of the second point is almost always in reference to the first point. Where there is a conflict, `time2` will have supremacy. It is `time2` because `origin2` places the same point, the end.
    """
    _refuse_timeline("ramp", vtvc_dict)
    inherit.resolve(context)  # refused where it is written, not where it is applied
    _refuse_value_origin(origin)
    _refuse_infinite_ends(time=time, time2=time2, duration=duration)

    return wt_util.stage(
        _ramp,
        ramp,
        dict(
            duration=duration,
            time=time,
            time2=time2,
            context=context,
            origin=origin,
            origin2=origin2,
            function=function,
            **vtvc_dict,
        ),
    )

stack(*stages, **kws)

Composes stages, in execution order, into one stage::

stage = stack(update(...), ramp(...), anchor(...))
timeline = to_timeline(stage, onto=timeline)

stack takes stages only and returns one. To apply it, use to_timeline.

Constituents must already have been called: stack(MOT(duration=15)), never stack(MOT), which raises. This is the opposite of cascade, which takes the stage functions themselves and calls them with the keywords routed to each.

Keywords given to stack are passed on to its constituents. This is particularly convenient for a shared context::

stack(update(...), ramp(...), anchor(...), context="MOT")

A forwarded keyword is a default for the constituents, not an override: it fills what a constituent left unstated and leaves what it stated alone, so stack(update(x=1, context="ADwin_Finish"), anchor(1.0), context="finalRamps") keeps x in its reserved context. The same holds between stacks: a stage is itself a stack, and a keyword given to an outer one reaches the stages inside it, where each stack's own keywords, stated closer, take precedence.

Source code in src/wignertime/timeline.py

def stack(*stages: Callable, **kws) -> Callable:
    """
    Composes stages, in execution order, into one stage::

        stage = stack(update(...), ramp(...), anchor(...))
        timeline = to_timeline(stage, onto=timeline)

    `stack` takes stages only and returns one. To apply it, use `to_timeline`.

    Constituents must **already have been called**: `stack(MOT(duration=15))`, never
    `stack(MOT)`, which raises. This is the opposite of `cascade`, which takes the stage
    functions themselves and calls them with the keywords routed to each.

    Keywords given to `stack` are passed on to its constituents. This is particularly
    convenient for a shared context::

        stack(update(...), ramp(...), anchor(...), context="MOT")

    A forwarded keyword is a **default** for the constituents, not an override: it fills
    what a constituent left unstated and leaves what it stated alone, so
    `stack(update(x=1, context="ADwin_Finish"), anchor(1.0), context="finalRamps")` keeps
    `x` in its reserved context. The same holds between stacks: a stage is itself a
    `stack`, and a keyword given to an outer one reaches the stages inside it, where each
    stack's own keywords, stated closer, take precedence.
    """

    for f in stages:
        _ensure_stackable(f)

    constituents = list(stages)

    keywords__available = _keywords_available(constituents)
    if keywords__available is not None:
        unplaced = sorted(k for k in kws if k not in keywords__available)
        if unplaced:
            raise TypeError(_message__unplaced(unplaced, keywords__available))

    # Where no constituent says what it consumes, a keyword that none of them can even
    # accept would otherwise be dropped below without a word.
    unaccepted = sorted(
        k for k in kws if not any(wt_util.accepts_keyword(f, k) for f in constituents)
    )
    if unaccepted:
        raise TypeError(_message__unplaced(unaccepted, keywords__available or set()))

    # One closure that threads the keywords through, rather than a `funcy.compose` of
    # single-argument wrappers. That composition took the timeline and nothing else, so a
    # keyword forwarded into a nested stack -- every stage is one -- raised a `TypeError`
    # naming a lambda (B10, #136). A constituent that takes no keywords, such as a
    # hand-written `lambda tline: ...`, is given none: the guard above has already made
    # sure that each keyword reaches something that uses it.
    def composed(timeline, **kws__outer):
        kws__all = {**kws__outer, **kws}
        for f in constituents:
            timeline = f(
                timeline,
                **{k: v for k, v in kws__all.items() if wt_util.accepts_keyword(f, k)},
            )
        return timeline

    return wt_util.mark_keywords(
        wt_util.mark_deferred(composed), keywords__available or ()
    )

to_timeline(stage, onto=None)

The one way from a stage to a timeline: stage applied to onto, or to an empty timeline when onto is not given.

A stage is what update, ramp, anchor, stack and cascade return, and what a user-defined stage returns: a function of a timeline. So it is written once, relative to its own beginning, and can be placed anywhere. A timeline is data -- the table that is plotted, archived and converted for the hardware. to_timeline is where the one becomes the other, and nothing else is::

timeline = to_timeline(cascade(init, MOT, molasses, finish, MOT_duration=15))

onto is first-class, not a convenience. Building a timeline can be expensive, so a parameter scan keeps its base as a table and places each variation onto it::

base = to_timeline(experiment)
shots = [to_timeline(imaging(delay), onto=base) for delay in delays]

to_timeline(b, onto=to_timeline(a)) is the same table as to_timeline(stack(a, b)).

On an empty timeline there is nothing to be relative to, so the first rows are placed in absolute time, and nothing to inherit a context from, so they must name theirs (#156).

Source code in src/wignertime/timeline.py

def to_timeline(stage: Callable, onto: wt_frame.CLASS | None = None) -> wt_frame.CLASS:
    """
    The one way from a stage to a timeline: `stage` applied to `onto`, or to an empty
    timeline when `onto` is not given.

    A *stage* is what `update`, `ramp`, `anchor`, `stack` and `cascade` return, and what
    a user-defined stage returns: a function of a timeline. So it is written once,
    relative to its own beginning, and can be placed anywhere. A timeline is data -- the
    table that is plotted, archived and converted for the hardware. `to_timeline` is
    where the one becomes the other, and nothing else is::

        timeline = to_timeline(cascade(init, MOT, molasses, finish, MOT_duration=15))

    `onto` is first-class, not a convenience. Building a timeline can be expensive, so a
    parameter scan keeps its base as a table and places each variation onto it::

        base = to_timeline(experiment)
        shots = [to_timeline(imaging(delay), onto=base) for delay in delays]

    `to_timeline(b, onto=to_timeline(a))` is the same table as `to_timeline(stack(a, b))`.

    On an empty timeline there is nothing to be relative to, so the first rows are
    placed in absolute time, and nothing to inherit a context from, so they must name
    theirs (#156).
    """
    if isinstance(stage, wt_frame.CLASS):
        raise TypeError(
            "`to_timeline` was given a timeline, which is one already. To add a stage"
            " to it, pass it as `onto`: `to_timeline(stage, onto=timeline)`."
        )
    _ensure_stackable(stage)

    onto = (
        _empty()
        if onto is None
        else wt_util.ensure_timeline(
            onto, "to_timeline", name__argument="onto", columns__required=_SCHEMA
        )
    )

    out = stage(onto)
    if not isinstance(out, wt_frame.CLASS):
        raise TypeError(
            "A stage must return a timeline, and {} returned {}.".format(
                getattr(stage, "__name__", repr(stage)), type(out).__name__
            )
        )
    return out

update(*, time=0.0, context=wt_config.CONTEXT__INFER, origin=wt_config.ORIGIN__INFER, **vtvc_dict)

Commands variables at instants: the rows of a timeline.

It returns a stage, for a stack and for to_timeline. The first rows of a timeline are an update like any other, applied to an empty timeline::

initial = to_timeline(update(AOM__MOT=1, shutter__MOT=0, time=0.0, context="init"))

Input grammar

A variable is named as a keyword, and followed by what it does::

update(AOM__MOT=<follows>)

where <follows> is one of

====================================== ========================================== value at time [time, value] [time, value, context] [[time, value], [time, value], ...] several instants for one variable ====================================== ==========================================

Several variables are given at once, and a computed set through **::

update(AOM__MOT=1, shutter__MOT=[0.1, 1, "MOT"])
update(**{name: value for name, value in ...})

time and context are defaults, not overrides — a variable stating its own keeps it. The keyword namespace is open by design, so an unrecognised keyword is read as a variable name rather than rejected (see the manuscript's sec:forwarding); that is what makes the injection idiom work, and it is why there is no second, positional way in to be confused with it.

Origin and context

origin defaults to wt_config.INFER: the anchor-then-last chain (config.ORIGIN__DEFAULTS) fills whichever slots are left unstated, so time is a duration from the end of the preceding stage. None means the same. For absolute time, write origin=0.0. On an empty timeline there is nothing to be relative to, and the origin is absolute zero.

time=-math.inf is before the run and time=math.inf after it (#154): the initial and the final state, in the contexts a backend reserves for them (ADwin_LowInit, ADwin_Finish). Neither is an instant, so nothing is placed relative to them, and until something is written at an instant the origin is absolute zero.

context defaults to wt_config.INFER too: an unstated row inherits the latest context of the timeline it joins, from a row at an instant. None means the same. Every row has a context (#156): on an empty timeline, or one holding only the state before the run, there is nothing to inherit, so the rows must name theirs, and context="" is refused. A row added to a finished timeline inherits from the last stage of the run, not from the final state at +∞.

Source code in src/wignertime/timeline.py

def update(
    *,
    time=0.0,
    context=wt_config.CONTEXT__INFER,
    origin=wt_config.ORIGIN__INFER,
    **vtvc_dict,
) -> Callable:
    """
    Commands variables at instants: the rows of a timeline.

    It returns a stage, for a `stack` and for `to_timeline`. The first rows of a
    timeline are an `update` like any other, applied to an empty timeline::

        initial = to_timeline(update(AOM__MOT=1, shutter__MOT=0, time=0.0, context="init"))

    Input grammar
    -------------
    A variable is named as a keyword, and followed by what it does::

        update(AOM__MOT=<follows>)

    where ``<follows>`` is one of

    ======================================  ==========================================
    ``value``                               at ``time``
    ``[time, value]``
    ``[time, value, context]``
    ``[[time, value], [time, value], ...]``  several instants for one variable
    ======================================  ==========================================

    Several variables are given at once, and a computed set through ``**``::

        update(AOM__MOT=1, shutter__MOT=[0.1, 1, "MOT"])
        update(**{name: value for name, value in ...})

    ``time`` and ``context`` are **defaults, not overrides** — a variable stating its own
    keeps it. The keyword namespace is open by design, so an unrecognised keyword is
    read as a variable name rather than rejected (see the manuscript's `sec:forwarding`);
    that is what makes the injection idiom work, and it is why there is no second,
    positional way in to be confused with it.

    Origin and context
    ------------------
    `origin` defaults to `wt_config.INFER`: the anchor-then-last chain
    (`config.ORIGIN__DEFAULTS`) fills whichever slots are left unstated, so `time` is a
    duration from the end of the preceding stage. `None` means the same. For absolute
    time, write `origin=0.0`. On an empty timeline there is nothing to be relative to,
    and the origin is absolute zero.

    `time=-math.inf` is before the run and `time=math.inf` after it (#154): the initial
    and the final state, in the contexts a backend reserves for them (`ADwin_LowInit`,
    `ADwin_Finish`). Neither is an instant, so nothing is placed relative to them, and
    until something is written at an instant the origin is absolute zero.

    `context` defaults to `wt_config.INFER` too: an unstated row inherits the latest
    context of the timeline it joins, from a row at an instant. `None` means the same.
    Every row has a context (#156): on an empty timeline, or one holding only the state
    before the run, there is nothing to inherit, so the rows must name theirs, and
    `context=""` is refused. A row added to a finished timeline inherits from the last
    stage of the run, not from the final state at +∞.
    """
    _refuse_timeline("update", vtvc_dict)
    inherit.resolve(context)  # refused where it is written, not where it is applied

    return wt_util.stage(
        _update, update, dict(time=time, context=context, origin=origin, **vtvc_dict)
    )

variable

Outlines the conventions for variables and provides some convenience functions for working with them.

parse(variable)

A dictionary of device, UID and unit.

The convention is <device>__<UID>__<unit> for an analog variable and <device>__<UID> for a digital one, e.g. coil__MOT_lower__A and shutter__MOT. <device> and <unit> contain no _; the UID may contain single ones.

The convention is spelled out by config.VARIABLE__REGEX, which is read here on every call so that a site applying a different one can rebind it.

Source code in src/wignertime/variable.py

def parse(variable: str) -> dict:
    """
    A dictionary of device, UID and unit.

    The convention is `<device>__<UID>__<unit>` for an analog variable and
    `<device>__<UID>` for a digital one, e.g. `coil__MOT_lower__A` and `shutter__MOT`.
    `<device>` and `<unit>` contain no `_`; the UID may contain single ones.

    The convention is spelled out by `config.VARIABLE__REGEX`, which is read here on every call so that a site applying a different one can rebind it.
    """

    match = re.match(wt_config.VARIABLE__REGEX, variable)

    if match is not None:
        d, uid, u = match.groups()
        if u:
            unit = u
        elif wt_config.LABEL__ANCHOR in d:
            unit = wt_config.LABEL__ANCHOR
        else:
            unit = "digital"

        return Munch(device=d, uid=uid, unit=unit)
    else:
        raise ValueError(
            f"Variable {variable} doesn't meet the current naming convention."
        )

units(timeline, do_digital=True)

Returns a set of different timeline units (strs).

Source code in src/wignertime/variable.py

def units(timeline: wt_frame.CLASS, do_digital: bool = True):
    """
    Returns a set of different timeline units (strs).
    """
    us = set(map(unit, timeline["variable"].unique()))
    if do_digital:
        return us
    else:
        us.discard("digital")
        return us

without_unit(variable)

coil__MOT_lower__A without its unit, coil__MOT_lower; a digital name as it is.

Source code in src/wignertime/variable.py

def without_unit(variable):
    """`coil__MOT_lower__A` without its unit, `coil__MOT_lower`; a digital name as it is."""
    p = parse(variable)
    return "{}__{}".format(p.device, p.uid)