visualdynamics.core.run_log_model

run_log_model

The run log read as tests, axes and runs: their states, the holds and owed lists, the views a report reads, and the projection to full level.

A RunLog (core.run_log) is a list of entries, each with plain test, axis and run references. RunLogModel reads a log's corrected view and builds the rest from it, every time it is made: nothing here is stored in the file, so a hold, an owed line or a credited duration is always what the entries say now, and a correction changes it at once.

Tests. A test is opened by a test_opened entry, which carries its kind (one of TEST_KINDS: the project types plus Functional and Other) and its header (run_log.HEADER_FIELDS): what the test is, the article, the documents and revisions, and the plan's numbers the log checks against. A header_change changes one header field. A test has three states: open (it takes runs), closed to runs (when it is closed, or when the next test opens) and released (its release entry, a post-test step 'release'). A test closed to runs still takes findings, post-test steps and dispositions; a released test takes no new runs, findings or header changes.

The roll (roll) closes the current test to runs and opens the next with its header carried from the last. Every carried field is marked carried until it is confirmed (confirm) or changed.

Runs are numbered within a test, and each sits on one axis. What a run holds depends on the test's kind: random level rows ending at full level with the counter, sine sweeps with their direction and count, shock shots with their four checks and the pulse count against the budget, and for a survey or functional run its result and who ran it.

Holds are what the log asks to be answered before the test goes on: an abort with no cause, an over-test question unanswered, a restart with no cause, a "not explained" answer with no later answer, a notch not in the header's planned list with no deviation, a limit, tolerance or abort-band change with no deviation, and a go at a hold-point level with no customer agreement logged. They are shown, never enforced: the log never drives the controller.

Owed lists are checklists with a line each, done by an entry (who and when) or still owed: the setup before the first low level, the axis change, the release checklist, an abort's first-minute fields, a shot's four checks, and a day with no clock check.

The projection (project) takes readings at a level L in dB relative to full level (negative) to full level by the factor 10^(-L/20), exactly, and judges the drive fraction against its window, the peak current at the plan's sigma against the peak rating and the peak voltage against the rail. A check that cannot be made reads "not judged", with why, and is never a pass.

Classes:

Name	Description
Check	One headroom number at full level: what it is, its value at full
Projection	Readings at a level projected to full level, and the headroom
Comparison	A level row compared with an earlier one: the earlier entry, what
LevelReading	A level row read against its test's header: the projection to full
TestHeader	A test's header as it stands at some entry: the value of each
Run	One run of a test: its number, its axis, what its kind of test
Test	One test as the log holds it: its number and kind, the entries
Hold	A hold: its rule (a key of HOLD_RULES), its test, the entry that
OwedLine	One line of an owed list: its key, what it asks, the entry that
OwedList	An owed list: its name, its title, its test and axis, its lines in
Interruption	A stop the test did not plan to run through, or a scheduled one:
Credit	The time credited on one axis of a test, from the counters, against
PulseCount	The pulses a shock test has given, by axis and in all, against the
Finding	A deviation or anomaly: its number, kind ('deviation' or
FileRow	A file an entry names: the entry and the file.
ChronologyRow	An entry in the order of events, with its time on the recorders'
RunLogModel	A run log read as tests, axes and runs, with its holds, owed lists

Functions:

Name	Description
notch_in_plan	Whether a notch with fields (its band, and its depth_db where
full_level_factor	The factor that takes a reading at level_db to full level:
parse_counter	A controller's counter as seconds: '750', '750 s', '12:30' (minutes
clock_text	Seconds as the counter reads them: m:ss, or h:mm:ss from an hour,
project	Readings at a level projected to full level, with the headroom
judge_pulses	The pulses counted by axis (by_axis) judged against budget, a
admit	Refuse an entry of kind for test that the test's state does not
append	RunLog.append, refused first when the test's state does not take
open_test	Open the log's next test, when no test is open to runs.
roll	Close the current test to runs and open the next, its header
close_test	Close test test to runs without opening another. It still takes
change_header	Change one field of test test's header, the value it had
confirm	Confirm carried fields of test test as they stand (every carried
release	Write test test's release, the release checklist's 'release'

Classes

Check dataclass

Check(name: str, at_full: float | None, limit: float | tuple[float, float] | None, verdict: str, message: str)

One headroom number at full level: what it is, its value at full level, the limit it is judged against, and the verdict ('within', 'over', 'under' or 'not judged') with words that say why.

Attributes:

Name	Type	Description
judged	bool	Whether the check was made; one that was not is never a

Attributes

judged property

judged: bool

Whether the check was made; one that was not is never a pass.

Projection dataclass

Projection(level_db: float, factor: float, voltage: float | None, current: float | None, drive_fraction: float | None, checks: tuple[Check, ...])

Readings at a level projected to full level, and the headroom checks. verdict is 'within' only when every check was judged within; 'outside' when any is over or under; otherwise 'not judged'.

Methods:

Name	Description
check	The check named name ('drive fraction', 'peak current',

Attributes:

Name	Type	Description
verdict	str	'within', 'outside' or 'not judged', as the class says.

Attributes

verdict property

verdict: str

'within', 'outside' or 'not judged', as the class says.

Methods:

check

check(name: str) -> Check

The check named name ('drive fraction', 'peak current', 'rms current', 'peak voltage').

Source code in src/visualdynamics/core/run_log_model.py

def check(self, name: str) -> Check:
    """The check named `name` ('drive fraction', 'peak current',
    'rms current', 'peak voltage')."""
    for found in self.checks:
        if found.name == name:
            return found
    raise KeyError(f'no {name!r} check; the checks are: '
                   + ', '.join(check.name for check in self.checks))

Comparison dataclass

Comparison(against: Entry, what: str, ratios: Mapping[str, float | None], flagged: tuple[str, ...])

A level row compared with an earlier one: the earlier entry, what it is (the previous level, or the last run of the same setup), and the ratios of the readings now to then, each None where either reading is missing or zero. flagged names each ratio that moved more than VA_CHANGE.

LevelReading dataclass

LevelReading(entry: Entry, projection: Projection | None, message: str, previous: Comparison | None, last_run: Comparison | None)

A level row read against its test's header: the projection to full level (None when it cannot be made, message saying why), the volts-per-amp ratio against the previous level, and the comparison with the last run of the same setup when the log holds one.

Attributes:

Name	Type	Description
verdict	str	The projection's verdict, or 'not judged' without one.

Attributes

verdict property

verdict: str

The projection's verdict, or 'not judged' without one.

TestHeader dataclass

TestHeader(values: Mapping[str, Any], set_by: Mapping[str, int], carried: tuple[str, ...])

A test's header as it stands at some entry: the value of each field, the entry that set it, and the fields carried from the last test and not yet confirmed or changed.

Methods:

Name	Description
get	The value of header field name, or default.
is_carried	Whether field name is carried and not yet confirmed.

Methods:

get

get(name: str, default: Any = None) -> Any

The value of header field name, or default.

Source code in src/visualdynamics/core/run_log_model.py

def get(self, name: str, default: Any = None) -> Any:
    """The value of header field `name`, or `default`."""
    return self.values.get(name, default)

is_carried

is_carried(name: str) -> bool

Whether field name is carried and not yet confirmed.

Source code in src/visualdynamics/core/run_log_model.py

def is_carried(self, name: str) -> bool:
    """Whether field `name` is carried and not yet confirmed."""
    return name in self.carried

Run dataclass

Run(test: int, number: int, axis: str | None, shape: str, entries: tuple[Entry, ...])

One run of a test: its number, its axis, what its kind of test makes it hold (shape), and its entries in the order written.

Attributes:

Name	Type	Description
levels	tuple[Entry, ...]	The level rows, in the order written.
sweeps	tuple[Entry, ...]	The sweeps, in the order written.
shots	tuple[Entry, ...]	The shots, in the order written.
start	Entry | None	The run's first start entry, or None.
stop	Entry | None	The run's last stop or abort, or None.
outcome	str	'aborted' (an abort since its last start or restart, whether or
reached_full	bool	Whether any entry of the run gives full level (0 dB).
counter	str | None	The counter at the run's last stop, as written, or None.
result	str | None	A survey or functional run's result, from its last stop.
by	str | None	Who ran a survey or functional run, from its last stop.

Attributes

levels property

levels: tuple[Entry, ...]

The level rows, in the order written.

sweeps property

sweeps: tuple[Entry, ...]

The sweeps, in the order written.

shots property

shots: tuple[Entry, ...]

The shots, in the order written.

start property

start: Entry | None

The run's first start entry, or None.

stop property

stop: Entry | None

The run's last stop or abort, or None.

outcome property

outcome: str

'aborted' (an abort since its last start or restart, whether or not a run stop was written beside it), 'stopped' (a scheduled stop), 'completed', or 'running' when no stop follows its last start or restart.

reached_full property

reached_full: bool

Whether any entry of the run gives full level (0 dB).

counter property

counter: str | None

The counter at the run's last stop, as written, or None.

result property

result: str | None

A survey or functional run's result, from its last stop.

by property

by: str | None

Who ran a survey or functional run, from its last stop.

Test dataclass

Test(number: int, kind: str, opened: Entry | None, closed: Entry | None, released: Entry | None, header: TestHeader, axes: tuple[str, ...], runs: tuple[Run, ...], entries: tuple[Entry, ...], problems: tuple[str, ...])

One test as the log holds it: its number and kind, the entries that opened, closed and released it, its header as it stands, its axes in the order first named, its runs, its entries in the order written, and what in them does not fit its states (problems).

Methods:

Name	Description
run	Run number of the test.

Attributes:

Name	Type	Description
title	str	The header's title, or 'Test N'.
state	str	'open', 'closed to runs' or 'released' (STATES).
shape	str	What its runs hold (RUN_SHAPES).

Attributes

title property

title: str

The header's title, or 'Test N'.

state property

state: str

'open', 'closed to runs' or 'released' (STATES).

shape property

shape: str

What its runs hold (RUN_SHAPES).

Methods:

run

run(number: int) -> Run

Run number of the test.

Source code in src/visualdynamics/core/run_log_model.py

def run(self, number: int) -> Run:
    """Run `number` of the test."""
    for run in self.runs:
        if run.number == number:
            return run
    raise KeyError(f'test {self.number} has no run {number}')

Hold dataclass

Hold(rule: str, test: int | None, entry: Entry, message: str)

A hold: its rule (a key of HOLD_RULES), its test, the entry that raised it and what it asks.

OwedLine dataclass

OwedLine(key: str, label: str, done: Entry | None = None, late: bool = False)

One line of an owed list: its key, what it asks, the entry that did it (None while owed), and whether it was done after the list was due.

Attributes:

Name	Type	Description
owed	bool	Whether the line is still owed.

Attributes

owed property

owed: bool

Whether the line is still owed.

OwedList dataclass

OwedList(name: str, title: str, test: int | None, axis: str | None, lines: tuple[OwedLine, ...], due: Entry | None = None)

An owed list: its name, its title, its test and axis, its lines in order, and the entry it was due before (None when it has no due point).

Attributes:

Name	Type	Description
owed	tuple[OwedLine, ...]	The lines still owed.
complete	bool	Whether every line is done.

Attributes

owed property

owed: tuple[OwedLine, ...]

The lines still owed.

complete property

complete: bool

Whether every line is done.

Interruption dataclass

Interruption(entry: Entry, kind: str, cause: Entry | None, path: Entry | None, restart: Entry | None)

A stop the test did not plan to run through, or a scheduled one: the entry (an abort, or a scheduled run stop), its kind ('abort', 'trip' or 'scheduled'), and the cause, path and restart that answer it, each None until logged.

Credit dataclass

Credit(test: int, axis: str | None, seconds: float, complete: bool, notes: tuple[str, ...], duration: float | None, lower: float | None, upper: float | None, verdict: str, message: str)

The time credited on one axis of a test, from the counters, against the header's duration and its tolerance: the seconds credited, whether every segment could be read (complete), why not (notes), the duration and its bounds, and the verdict ('within', 'short', 'over' or 'not judged') with words that say why.

PulseCount dataclass

PulseCount(test: int, count: int, by_axis: Mapping[str | None, int], budget: int | None, verdict: str, message: str)

The pulses a shock test has given, by axis and in all, against the header's budget: 'within', 'over' or 'not judged', with words.

Finding dataclass

Finding(number: str, kind: str, entry: Entry, test: int | None, dispositions: tuple[Entry, ...])

A deviation or anomaly: its number, kind ('deviation' or 'anomaly'), the entry that opened it (corrected), its test, and the dispositions written against it.

Attributes:

Name	Type	Description
open	bool	Whether an anomaly is still open (no disposition); a deviation
owner	str | None	An anomaly's owner, as typed.
fact	str | None	The finding's fact, as typed at the console.

Attributes

open property

open: bool

Whether an anomaly is still open (no disposition); a deviation is never held open waiting for an approval.

owner property

owner: str | None

An anomaly's owner, as typed.

fact property

fact: str | None

The finding's fact, as typed at the console.

FileRow dataclass

FileRow(entry: Entry, file: FileRef)

A file an entry names: the entry and the file.

ChronologyRow dataclass

ChronologyRow(entry: Entry, recorder_time: datetime | None, late: bool)

An entry in the order of events, with its time on the recorders' clock (None with no clock check before it) and whether it is late.

RunLogModel

RunLogModel(log: RunLog)

A run log read as tests, axes and runs, with its holds, owed lists and views. Built from the log's corrected view when made, and never stored: make a new one after the log changes.

Parameters:

Name	Type	Description	Default
log	RunLog	The log, open for writing or read-only.	required

Methods:

Name	Description
tests	Every test, in the order first opened (or first named).
test	Test number; RunLogError when the log has none.
next_test	The number the next test opened takes.
next_run	The number the next run of test test takes.
header_at	The header of entry's test as it stood when entry was
runs	The runs of one test (test its number) or of every test
notch_planned	Whether notch entry is in its test's planned list as the
holds	The holds of one test or every test, in the order of the
owed	Every list with a line still owed, for one test or the whole
checklist	One checklist of test test, every line with who did it and
abort_fields	Each abort's first-minute fields, done by the abort or by the
shot_checks	Each shot's four checks, a line done when the shot (as
clock_days	A line for every day the selected entries fall on (by each
interruptions	The aborts, trips and scheduled stops, each with the cause, the
credited	The time credited on each axis of one test or every test, from
pulses	The pulses test test has given (each shot one, or its
findings	The deviations and anomalies, by number, each with the
open_anomalies	The anomalies with no disposition, with their owners.
files	Every file the selected entries name, in the order written.
late	The selected entries written down late (Entry.is_late).
chronology	The selected entries in the order of their events (by event
reading	Level row entry read against its test's header as it stood

Attributes:

Name	Type	Description
current	Test | None	The test open to runs, the latest opened; None when none is.

Source code in src/visualdynamics/core/run_log_model.py

def __init__(self, log: RunLog) -> None:
    self.log: RunLog = log
    self.entries: tuple[Entry, ...] = tuple(log.view())
    self._by_id = {entry.id: entry for entry in self.entries}
    self._tests: dict[int, Test] = {}
    self._build()

Attributes

current property

current: Test | None

The test open to runs, the latest opened; None when none is.

Methods:

tests

tests() -> list[Test]

Every test, in the order first opened (or first named).

Source code in src/visualdynamics/core/run_log_model.py

def tests(self) -> list[Test]:
    """Every test, in the order first opened (or first named)."""
    return sorted(self._tests.values(),
                  key=lambda test: (test.opened or test.entries[0]).id)

test

test(number: int) -> Test

Test number; RunLogError when the log has none.

Source code in src/visualdynamics/core/run_log_model.py

def test(self, number: int) -> Test:
    """Test `number`; `RunLogError` when the log has none."""
    if number not in self._tests:
        raise RunLogError(f'the log has no test {number}')
    return self._tests[number]

next_test

next_test() -> int

The number the next test opened takes.

Source code in src/visualdynamics/core/run_log_model.py

def next_test(self) -> int:
    """The number the next test opened takes."""
    return max(self._tests, default=0) + 1

next_run

next_run(test: int) -> int

The number the next run of test test takes.

Source code in src/visualdynamics/core/run_log_model.py

def next_run(self, test: int) -> int:
    """The number the next run of test `test` takes."""
    return max((run.number for run in self.test(test).runs),
               default=0) + 1

header_at

header_at(entry: Entry) -> TestHeader

The header of entry's test as it stood when entry was written: what a notch, a go or a reading is judged against, so a header changed later does not change the judgment.

Source code in src/visualdynamics/core/run_log_model.py

def header_at(self, entry: Entry) -> TestHeader:
    """The header of `entry`'s test as it stood when `entry` was
    written: what a notch, a go or a reading is judged against, so a
    header changed later does not change the judgment."""
    test = self.test(entry.test) if entry.test is not None else None
    if test is None:
        return TestHeader({}, {}, ())
    return self._header(test.entries, test.opened, before=entry.id)

runs

runs(test: Any = ALL) -> list[Run]

The runs of one test (test its number) or of every test (ALL), test by test.

Source code in src/visualdynamics/core/run_log_model.py

def runs(self, test: Any = ALL) -> list[Run]:
    """The runs of one test (`test` its number) or of every test
    (`ALL`), test by test."""
    return [run for found in self._selected_tests(test)
            for run in found.runs]

notch_planned

notch_planned(entry: Entry) -> tuple[bool, str]

Whether notch entry is in its test's planned list as the header stood when it was written, and why: on its axis (a planned notch with no axis is on every axis), its band inside the planned band, and its depth no deeper than the planned depth where the plan gives one. The typed planned mark is not what decides it.

Source code in src/visualdynamics/core/run_log_model.py

def notch_planned(self, entry: Entry) -> tuple[bool, str]:
    """Whether notch `entry` is in its test's planned list as the
    header stood when it was written, and why: on its axis (a planned
    notch with no axis is on every axis), its band inside the planned
    band, and its depth no deeper than the planned depth where the
    plan gives one. The typed `planned` mark is not what decides it."""
    entry = self._by_id.get(entry.id, entry)
    return notch_in_plan(self.header_at(entry).get('notches_planned'),
                         entry.axis, entry.fields)

holds

holds(test: Any = ALL) -> list[Hold]

The holds of one test or every test, in the order of the entries that raised them (HOLD_RULES).

Source code in src/visualdynamics/core/run_log_model.py

def holds(self, test: Any = ALL) -> list[Hold]:
    """The holds of one test or every test, in the order of the
    entries that raised them (`HOLD_RULES`)."""
    found: list[Hold] = []
    for each in self._selected_tests(test):
        found.extend(self._holds_of(each))
    return sorted(found, key=lambda hold: hold.entry.id)

owed

owed(test: Any = ALL) -> list[OwedList]

Every list with a line still owed, for one test or the whole log: the setup before the first low level, each axis change, the release checklist once the test is closed to runs, each abort's first-minute fields, each shot's four checks, and the days with no clock check.

Source code in src/visualdynamics/core/run_log_model.py

def owed(self, test: Any = ALL) -> list[OwedList]:
    """Every list with a line still owed, for one test or the whole
    log: the setup before the first low level, each axis change, the
    release checklist once the test is closed to runs, each abort's
    first-minute fields, each shot's four checks, and the days with
    no clock check."""
    lists: list[OwedList] = []
    for each in self._selected_tests(test):
        lists.extend(self._checklists(each))
        lists.extend(self.abort_fields(each.number))
        lists.extend(self.shot_checks(each.number))
    lists.append(self.clock_days(test))
    return [found for found in lists if not found.complete]

checklist

checklist(test: int, name: str, axis: str | None = None) -> OwedList

One checklist of test test, every line with who did it and when or still owed: 'before low level' (the test's first axis), 'axis change' (to axis, an axis after the first) or 'release'. A line is done by a setup step (a post-test step for the release checklist) naming it by its key or its words; an axis change's steps are on the new axis, and its first line may name the axis it closes.

Source code in src/visualdynamics/core/run_log_model.py

def checklist(self, test: int, name: str,
              axis: str | None = None) -> OwedList:
    """One checklist of test `test`, every line with who did it and
    when or still owed: 'before low level' (the test's first axis),
    'axis change' (to `axis`, an axis after the first) or 'release'.
    A line is done by a setup step (a post-test step for the release
    checklist) naming it by its key or its words; an axis change's
    steps are on the new axis, and its first line may name the axis
    it closes."""
    found = self.test(test)
    if name == 'before low level':
        first = found.axes[0] if found.axes else None
        return self._checklist(
            name, 'Before the first low level', found, first,
            BEFORE_LOW_LEVEL, 'setup_step', None,
            self._first_drive(found, first))
    if name == 'axis change':
        if axis not in found.axes[1:]:
            raise RunLogError(f'test {test} has no axis change to '
                              f'{axis!r}; its axes are: '
                              + (', '.join(found.axes) or 'none'))
        index = found.axes.index(axis)
        rows = self._checklist(name, f'The change to axis {axis}', found,
                               axis, AXIS_CHANGE, 'setup_step', [axis],
                               self._first_drive(found, axis))
        if rows.lines[0].owed:
            closing = self._checklist(
                name, '', found, axis, AXIS_CHANGE[:1], 'setup_step',
                [found.axes[index - 1]], rows.due).lines[0]
            rows = OwedList(rows.name, rows.title, rows.test, rows.axis,
                            (closing, *rows.lines[1:]), rows.due)
        return rows
    if name == 'release':
        return self._checklist(name, 'The release checklist', found, None,
                               RELEASE, 'post_test_step', None, None)
    raise RunLogError(f'{name!r} is not a checklist; the checklists are: '
                      'before low level, axis change, release')

abort_fields

abort_fields(test: Any = ALL) -> list[OwedList]

Each abort's first-minute fields, done by the abort or by the correction that added the field: the level, the channel, the line or band, the panel fault, the counter, who called it, what was heard, the records read and the over-test answer. A partial abort is accepted; what it lacks is owed here.

Source code in src/visualdynamics/core/run_log_model.py

def abort_fields(self, test: Any = ALL) -> list[OwedList]:
    """Each abort's first-minute fields, done by the abort or by the
    correction that added the field: the level, the channel, the
    line or band, the panel fault, the counter, who called it, what
    was heard, the records read and the over-test answer. A partial
    abort is accepted; what it lacks is owed here."""
    lists = []
    optional = KINDS['abort'].optional
    for each in self._selected_tests(test):
        for abort in each.entries:
            if abort.kind != 'abort':
                continue
            supplied = {name: abort for name in abort.fields}
            for correction in self.log.corrections_of(abort.id):
                for name, value in correction.fields.items():
                    if name in optional:
                        if value is None:
                            supplied.pop(name, None)
                        else:
                            supplied[name] = correction
            lines = tuple(
                OwedLine(name, name.replace('_', ' '),
                         supplied.get(name) if name in abort.fields
                         else None)
                for name in optional)
            lists.append(OwedList('abort', f'The first minute of abort '
                                  f'#{abort.id}', each.number,
                                  self._axis_of(each, abort), lines))
    return lists

shot_checks

shot_checks(test: Any = ALL) -> list[OwedList]

Each shot's four checks, a line done when the shot (as corrected) carries every field of the check.

Source code in src/visualdynamics/core/run_log_model.py

def shot_checks(self, test: Any = ALL) -> list[OwedList]:
    """Each shot's four checks, a line done when the shot (as
    corrected) carries every field of the check."""
    lists = []
    for each in self._selected_tests(test):
        for shot in each.entries:
            if shot.kind != 'shot':
                continue
            lines = tuple(
                OwedLine(key, label,
                         shot if all(name in shot.fields
                                     for name in names) else None)
                for key, names, label in SHOT_CHECKS)
            lists.append(OwedList('shot', f'The checks after shot '
                                  f'#{shot.id}', each.number,
                                  self._axis_of(each, shot), lines))
    return lists

clock_days

clock_days(test: Any = ALL) -> OwedList

A line for every day the selected entries fall on (by each entry's own date), done by that day's first clock check, which may be anywhere in the log.

Source code in src/visualdynamics/core/run_log_model.py

def clock_days(self, test: Any = ALL) -> OwedList:
    """A line for every day the selected entries fall on (by each
    entry's own date), done by that day's first clock check, which
    may be anywhere in the log."""
    checks: dict[date, Entry] = {}
    for entry in self.entries:
        if entry.kind == 'clock_check':
            checks.setdefault(entry.event_time.date(), entry)
    days: list[date] = []
    for entry in self._selected_entries(test):
        day = entry.event_time.date()
        if day not in days:
            days.append(day)
    lines = tuple(OwedLine(day.isoformat(), f'A clock check on {day}',
                           checks.get(day)) for day in sorted(days))
    return OwedList('clock', 'A clock check each day',
                    test if isinstance(test, int) else None, None, lines)

interruptions

interruptions(test: Any = ALL) -> list[Interruption]

The aborts, trips and scheduled stops, each with the cause, the path and the restart that answer it (by refers_to, or the next written in the same test).

Source code in src/visualdynamics/core/run_log_model.py

def interruptions(self, test: Any = ALL) -> list[Interruption]:
    """The aborts, trips and scheduled stops, each with the cause, the
    path and the restart that answer it (by `refers_to`, or the next
    written in the same test)."""
    found = []
    for each in self._selected_tests(test):
        stops = [entry for entry in each.entries
                 if entry.kind == 'abort' or (
                     entry.kind == 'run_stop'
                     and entry.fields.get('scheduled'))]
        causes = self._link(each, stops, 'cause_found')
        paths = self._link(each, stops, 'path_decided')
        restarts = self._link(each, stops, 'restart')
        for stop in stops:
            kind = ('scheduled' if stop.kind == 'run_stop' else
                    'trip' if stop.fields['stop'] in ('amplifier trip',
                                                      'power loss')
                    else 'abort')
            found.append(Interruption(stop, kind, causes.get(stop.id),
                                      paths.get(stop.id),
                                      restarts.get(stop.id)))
    return found

credited

credited(test: Any = ALL) -> list[Credit]

The time credited on each axis of one test or every test, from the counters, against the header's duration and tolerance.

A run's time counts once it reaches full level (an entry giving 0 dB). A segment runs from its start (a run start's counter, 0 when not given, or a restart's counter) to the counter at its stop or abort. A restart that resumes continues the count; one that starts the test again discards what was credited before it on the axis. A segment whose counter is missing or unreadable leaves the total a lower bound, and the verdict not judged.

Source code in src/visualdynamics/core/run_log_model.py

def credited(self, test: Any = ALL) -> list[Credit]:
    """The time credited on each axis of one test or every test, from
    the counters, against the header's duration and tolerance.

    A run's time counts once it reaches full level (an entry giving
    0 dB). A segment runs from its start (a run start's counter, 0
    when not given, or a restart's counter) to the counter at its
    stop or abort. A restart that resumes continues the count; one
    that starts the test again discards what was credited before it
    on the axis. A segment whose counter is missing or unreadable
    leaves the total a lower bound, and the verdict not judged.
    """
    found = []
    for each in self._selected_tests(test):
        axes: list[str | None] = []
        for entry in each.entries:
            if entry.kind in RUN_KINDS:
                axis = self._axis_of(each, entry)
                if axis not in axes:
                    axes.append(axis)
        found.extend(self._credit(each, axis) for axis in axes)
    return found

pulses

pulses(test: int) -> PulseCount

The pulses test test has given (each shot one, or its pulses), by axis and in all, against the header's budget. Every shot entry counts, a missed one too: a missed pulse spends the budget.

Source code in src/visualdynamics/core/run_log_model.py

def pulses(self, test: int) -> PulseCount:
    """The pulses test `test` has given (each shot one, or its
    `pulses`), by axis and in all, against the header's budget. Every
    shot entry counts, a missed one too: a missed pulse spends the
    budget."""
    found = self.test(test)
    by_axis: dict[str | None, int] = {}
    for shot in found.entries:
        if shot.kind == 'shot':
            axis = self._axis_of(found, shot)
            by_axis[axis] = by_axis.get(axis, 0) + shot.fields.get(
                'pulses', 1)
    return judge_pulses(test, by_axis, found.header.get('pulse_budget'))

findings

findings(test: Any = ALL) -> list[Finding]

The deviations and anomalies, by number, each with the dispositions written against it anywhere in the log.

Source code in src/visualdynamics/core/run_log_model.py

def findings(self, test: Any = ALL) -> list[Finding]:
    """The deviations and anomalies, by number, each with the
    dispositions written against it anywhere in the log."""
    dispositions: dict[str, list[Entry]] = {}
    for entry in self.entries:
        if entry.kind == 'anomaly_dispositioned':
            dispositions.setdefault(entry.fields['number'], []).append(
                entry)
    found = []
    for entry in self._selected_entries(test):
        if entry.kind in ('deviation', 'anomaly_opened') \
                and 'number' in entry.fields:
            number = entry.fields['number']
            kind = 'deviation' if entry.kind == 'deviation' else 'anomaly'
            found.append(Finding(number, kind, entry, entry.test,
                                 tuple(dispositions.get(number, ())
                                       if kind == 'anomaly' else ())))
    return found

open_anomalies

open_anomalies(test: Any = ALL) -> list[Finding]

The anomalies with no disposition, with their owners.

Source code in src/visualdynamics/core/run_log_model.py

def open_anomalies(self, test: Any = ALL) -> list[Finding]:
    """The anomalies with no disposition, with their owners."""
    return [found for found in self.findings(test) if found.open]

files

files(test: Any = ALL) -> list[FileRow]

Every file the selected entries name, in the order written.

Source code in src/visualdynamics/core/run_log_model.py

def files(self, test: Any = ALL) -> list[FileRow]:
    """Every file the selected entries name, in the order written."""
    return [FileRow(entry, ref) for entry in self._selected_entries(test)
            for ref in entry.files]

late

late(test: Any = ALL) -> list[Entry]

The selected entries written down late (Entry.is_late).

Source code in src/visualdynamics/core/run_log_model.py

def late(self, test: Any = ALL) -> list[Entry]:
    """The selected entries written down late (`Entry.is_late`)."""
    return [entry for entry in self._selected_entries(test)
            if entry.is_late()]

chronology

chronology(test: Any = ALL) -> list[ChronologyRow]

The selected entries in the order of their events (by event time, then by id), each with its time on the recorders' clock.

Source code in src/visualdynamics/core/run_log_model.py

def chronology(self, test: Any = ALL) -> list[ChronologyRow]:
    """The selected entries in the order of their events (by event
    time, then by id), each with its time on the recorders' clock."""
    rows = sorted(self._selected_entries(test),
                  key=lambda entry: (entry.event_time, entry.id))
    return [ChronologyRow(entry, self.log.recorder_time(entry),
                          entry.is_late()) for entry in rows]

reading

reading(entry: Entry | int) -> LevelReading

Level row entry read against its test's header as it stood when the row was written: projected to full level (project), its volts per amp against the previous level of the run, and its readings against the last run of the same setup (the same level on the same axis, in an earlier run of the test or of a test of the same kind and configuration). What cannot be read reads not judged, with why.

Source code in src/visualdynamics/core/run_log_model.py

def reading(self, entry: Entry | int) -> LevelReading:
    """Level row `entry` read against its test's header as it stood
    when the row was written: projected to full level (`project`),
    its volts per amp against the previous level of the run, and its
    readings against the last run of the same setup (the same level
    on the same axis, in an earlier run of the test or of a test of
    the same kind and configuration). What cannot be read reads not
    judged, with why."""
    entry_id = entry.id if isinstance(entry, Entry) else entry
    if entry_id not in self._by_id:
        raise RunLogError(f'the log has no entry #{entry_id}')
    row = self._by_id[entry_id]
    if row.kind != 'level_row':
        raise RunLogError(f'#{row.id} is a {row.label} entry, not a level '
                          'row')
    fields = row.fields
    header = self.header_at(row)
    projection = None
    if 'level_db' not in fields:
        message = 'not judged: the row gives no level'
    elif row.test is None:
        message = 'not judged: the row belongs to no test, so no header'
    else:
        try:
            projection = project(
                fields['level_db'], voltage=fields.get('voltage'),
                current=fields.get('current'),
                drive_fraction=fields.get('drive_fraction'),
                sigma=header.get('sigma'),
                rated_current_peak=header.get('rated_current_peak'),
                rated_current_rms=header.get('rated_current_rms'),
                rail_voltage=header.get('rail_voltage'),
                drive_window=header.get('drive_window'))
            message = str(projection)
        except RunLogError as error:
            message = f'not judged: {error}'
    return LevelReading(row, projection, message, self._previous(row),
                        self._last_run(row))

Functions:

notch_in_plan

notch_in_plan(planned: Sequence[Mapping[str, Any]] | None, axis: str | None, fields: Mapping[str, Any]) -> tuple[bool, str]

Whether a notch with fields (its band, and its depth_db where given) on axis is in the planned list planned (a header's notches_planned), and why: on its axis (a planned notch with no axis is on every axis), its band inside the planned band, and its depth no deeper than the planned depth where the plan gives one. What RunLogModel.notch_planned judges a written notch by, and what an entry form shows before the notch is written. A typed planned mark is not what decides it.

Source code in src/visualdynamics/core/run_log_model.py

def notch_in_plan(planned: Sequence[Mapping[str, Any]] | None,
                  axis: str | None, fields: Mapping[str, Any]
                  ) -> tuple[bool, str]:
    """Whether a notch with `fields` (its band, and its depth_db where
    given) on `axis` is in the planned list `planned` (a header's
    `notches_planned`), and why: on its axis (a planned notch with no
    axis is on every axis), its band inside the planned band, and its
    depth no deeper than the planned depth where the plan gives one.
    What `RunLogModel.notch_planned` judges a written notch by, and what
    an entry form shows before the notch is written. A typed `planned`
    mark is not what decides it."""
    planned = planned or []
    marked = (' It is marked planned, but the header\'s list decides.'
              if fields.get('planned') else '')
    if 'band' not in fields:
        return False, 'the notch gives no band, so it cannot be found in ' \
                      'the planned list.' + marked
    try:
        low, high = band_hz(fields['band'], 'its band')
    except RunLogError as error:
        return False, f'{error}.' + marked
    if not planned:
        return False, 'the header plans no notches.' + marked
    reasons = []
    for number, notch in enumerate(planned, start=1):
        if notch.get('axis') is not None and notch['axis'] != axis:
            continue
        try:
            p_low, p_high = band_hz(notch['band'])
        except RunLogError:
            continue
        if low < p_low or high > p_high:
            continue
        depth = notch.get('depth_db')
        if depth is not None:
            if 'depth_db' not in fields:
                reasons.append(f'it is inside planned notch {number}, '
                               'which has a depth, but it gives none')
                continue
            if abs(fields['depth_db']) > abs(depth) + LEVEL_MATCH_DB:
                reasons.append(f'it is inside planned notch {number} but '
                               f'deeper than its {abs(depth):g} dB')
                continue
        return True, f'inside planned notch {number}.'
    why = '; '.join(reasons) if reasons else (
        f'{low:g} to {high:g} Hz on axis {axis or "(none)"} is in no planned '
        'notch')
    return False, why + '.' + marked

full_level_factor

full_level_factor(level_db: float) -> float

The factor that takes a reading at level_db to full level: 10^(-L/20), with L in dB relative to full level (-12 for twelve below, which gives 3.98).

Refused with RunLogError: a level that is not a finite number, and a level above full (positive), which is not projected down.

Source code in src/visualdynamics/core/run_log_model.py

def full_level_factor(level_db: float) -> float:
    """The factor that takes a reading at `level_db` to full level:
    10^(-L/20), with L in dB relative to full level (-12 for twelve
    below, which gives 3.98).

    Refused with `RunLogError`: a level that is not a finite number, and
    a level above full (positive), which is not projected down.
    """
    if isinstance(level_db, bool) or not isinstance(level_db, (int, float)):
        raise RunLogError(f'a level is a number of dB relative to full level '
                          f'(-12 for twelve below), not {level_db!r}')
    if not math.isfinite(level_db):
        raise RunLogError(f'a level of {level_db} dB is not a finite number')
    if level_db > 0:
        raise RunLogError(f'a level of {level_db:+g} dB is above full level; '
                          'a level is given in dB relative to full level, '
                          'negative below it (-12 for twelve below)')
    return 10.0 ** (-level_db / 20.0)

parse_counter

parse_counter(text: Any) -> float

A controller's counter as seconds: '750', '750 s', '12:30' (minutes and seconds) or '1:02:30' (hours, minutes and seconds), with decimals on the seconds.

Refused with RunLogError naming the text: anything else, and minutes or seconds of 60 or more after the first part.

Source code in src/visualdynamics/core/run_log_model.py

def parse_counter(text: Any) -> float:
    """A controller's counter as seconds: '750', '750 s', '12:30' (minutes
    and seconds) or '1:02:30' (hours, minutes and seconds), with
    decimals on the seconds.

    Refused with `RunLogError` naming the text: anything else, and
    minutes or seconds of 60 or more after the first part.
    """
    if isinstance(text, bool) or not isinstance(text, (str, int, float)):
        raise RunLogError(f'a counter is a time such as 12:30, not {text!r}')
    raw = str(text).strip().lower()
    if raw.endswith('s'):
        raw = raw[:-1].rstrip()
    parts = raw.split(':')
    if not 1 <= len(parts) <= 3 or not all(
            re.fullmatch(r'\d+(\.\d+)?', part) for part in parts) or any(
            '.' in part for part in parts[:-1]):
        raise RunLogError(f'the counter {text!r} is not read as a time: give '
                          'seconds, minutes:seconds or hours:minutes:seconds')
    numbers = [float(part) for part in parts]
    if any(value >= 60 for value in numbers[1:]):
        raise RunLogError(f'the counter {text!r} has minutes or seconds of 60 '
                          'or more')
    seconds = 0.0
    for value in numbers:
        seconds = seconds * 60.0 + value
    return seconds

clock_text

clock_text(seconds: float) -> str

Seconds as the counter reads them: m:ss, or h:mm:ss from an hour, to the second.

Source code in src/visualdynamics/core/run_log_model.py

def clock_text(seconds: float) -> str:
    """Seconds as the counter reads them: m:ss, or h:mm:ss from an hour,
    to the second."""
    whole = round(seconds)
    hours, rest = divmod(whole, 3600)
    minutes, secs = divmod(rest, 60)
    if hours:
        return f'{hours}:{minutes:02d}:{secs:02d}'
    return f'{minutes}:{secs:02d}'

project

project(level_db: float, *, voltage: float | None = None, current: float | None = None, drive_fraction: float | None = None, sigma: float | None = None, rated_current_peak: float | None = None, rated_current_rms: float | None = None, rail_voltage: float | None = None, drive_window: Sequence[float] | None = None) -> Projection

Readings at a level projected to full level, with the headroom checks.

Each reading is multiplied by full_level_factor (10^(-L/20), exactly). The drive fraction at full level is judged against its window; the rms current times sigma (the peak a random drive reaches, at the plan's sigma) against the peak current rating; the rms voltage times sigma against the rail; and, when the header gives a continuous rating, the rms current against it, which is what a sine test budgets.

Parameters:

Name	Type	Description	Default
level_db	float	The level the readings were taken at, in dB relative to full level: -12 for twelve below. Not above 0.	required
voltage	float	The amplifier's rms voltage (V) and current (A) at the level.	None
current	float	The amplifier's rms voltage (V) and current (A) at the level.	None
drive_fraction	float	The drive as a fraction of its range, 0 to 1.	None
sigma	float	The plan's peak-to-rms ratio of the drive.	None
rated_current_peak	float	The amplifier's peak and continuous current ratings (A).	None
rated_current_rms	float	The amplifier's peak and continuous current ratings (A).	None
rail_voltage	float	The amplifier's rail, the peak voltage it can give (V).	None
drive_window	(float, float)	The drive fraction's window at full level; DRIVE_WINDOW when not given.	None

Returns:

Type	Description
Projection	The factor, the readings, and each check with its verdict. A check that cannot be made (a reading or a rating missing, or a reading of zero) reads "not judged", with why.

Raises:

Type	Description
RunLogError	A level above full or not finite, a reading that is negative or not finite, a drive fraction above 1, a sigma or rating that is zero, and a window that is not two fractions low then high.

Source code in src/visualdynamics/core/run_log_model.py

def project(level_db: float, *, voltage: float | None = None,
            current: float | None = None,
            drive_fraction: float | None = None,
            sigma: float | None = None,
            rated_current_peak: float | None = None,
            rated_current_rms: float | None = None,
            rail_voltage: float | None = None,
            drive_window: Sequence[float] | None = None) -> Projection:
    """Readings at a level projected to full level, with the headroom
    checks.

    Each reading is multiplied by `full_level_factor` (10^(-L/20),
    exactly). The drive fraction at full level is judged against its
    window; the rms current times `sigma` (the peak a random drive
    reaches, at the plan's sigma) against the peak current rating; the
    rms voltage times `sigma` against the rail; and, when the header
    gives a continuous rating, the rms current against it, which is
    what a sine test budgets.

    Parameters
    ----------
    level_db : float
        The level the readings were taken at, in dB relative to full
        level: -12 for twelve below. Not above 0.
    voltage, current : float, optional
        The amplifier's rms voltage (V) and current (A) at the level.
    drive_fraction : float, optional
        The drive as a fraction of its range, 0 to 1.
    sigma : float, optional
        The plan's peak-to-rms ratio of the drive.
    rated_current_peak, rated_current_rms : float, optional
        The amplifier's peak and continuous current ratings (A).
    rail_voltage : float, optional
        The amplifier's rail, the peak voltage it can give (V).
    drive_window : (float, float), optional
        The drive fraction's window at full level; `DRIVE_WINDOW` when
        not given.

    Returns
    -------
    Projection
        The factor, the readings, and each check with its verdict. A
        check that cannot be made (a reading or a rating missing, or a
        reading of zero) reads "not judged", with why.

    Raises
    ------
    RunLogError
        A level above full or not finite, a reading that is negative or
        not finite, a drive fraction above 1, a sigma or rating that is
        zero, and a window that is not two fractions low then high.
    """
    factor = full_level_factor(level_db)
    voltage = _reading(voltage, 'a voltage')
    current = _reading(current, 'a current')
    drive_fraction = _reading(drive_fraction, 'a drive fraction', at_most=1.0)
    sigma = _rating(sigma, 'a sigma')
    rated_current_peak = _rating(rated_current_peak, 'a peak current rating')
    rated_current_rms = _rating(rated_current_rms,
                                'a continuous current rating')
    rail_voltage = _rating(rail_voltage, 'a rail voltage')
    defaulted = drive_window is None
    window = DRIVE_WINDOW if drive_window is None else tuple(drive_window)
    if len(window) != 2 or not all(
            isinstance(edge, (int, float)) and not isinstance(edge, bool)
            and math.isfinite(edge) for edge in window) or not (
            0 <= window[0] < window[1] <= 1):
        raise RunLogError(f'a drive window is two fractions of the range, low '
                          f'then high, between 0 and 1, not {drive_window!r}')
    low, high = float(window[0]), float(window[1])
    checks = []
    named = 'the default window' if defaulted else 'the window'
    if drive_fraction is None:
        checks.append(Check('drive fraction', None, (low, high), 'not judged',
                            'not judged: no drive fraction was read'))
    elif drive_fraction == 0:
        checks.append(Check('drive fraction', None, (low, high), 'not judged',
                            'not judged: the drive fraction reads zero, which '
                            'says the controller is not driving'))
    else:
        at_full = drive_fraction * factor
        verdict = ('over' if at_full > high else
                   'under' if at_full < low else 'within')
        word = {'over': 'above', 'under': 'below', 'within': 'within'}[verdict]
        checks.append(Check('drive fraction', at_full, (low, high), verdict,
                            f'{at_full:.3g} of the range at full level, {word} '
                            f'{named} {low:.3g} to {high:.3g}'))
    checks.append(_judge_peak('peak current', current, 'current', factor,
                              sigma, 'sigma', rated_current_peak,
                              'peak current rating', 'A'))
    if rated_current_rms is not None:
        checks.append(_judge_peak('rms current', current, 'current', factor,
                                  1.0, '', rated_current_rms,
                                  'continuous current rating', 'A'))
    checks.append(_judge_peak('peak voltage', voltage, 'voltage', factor,
                              sigma, 'sigma', rail_voltage, 'rail', 'V'))
    return Projection(float(level_db), factor, voltage, current,
                      drive_fraction, tuple(checks))

judge_pulses

judge_pulses(test: int, by_axis: Mapping[str | None, int], budget: int | None) -> PulseCount

The pulses counted by axis (by_axis) judged against budget, a test header's pulse_budget: what RunLogModel.pulses reports, and what a shot form shows with the shot being typed counted in.

Source code in src/visualdynamics/core/run_log_model.py

def judge_pulses(test: int, by_axis: Mapping[str | None, int],
                 budget: int | None) -> PulseCount:
    """The pulses counted by axis (`by_axis`) judged against `budget`, a
    test header's `pulse_budget`: what `RunLogModel.pulses` reports, and
    what a shot form shows with the shot being typed counted in."""
    by_axis = dict(by_axis)
    count = sum(by_axis.values())
    if budget is None:
        return PulseCount(test, count, by_axis, None, 'not judged',
                          f'{count} pulses; not judged: the header gives '
                          'no pulse budget')
    if count > budget:
        return PulseCount(test, count, by_axis, budget, 'over',
                          f'{count} pulses, over the budget of {budget} '
                          f'by {count - budget}')
    return PulseCount(test, count, by_axis, budget, 'within',
                      f'{count} of {budget} pulses, {budget - count} left')

admit

admit(log: RunLog | RunLogModel, kind: str, test: int | None) -> None

Refuse an entry of kind for test that the test's state does not take, with RunLogError naming the test and its state: a run entry with no test, any entry for a test never opened, a run entry for a test closed to runs, and a run, finding or header change for a released test. A test is opened with open_test or roll, and closed with close_test.

Source code in src/visualdynamics/core/run_log_model.py

def admit(log: RunLog | RunLogModel, kind: str, test: int | None) -> None:
    """Refuse an entry of `kind` for `test` that the test's state does not
    take, with `RunLogError` naming the test and its state: a run entry
    with no test, any entry for a test never opened, a run entry for a
    test closed to runs, and a run, finding or header change for a
    released test. A test is opened with `open_test` or `roll`, and
    closed with `close_test`."""
    model = log if isinstance(log, RunLogModel) else RunLogModel(log)
    if kind in ('test_opened', 'test_closed'):
        raise RunLogError('a test is opened with open_test or roll, and '
                          'closed with close_test or roll')
    if test is None:
        if kind in RUN_KINDS:
            raise RunLogError(f'a {KINDS[kind].label} entry belongs to a run '
                              'of a test; give its test')
        return
    found = model._tests.get(test)
    if found is None or found.opened is None:
        raise RunLogError(f'test {test} has not been opened; open it first')
    if found.state == 'released' and kind in RUN_KINDS | RELEASED_REFUSES:
        raise RunLogError(f'test {test} is released; it takes no new runs, '
                          'findings or header changes')
    if found.state == 'closed to runs' and kind in RUN_KINDS:
        raise RunLogError(f'test {test} is closed to runs; it still takes '
                          'findings, post-test steps and dispositions')

append

append(log: RunLog, kind: str, *, author: str, test: int | None = None, **kwargs: Any) -> Entry

RunLog.append, refused first when the test's state does not take the entry (admit).

Source code in src/visualdynamics/core/run_log_model.py

def append(log: RunLog, kind: str, *, author: str,
           test: int | None = None, **kwargs: Any) -> Entry:
    """`RunLog.append`, refused first when the test's state does not take
    the entry (`admit`)."""
    admit(log, kind, test)
    return log.append(kind, author=author, test=test, **kwargs)

open_test

open_test(log: RunLog, *, author: str, test_kind: str, header: Mapping[str, Any] | None = None, text: str = '', event_time: datetime | None = None, files: Iterable[Any] = ()) -> Entry

Open the log's next test, when no test is open to runs.

Parameters:

Name	Type	Description	Default
log	RunLog	The log, open for writing.	required
author	str	The author's typed initials.	required
test_kind	str	One of TEST_KINDS.	required
header	mapping	The header's fields (run_log.HEADER_FIELDS).	None
text	str	Free text.	''
event_time	datetime	When it was opened, with its zone; now, if not given.	None
files	iterable	Files the header attaches (the plan, the procedure, the profile), each a path or a FileRef.	()

Returns:

Type	Description
Entry	The test opened entry.

Raises:

Type	Description
RunLogError	When a test is open (roll to the next instead), the kind is not one of TEST_KINDS, or a header field is refused.

Source code in src/visualdynamics/core/run_log_model.py

def open_test(log: RunLog, *, author: str, test_kind: str,
              header: Mapping[str, Any] | None = None, text: str = '',
              event_time: datetime | None = None,
              files: Iterable[Any] = ()) -> Entry:
    """Open the log's next test, when no test is open to runs.

    Parameters
    ----------
    log : RunLog
        The log, open for writing.
    author : str
        The author's typed initials.
    test_kind : str
        One of `TEST_KINDS`.
    header : mapping, optional
        The header's fields (`run_log.HEADER_FIELDS`).
    text : str, optional
        Free text.
    event_time : datetime, optional
        When it was opened, with its zone; now, if not given.
    files : iterable, optional
        Files the header attaches (the plan, the procedure, the
        profile), each a path or a `FileRef`.

    Returns
    -------
    Entry
        The test opened entry.

    Raises
    ------
    RunLogError
        When a test is open (roll to the next instead), the kind is not
        one of `TEST_KINDS`, or a header field is refused.
    """
    model = RunLogModel(log)
    if model.current is not None:
        raise RunLogError(f'test {model.current.number} is open; roll to the '
                          'next test, which closes it to runs')
    fields = {**(header or {}), 'test_kind': _test_kind(test_kind)}
    if 'carried' in fields:
        raise RunLogError('a test opened by hand carries nothing; roll to '
                          'carry a header')
    return log.append('test_opened', author=author, test=model.next_test(),
                      fields=fields, text=text, event_time=event_time,
                      files=files)

roll

roll(log: RunLog, *, author: str, test_kind: str | None = None, header: Mapping[str, Any] | None = None, reason: str = '', text: str = '', event_time: datetime | None = None, files: Iterable[Any] = ()) -> tuple[Entry, Entry]

Close the current test to runs and open the next, its header carried from the current one.

Every field of the current header not given in header (nor the kind, when test_kind is given) is carried, and marked carried until it is confirmed (confirm) or changed (change_header). The current test still takes findings, post-test steps and dispositions until its release entry; its release checklist is owed.

Parameters:

Name	Type	Description	Default
log	RunLog	The log, open for writing.	required
author	str	The author's typed initials.	required
test_kind	str	The next test's kind; carried when not given.	None
header	mapping	The next test's header fields that differ, or are confirmed.	None
reason	str	Why the current test closes; 'rolled to test N' when not given.	''
text	str	Free text on the next test's opening.	''
event_time	datetime	When, with its zone; now, if not given.	None
files	iterable	Files the next test's header attaches.	()

Returns:

Type	Description
(Entry, Entry)	The current test's closing and the next test's opening.

Raises:

Type	Description
RunLogError	When no test is open, the kind is not one of TEST_KINDS, or a header field is refused; nothing is written then.

Source code in src/visualdynamics/core/run_log_model.py

def roll(log: RunLog, *, author: str, test_kind: str | None = None,
         header: Mapping[str, Any] | None = None, reason: str = '',
         text: str = '', event_time: datetime | None = None,
         files: Iterable[Any] = ()) -> tuple[Entry, Entry]:
    """Close the current test to runs and open the next, its header
    carried from the current one.

    Every field of the current header not given in `header` (nor the
    kind, when `test_kind` is given) is carried, and marked carried
    until it is confirmed (`confirm`) or changed (`change_header`). The
    current test still takes findings, post-test steps and dispositions
    until its release entry; its release checklist is owed.

    Parameters
    ----------
    log : RunLog
        The log, open for writing.
    author : str
        The author's typed initials.
    test_kind : str, optional
        The next test's kind; carried when not given.
    header : mapping, optional
        The next test's header fields that differ, or are confirmed.
    reason : str, optional
        Why the current test closes; 'rolled to test N' when not given.
    text : str, optional
        Free text on the next test's opening.
    event_time : datetime, optional
        When, with its zone; now, if not given.
    files : iterable, optional
        Files the next test's header attaches.

    Returns
    -------
    (Entry, Entry)
        The current test's closing and the next test's opening.

    Raises
    ------
    RunLogError
        When no test is open, the kind is not one of `TEST_KINDS`, or a
        header field is refused; nothing is written then.
    """
    model = RunLogModel(log)
    current = model.current
    if current is None:
        raise RunLogError('no test is open to roll from; open one with '
                          'open_test')
    given = dict(header or {})
    if 'carried' in given:
        raise RunLogError('what is carried is worked out by the roll, not '
                          'given')
    if test_kind is not None:
        given['test_kind'] = _test_kind(test_kind)
    fields = {**current.header.values, **given}
    carried = [name for name in current.header.values if name not in given]
    if carried:
        fields['carried'] = carried
    fields = check_fields('test_opened', fields)      # refused before writing
    if fields.get('test_kind') not in TEST_KINDS:
        fields['test_kind'] = _test_kind(fields.get('test_kind'))
    number = model.next_test()
    closing = log.append('test_closed', author=author, test=current.number,
                         fields={'reason': reason or f'rolled to test '
                                                     f'{number}'},
                         event_time=event_time)
    opening = log.append('test_opened', author=author, test=number,
                         fields=fields, text=text,
                         event_time=closing.event_time, files=files)
    return closing, opening

close_test

close_test(log: RunLog, test: int, *, author: str, reason: str = '', event_time: datetime | None = None) -> Entry

Close test test to runs without opening another. It still takes findings, post-test steps and dispositions until its release.

Source code in src/visualdynamics/core/run_log_model.py

def close_test(log: RunLog, test: int, *, author: str, reason: str = '',
               event_time: datetime | None = None) -> Entry:
    """Close test `test` to runs without opening another. It still takes
    findings, post-test steps and dispositions until its release."""
    found = RunLogModel(log).test(test)
    if found.state != 'open':
        raise RunLogError(f'test {test} is already {found.state}')
    return log.append('test_closed', author=author, test=test,
                      fields={'reason': reason}, event_time=event_time)

change_header

change_header(log: RunLog, test: int, field: str, value: Any, *, author: str, text: str = '', event_time: datetime | None = None) -> Entry

Change one field of test test's header, the value it had recorded beside the new one. A carried field changed is confirmed. Once the test is running, a change of a limit, tolerance, abort band, planned notch or hold point raises a hold until a deviation names it.

Source code in src/visualdynamics/core/run_log_model.py

def change_header(log: RunLog, test: int, field: str, value: Any, *,
                  author: str, text: str = '',
                  event_time: datetime | None = None) -> Entry:
    """Change one field of test `test`'s header, the value it had
    recorded beside the new one. A carried field changed is confirmed.
    Once the test is running, a change of a limit, tolerance, abort band,
    planned notch or hold point raises a hold until a deviation names
    it."""
    model = RunLogModel(log)
    admit(model, 'header_change', test)
    was = model.test(test).header.get(field)
    fields = {'field': field, 'value': value}
    if was is not None:
        fields['was'] = was
    return log.append('header_change', author=author, test=test,
                      fields=fields, text=text, event_time=event_time)

confirm

confirm(log: RunLog, test: int, names: Iterable[str] | None = None, *, author: str, event_time: datetime | None = None) -> list[Entry]

Confirm carried fields of test test as they stand (every carried field when names is None): a header change of each to its own value, which is a confirmation, not a change. Refused for a field that is not carried.

Source code in src/visualdynamics/core/run_log_model.py

def confirm(log: RunLog, test: int, names: Iterable[str] | None = None, *,
            author: str, event_time: datetime | None = None) -> list[Entry]:
    """Confirm carried fields of test `test` as they stand (every carried
    field when `names` is None): a header change of each to its own
    value, which is a confirmation, not a change. Refused for a field
    that is not carried."""
    model = RunLogModel(log)
    found = model.test(test)
    names = list(found.header.carried if names is None else names)
    for name in names:
        if not found.header.is_carried(name):
            raise RunLogError(f'{name!r} is not carried in test {test}; the '
                              'carried fields are: '
                              + (', '.join(found.header.carried) or 'none'))
    admit(model, 'header_change', test)
    return [log.append('header_change', author=author, test=test,
                       fields={'field': name,
                               'value': found.header.get(name),
                               'was': found.header.get(name)},
                       text='confirmed as carried', event_time=event_time)
            for name in names]

release

release(log: RunLog, test: int, *, author: str, text: str = '', event_time: datetime | None = None) -> Entry

Write test test's release, the release checklist's 'release' line: the article and facility released and the setup free to be broken. The names typed are recorded; the log tracks no signatures. Refused while the test is open to runs.

Source code in src/visualdynamics/core/run_log_model.py

def release(log: RunLog, test: int, *, author: str, text: str = '',
            event_time: datetime | None = None) -> Entry:
    """Write test `test`'s release, the release checklist's 'release'
    line: the article and facility released and the setup free to be
    broken. The names typed are recorded; the log tracks no signatures.
    Refused while the test is open to runs."""
    found = RunLogModel(log).test(test)
    if found.state == 'open':
        raise RunLogError(f'test {test} is open to runs; close it (or roll) '
                          'before its release')
    if found.released is not None:
        raise RunLogError(f'test {test} is already released, at '
                          f'#{found.released.id}')
    return log.append('post_test_step', author=author, test=test,
                      fields={'step': 'release'}, text=text,
                      event_time=event_time)