visualdynamics.gui.run_log_dock

run_log_dock

The Run Log window: a dock in the main window where the log keeper writes the run log while a test runs.

The log keeper is a named person who is not at the controller, so the dock is built to be read across a room and typed into fast. It floats to a second monitor and remembers where it was put, and visualdynamics --run-log [file] opens the main window with the dock filling it and the analysis panes hidden (MainWindow.set_log_only), for a lab that analyzes its data elsewhere. Nothing here drives the controller: every check is a reminder, never an interlock.

What it shows, in large type:

• The status strip: the current test, its axis, run and level, whether the run is running or stopped (in words), the time since the run started, and the holds, counted, with the list a click away. A log open read-only says who holds it, and a mirror that could not be written says so.
• The tests as a strip of tabs, the whole log first, so an earlier test is a click away.
• The day as a list in the order written, with a line for each date: a corrected entry is shown struck through with what corrected it beneath, and a late entry is marked with when it was entered.
• The owed lists, shown and never enforced.

What it takes:

• The level row: one entry per step, its level offered from the header's step schedule, then the voltage, current and drive fraction in tab order, the analyst's answer with answers by channel, and the go; Enter writes it. The projection to full level (run_log_model.project) is shown live as the row is typed, a check that cannot be made saying why.
• The abort, one keystroke (Ctrl+Shift+A) that takes the event time at the key and opens a form needing only the kind of stop and the author's initials; Enter writes the run stop and the abort together, and what the first minute still owes is listed.
• Small forms for the rest: a new or opened log, a test closed or released, a clock check, a person in or out, a note, a run started or stopped, a cause found and a correction. Each takes its event time when it opens.
• The header and readings forms (gui.run_log_forms): the test header in full for a test opened, the roll and the current test, with carried fields marked until confirmed; a sweep, a shot, a survey's result, a notch, a deviation, an anomaly and its disposition, a setting change and an opinion given, each computing what the log will judge as it is typed, and the predictions the window drew shown beside what is read.

Every read and write of the file, and every rebuild of the model from it, runs on one worker thread of its own (LogWorker), in order, and never gates input: the keeper keeps typing while an entry is synced, a mirror is copied or a file is hashed. The window only ever shows a Snapshot, which the worker builds after each change.

Classes:

Name	Description
Row	One entry of the day: as corrected, as written, what each of its
TestTab	A test as the strip of tests names it: its number, title, kind and
Status	What the status strip reads: the current test (None when no test
Snapshot	A log as the window shows it, built on the worker after each
LogWorker	One worker thread for everything that touches a log's file: opening
StatusStrip	The strip across the top: the current test, its axis, run and
EntryForm	A small form for one act: its event time, taken when the form
LevelRow	The level row: one entry per step of the step-up. The level is
RunLogPanel	Everything inside the Run Log dock: the status strip, the holds,
RunLogDock	The Run Log dock: movable, floatable to a second monitor, and

Functions:

Name	Description
snapshot	The window's view of log: its model read once (RunLogModel)
entry_line	An entry on one line, as the day list shows it: its time, kind,
next_level	The level a level row offers: the first step of the header's step
projection_text	The projection of a level row's readings to full level, in words,
channel_answers	Answers by channel as typed in a level row: 'channel: answer'
planned_notches	A header's planned notches as typed: items separated by
notches_text	Planned notches as planned_notches reads them back.
hashed	Each file at paths as a FileRef, read and hashed now: called
event_time	The event time a form gives: taken (the moment the form was
remember_place	Store where dock is: floating, with its geometry, or docked in
restore_place	Put dock where it was last remembered (remember_place): back to
log_argument	Take --run-log [file] out of the command line: the arguments

Classes

Row dataclass

Row(entry: Entry, stored: Entry, corrections: tuple[str, ...], late: bool)

One entry of the day: as corrected, as written, what each of its corrections changed in words, and whether it was entered late.

Attributes:

Name	Type	Description
corrected	bool	Whether a correction changed the entry.

Attributes

corrected property

corrected: bool

Whether a correction changed the entry.

TestTab dataclass

TestTab(number: int, title: str, kind: str, state: str)

A test as the strip of tests names it: its number, title, kind and state.

Status dataclass

Status(test: int | None = None, title: str = '', kind: str = '', axis: str | None = None, run: int | None = None, level_db: float | None = None, outcome: str = '', started: datetime | None = None, stopped: datetime | None = None)

What the status strip reads: the current test (None when no test is open to runs), its title and kind, the axis, run and level of its latest run, that run's outcome ('running', 'aborted', 'stopped', 'completed', or '' with no run yet), and when it started and stopped.

Snapshot dataclass

Snapshot(path: str, title: str, writable: bool, read_only: str, mirror: str | None, mirror_problem: str, problems: tuple[str, ...], rows: tuple[Row, ...], tests: tuple[TestTab, ...], status: Status, header: Mapping[str, Any], holds: tuple[Hold, ...], owed: tuple[OwedList, ...], run: int | None, levels: tuple[float, ...], open_aborts: tuple[int, ...], unreleased: tuple[int, ...], carried: tuple[str, ...] = (), sweeps: tuple[Mapping[str, Any], ...] = (), pulses: PulseCount | None = None, deviation_wanted: tuple[int, ...] = (), open_anomalies: tuple[str, ...] = ())

A log as the window shows it, built on the worker after each change (snapshot), so nothing on the window's thread reads the file or rebuilds the model.

writable is False for a log opened read-only, read_only saying why (who holds it). rows is the corrected view with what each correction changed; holds and owed are the whole log's; header is the current test's header as it stands. run is the run a level row goes into (the latest, while it runs, otherwise the next) and levels the levels already written in it. open_aborts are the aborts with no cause, unreleased the tests closed to runs and not yet released.

For the readings forms: carried names the current header's fields carried from the last test and not yet confirmed; sweeps are the fields of the sweeps already written in the run a sweep goes into; pulses is the current test's pulse count against its budget (None with no test open); deviation_wanted are the entries whose hold a deviation naming them answers (an unplanned notch, a change to the test), and open_anomalies the numbers of the anomalies with no disposition.

LogWorker

LogWorker(parent: QObject)

One worker thread for everything that touches a log's file: opening and reading it, each entry written and synced, the mirror copied, files hashed, and the model rebuilt after each change. Jobs run one at a time in the order given, so entries are written in the order the keeper wrote them.

Unlike the analysis verbs' runner, it never gates input: the window keeps taking keys and clicks while a job runs, and each result comes back through a queued signal to the callback given with it. A job holds no Qt object and no callback, so nothing of Qt's is let go on the worker; the callbacks stay on the window's thread, by number.

Methods:

Name	Description
submit	Run work on the worker; then(ok, value) is called on the
shut_down	Run last after every job already given, wait for it, and take

Attributes:

Name	Type	Description
pending	int	How many jobs have not yet come back.

Source code in src/visualdynamics/gui/run_log_dock.py

def __init__(self, parent: QObject) -> None:
    self._pool = ThreadPoolExecutor(max_workers=1,
                                    thread_name_prefix='run-log')
    self._relay = _Relay(parent)
    self._relay.done.connect(self._deliver)
    self._waiting: dict[int, Callable[[bool, Any], None]] = {}
    self._next = 0
    self._stopped = False

Attributes

pending property

pending: int

How many jobs have not yet come back.

Methods:

submit

submit(work: Callable[[], Any], then: Callable[[bool, Any], None]) -> None

Run work on the worker; then(ok, value) is called on the window's thread with its result, or with the exception it raised.

Source code in src/visualdynamics/gui/run_log_dock.py

def submit(self, work: Callable[[], Any],
           then: Callable[[bool, Any], None]) -> None:
    """Run `work` on the worker; `then(ok, value)` is called on the
    window's thread with its result, or with the exception it
    raised."""
    if self._stopped:
        then(False, RunLogError('the run log is closed'))
        return
    self._next += 1
    number = self._next
    self._waiting[number] = then
    relay = self._relay

    def job() -> None:
        try:
            outcome = (True, work())
        except Exception as error:  # noqa: BLE001 — handed back whole
            outcome = (False, error)
        relay.done.emit(number, *outcome)

    self._pool.submit(job)

shut_down

shut_down(last: Callable[[], Any] | None = None) -> None

Run last after every job already given, wait for it, and take no more. What comes back after this is not delivered.

Source code in src/visualdynamics/gui/run_log_dock.py

def shut_down(self, last: Callable[[], Any] | None = None) -> None:
    """Run `last` after every job already given, wait for it, and take
    no more. What comes back after this is not delivered."""
    if self._stopped:
        return
    self._stopped = True
    if last is not None:
        self._pool.submit(last)
    self._pool.shutdown(wait=True)
    self._waiting.clear()

StatusStrip

StatusStrip()

Bases: QFrame

The strip across the top: the current test, its axis, run and level, running or stopped in words, the time since the run started, the holds counted (the button shows their list), and a notice line for a log read-only or a mirror that could not be written.

Methods:

Name	Description
show_snapshot	Read the strip off seen (None: no log open) at time now.
tick	The time since the run started, to now while it runs; once it

Source code in src/visualdynamics/gui/run_log_dock.py

def __init__(self) -> None:
    super().__init__()
    self.setFrameShape(QFrame.Shape.StyledPanel)
    self.test_label: QLabel = QLabel('No log open')
    self.place_label: QLabel = QLabel('')
    self.state_label: QLabel = QLabel('')
    self.since_label: QLabel = QLabel('')
    self.holds_button: QToolButton = QToolButton()
    self.holds_button.setCheckable(True)
    self.holds_button.setText('Holds: 0')
    self.holds_button.setToolTip('Show the holds: what the log asks to be '
                                 'answered before the test goes on')
    self.notice_label: QLabel = QLabel('')
    self.notice_label.setWordWrap(True)
    self.notice_label.hide()
    for label in (self.test_label, self.place_label, self.state_label,
                  self.since_label, self.holds_button):
        _scaled(label, STRIP_SCALE, bold=label is self.state_label)
    _scaled(self.notice_label, TYPE_SCALE)
    for label in (self.test_label, self.place_label, self.state_label,
                  self.since_label, self.notice_label):
        label.setTextInteractionFlags(
            Qt.TextInteractionFlag.TextSelectableByMouse)
    row = QHBoxLayout()
    row.setContentsMargins(0, 0, 0, 0)
    row.addWidget(self.test_label, 3)
    row.addWidget(self.place_label, 3)
    row.addWidget(self.state_label, 2)
    row.addWidget(self.since_label, 1)
    row.addWidget(self.holds_button)
    column = QVBoxLayout(self)
    column.setContentsMargins(8, 6, 8, 6)
    column.addLayout(row)
    column.addWidget(self.notice_label)
    self._status = Status()

Methods:

show_snapshot

show_snapshot(seen: Snapshot | None, now: datetime) -> None

Read the strip off seen (None: no log open) at time now.

Source code in src/visualdynamics/gui/run_log_dock.py

def show_snapshot(self, seen: Snapshot | None, now: datetime) -> None:
    """Read the strip off `seen` (None: no log open) at time `now`."""
    if seen is None:
        self._status = Status()
        self.test_label.setText('No log open')
        for label in (self.place_label, self.state_label,
                      self.since_label):
            label.setText('')
        self.holds_button.setText('Holds: 0')
        self.notice_label.hide()
        return
    status = self._status = seen.status
    if status.test is None:
        self.test_label.setText('No test open')
        self.place_label.setText('')
    else:
        self.test_label.setText(f'Test {status.test}: {status.title} '
                                f'({status.kind})')
        parts = [f'axis {status.axis}' if status.axis else 'no axis']
        if status.run is not None:
            parts.append(f'run {status.run}')
        if status.level_db is not None:
            parts.append('full level' if status.level_db == 0
                         else f'{status.level_db:+g} dB')
        self.place_label.setText(', '.join(parts))
    word = {'running': 'RUNNING', 'aborted': 'STOPPED: aborted',
            'stopped': 'STOPPED: scheduled stop',
            'completed': 'STOPPED: completed'}.get(status.outcome,
                                                  'STOPPED')
    self.state_label.setText(word if status.test is not None else '')
    count = len(seen.holds)
    self.holds_button.setText(f'Holds: {count}')
    notices = []
    if not seen.writable:
        notices.append('Read-only: ' + (seen.read_only or 'opened to read'))
    if seen.mirror_problem:
        notices.append('Mirror: ' + seen.mirror_problem)
    if seen.problems:
        notices.append(f'{len(seen.problems)} line'
                       f'{"s" * (len(seen.problems) != 1)} not read or '
                       'not fitting: ' + '; '.join(seen.problems[:3]))
    self.notice_label.setText('\n'.join(notices))
    self.notice_label.setVisible(bool(notices))
    self.tick(now)

tick

tick(now: datetime) -> None

The time since the run started, to now while it runs; once it has stopped, how long it ran.

Source code in src/visualdynamics/gui/run_log_dock.py

def tick(self, now: datetime) -> None:
    """The time since the run started, to `now` while it runs; once it
    has stopped, how long it ran."""
    status = self._status
    if status.started is None:
        self.since_label.setText('')
    elif status.stopped is None:
        self.since_label.setText(
            f'{_span((now - status.started).total_seconds())} since start')
    else:
        self.since_label.setText(
            f'stopped after '
            f'{_span((status.stopped - status.started).total_seconds())}')

EntryForm

EntryForm(name: str, title: str, fields: Sequence[tuple[str, str, str, Any]], taken: datetime, *, required: Sequence[str] = (), text: str = 'Note', note: str = '', defaults: Mapping[str, Any] | None = None, with_time: bool = True, live: Callable[[EntryForm], str] | None = None)

Bases: QWidget

A small form for one act: its event time, taken when the form opened and editable for an event entered late, its fields, a line of free text, and Write and Cancel. Enter in any box writes; Esc cancels. What a box cannot hold is said on the form's message line, and nothing is written.

A field of kind 'heading' is a group's title, not a box. live, when given, is read on every change to a box and shown under the boxes: what the form computes from what is typed (a budget, a count, a prediction), without writing anything.

Methods:

Name	Description
refresh	Read live again from what the boxes hold now.
box_text	What box field holds, as text.
set_box	Put value in box field: typed into a text box, chosen in a
value	Box field read as its field holds it, None when blank.
values	Every box read as its field holds it, blank boxes left out.
write	Read the form and hand it to on_write; say why on the form if
say	Show message on the form's message line.
focus_first	Put the cursor in the first box after the time.

Source code in src/visualdynamics/gui/run_log_dock.py

def __init__(self, name: str, title: str,
             fields: Sequence[tuple[str, str, str, Any]], taken: datetime,
             *, required: Sequence[str] = (), text: str = 'Note',
             note: str = '', defaults: Mapping[str, Any] | None = None,
             with_time: bool = True,
             live: Callable[[EntryForm], str] | None = None) -> None:
    super().__init__()
    self.name: str = name
    self.taken: datetime = taken
    self.required: tuple[str, ...] = tuple(required)
    self.on_write: Callable[[dict[str, Any], str, datetime], None] | None \
        = None
    #: the entry a correction form corrects
    self.target: Entry | None = None
    self.boxes: dict[str, QWidget] = {}
    #: the label beside each box, by field
    self.labels: dict[str, QLabel] = {}
    #: the groups' titles, in order
    self.headings: list[str] = []
    self.live: Callable[[EntryForm], str] | None = live
    self._kinds: dict[str, str] = {}
    self._labels: dict[str, str] = {}
    column = QVBoxLayout(self)
    column.setContentsMargins(4, 4, 4, 4)
    self.title_label: QLabel = QLabel(title)
    _scaled(self.title_label, TYPE_SCALE, bold=True)
    column.addWidget(self.title_label)
    if note:
        said = QLabel(note)
        said.setWordWrap(True)
        column.addWidget(said)
    grid = QGridLayout()
    column.addLayout(grid)
    self.time_box: QLineEdit = QLineEdit(_clock(taken))
    self.time_box.setToolTip('When it happened: taken when this form '
                             'opened; type another time for an event '
                             'entered late')
    row = 0
    if with_time:
        grid.addWidget(QLabel('Event time'), row, 0)
        grid.addWidget(self.time_box, row, 1)
        row += 1
    defaults = dict(defaults or {})
    for field, label, kind, choices in fields:
        if kind not in self.BOX_KINDS:
            raise ValueError(f'{kind!r} is not a kind of box')
        if kind == 'heading':
            heading = QLabel(label)
            font = QFont(heading.font())
            font.setBold(True)
            heading.setFont(font)
            grid.addWidget(heading, row, 0, 1, 2)
            self.headings.append(label)
            row += 1
            continue
        box = self._box(kind, choices, defaults.get(field))
        marked = label + (' *' if field in self.required else '')
        said = QLabel(marked)
        grid.addWidget(said, row, 0)
        grid.addWidget(box, row, 1)
        self.boxes[field] = box
        self.labels[field] = said
        self._kinds[field] = kind
        self._labels[field] = label
        row += 1
    self.text_box: QLineEdit | None = None
    if text:
        self.text_box = QLineEdit()
        grid.addWidget(QLabel(text), row, 0)
        grid.addWidget(self.text_box, row, 1)
    self.live_label: QLabel = QLabel('')
    self.live_label.setWordWrap(True)
    self.live_label.setTextInteractionFlags(
        Qt.TextInteractionFlag.TextSelectableByMouse)
    # placed before it is hidden or shown: a label shown with no
    # parent is a window of its own
    column.addWidget(self.live_label)
    if live is None:
        self.live_label.hide()
    self.message: QLabel = QLabel('')
    self.message.setWordWrap(True)
    column.addWidget(self.message)
    buttons = QHBoxLayout()
    self.write_button: QPushButton = QPushButton('Write (Enter)')
    self.cancel_button: QPushButton = QPushButton('Cancel (Esc)')
    self.write_button.clicked.connect(self.write)
    self.cancel_button.clicked.connect(self.cancelled.emit)
    buttons.addWidget(self.write_button)
    buttons.addWidget(self.cancel_button)
    buttons.addStretch(1)
    column.addLayout(buttons)
    column.addStretch(1)
    for widget in self._inputs():
        widget.installEventFilter(self)
        if isinstance(widget, QLineEdit):
            widget.textChanged.connect(self.refresh)
        elif isinstance(widget, QComboBox):
            widget.currentTextChanged.connect(self.refresh)
    _scaled(self, TYPE_SCALE)
    self.refresh()

Methods:

refresh

refresh(*_args: Any) -> None

Read live again from what the boxes hold now.

Source code in src/visualdynamics/gui/run_log_dock.py

def refresh(self, *_args: Any) -> None:
    """Read `live` again from what the boxes hold now."""
    if self.live is None:
        return
    try:
        said = self.live(self)
    except RunLogError as error:
        said = f'not read: {error}'
    self.live_label.setText(said)

box_text

box_text(field: str) -> str

What box field holds, as text.

Source code in src/visualdynamics/gui/run_log_dock.py

def box_text(self, field: str) -> str:
    """What box `field` holds, as text."""
    box = self.boxes[field]
    box = getattr(box, 'line', box)
    if isinstance(box, QComboBox):
        return box.currentText()
    assert isinstance(box, QLineEdit)
    return box.text()

set_box

set_box(field: str, value: str) -> None

Put value in box field: typed into a text box, chosen in a list.

Source code in src/visualdynamics/gui/run_log_dock.py

def set_box(self, field: str, value: str) -> None:
    """Put `value` in box `field`: typed into a text box, chosen in a
    list."""
    box = self.boxes[field]
    box = getattr(box, 'line', box)
    if isinstance(box, QComboBox):
        box.setCurrentText(value)
    else:
        assert isinstance(box, QLineEdit)
        box.setText(value)

value

value(field: str) -> Any

Box field read as its field holds it, None when blank. RunLogError naming the box for one that cannot be read.

Source code in src/visualdynamics/gui/run_log_dock.py

def value(self, field: str) -> Any:
    """Box `field` read as its field holds it, None when blank.
    `RunLogError` naming the box for one that cannot be read."""
    kind = self._kinds[field]
    text = self.box_text(field)
    label = self._labels[field]
    if kind in ('text', 'path', 'choice'):
        return text.strip() or None
    if kind == 'number':
        return _number(text, label)
    if kind in ('whole', 'entry'):
        return _whole(text, label)
    if kind == 'numbers':
        return _numbers(text, label)
    if kind == 'texts':
        return _texts(text)
    if kind == 'notches':
        return planned_notches(text)
    if kind == 'files':
        return _paths(text)
    return {'yes': True, 'no': False}.get(text)

values

values() -> dict[str, Any]

Every box read as its field holds it, blank boxes left out. RunLogError naming the box for one that cannot be read, or for a required box left blank.

Source code in src/visualdynamics/gui/run_log_dock.py

def values(self) -> dict[str, Any]:
    """Every box read as its field holds it, blank boxes left out.
    `RunLogError` naming the box for one that cannot be read, or for
    a required box left blank."""
    values: dict[str, Any] = {}
    for field in self._kinds:
        value = self.value(field)
        if value is None and field in self.required:
            raise RunLogError(f'{self._labels[field]} is needed')
        if value is not None:
            values[field] = value
    return values

write

write() -> None

Read the form and hand it to on_write; say why on the form if it cannot be read.

Source code in src/visualdynamics/gui/run_log_dock.py

def write(self) -> None:
    """Read the form and hand it to `on_write`; say why on the form if
    it cannot be read."""
    try:
        values = self.values()
        when = event_time(self.time_box.text(), self.taken)
    except RunLogError as error:
        self.say(str(error))
        return
    text = self.text_box.text().strip() if self.text_box else ''
    if self.on_write is not None:
        self.on_write(values, text, when)

say

say(message: str) -> None

Show message on the form's message line.

Source code in src/visualdynamics/gui/run_log_dock.py

def say(self, message: str) -> None:
    """Show `message` on the form's message line."""
    self.message.setText(message)

focus_first

focus_first() -> None

Put the cursor in the first box after the time.

Source code in src/visualdynamics/gui/run_log_dock.py

def focus_first(self) -> None:
    """Put the cursor in the first box after the time."""
    inputs = self._inputs()
    (inputs[1] if len(inputs) > 1 else inputs[0]).setFocus()

LevelRow

LevelRow(clock: Callable[[], datetime])

Bases: QWidget

The level row: one entry per step of the step-up. The level is offered from the header's step schedule; the voltage, current and drive fraction follow in tab order, then the analyst's answer, the answers by channel, the go and who agreed it at a hold point. Enter in any box writes the row. The projection to full level is redrawn on every keystroke, and the event time is taken at the row's first keystroke.

Methods:

Name	Description
inputs	The boxes in tab order.
offer	The test's header the projection is judged against, and the
fields	The row as a level row entry's fields; RunLogError naming
redraw	The projection to full level of what the row holds now, and
write	Hand the row to on_write with its event time.
clear	Empty the row after it is written, ready for the next step.

Source code in src/visualdynamics/gui/run_log_dock.py

def __init__(self, clock: Callable[[], datetime]) -> None:
    super().__init__()
    self._clock = clock
    self.header: Mapping[str, Any] = {}
    self.started: datetime | None = None
    self._offered: float | None = None
    self.on_write: Callable[[dict[str, Any], datetime], None] | None = None
    self.level_box: QLineEdit = QLineEdit()
    self.voltage_box: QLineEdit = QLineEdit()
    self.current_box: QLineEdit = QLineEdit()
    self.drive_box: QLineEdit = QLineEdit()
    self.answer_box: QComboBox = QComboBox()
    self.answer_box.addItems(['', *ANSWERS])
    self.channels_box: QLineEdit = QLineEdit()
    self.channels_box.setPlaceholderText('12: not explained, 4: explained')
    self.go_box: QCheckBox = QCheckBox('Go')
    self.agreed_box: QLineEdit = QLineEdit()
    self.agreed_box.setPlaceholderText('at a hold point')
    grid = QGridLayout()
    boxes = (('Level (dB)', self.level_box), ('Voltage (V)',
                                               self.voltage_box),
             ('Current (A)', self.current_box),
             ('Drive fraction', self.drive_box),
             ('Answer', self.answer_box),
             ('By channel', self.channels_box), ('', self.go_box),
             ('Agreed by', self.agreed_box))
    for column, (label, box) in enumerate(boxes):
        grid.addWidget(QLabel(label), 0, column)
        grid.addWidget(box, 1, column)
    grid.setColumnStretch(5, 2)
    self.time_label: QLabel = QLabel('')
    self.projection_label: QLabel = QLabel('')
    self.projection_label.setWordWrap(True)
    self.projection_label.setTextInteractionFlags(
        Qt.TextInteractionFlag.TextSelectableByMouse)
    #: the drive predicted for the test's setup beside the drive read,
    #: given by the panel from the row's fields; None shows nothing
    self.predicted: Callable[[Mapping[str, Any]], str] | None = None
    self.prediction_label: QLabel = QLabel('')
    self.prediction_label.setWordWrap(True)
    self.prediction_label.setTextInteractionFlags(
        Qt.TextInteractionFlag.TextSelectableByMouse)
    self.message: QLabel = QLabel('')
    self.message.setWordWrap(True)
    title = QLabel('Level row (Enter writes it)')
    _scaled(title, TYPE_SCALE, bold=True)
    column_layout = QVBoxLayout(self)
    column_layout.setContentsMargins(4, 4, 4, 4)
    heading = QHBoxLayout()
    heading.addWidget(title)
    heading.addStretch(1)
    heading.addWidget(self.time_label)
    column_layout.addLayout(heading)
    column_layout.addLayout(grid)
    column_layout.addWidget(self.projection_label)
    column_layout.addWidget(self.prediction_label)
    column_layout.addWidget(self.message)
    column_layout.addStretch(1)
    for line in (self.level_box, self.voltage_box, self.current_box,
                 self.drive_box, self.channels_box, self.agreed_box):
        line.textEdited.connect(self._typed)
    self.answer_box.activated.connect(self._typed)
    self.go_box.clicked.connect(self._typed)
    for widget in self.inputs():
        widget.installEventFilter(self)
    for line in (self.level_box, self.voltage_box, self.current_box,
                 self.drive_box):
        line.textChanged.connect(self.redraw)
    _scaled(self, TYPE_SCALE)
    self.redraw()

Methods:

inputs

inputs() -> list[QWidget]

The boxes in tab order.

Source code in src/visualdynamics/gui/run_log_dock.py

def inputs(self) -> list[QWidget]:
    """The boxes in tab order."""
    return [self.level_box, self.voltage_box, self.current_box,
            self.drive_box, self.answer_box, self.channels_box,
            self.go_box, self.agreed_box]

offer

offer(header: Mapping[str, Any], level: float | None) -> None

The test's header the projection is judged against, and the level offered for the next row (left alone once typed into).

Source code in src/visualdynamics/gui/run_log_dock.py

def offer(self, header: Mapping[str, Any], level: float | None) -> None:
    """The test's header the projection is judged against, and the
    level offered for the next row (left alone once typed into)."""
    self.header = header
    self._offered = level
    if self.started is None:
        self._put_level()
    self.redraw()

fields

fields() -> dict[str, Any]

The row as a level row entry's fields; RunLogError naming the box that cannot be read.

Source code in src/visualdynamics/gui/run_log_dock.py

def fields(self) -> dict[str, Any]:
    """The row as a level row entry's fields; `RunLogError` naming
    the box that cannot be read."""
    fields: dict[str, Any] = {
        'level_db': _number(self.level_box.text(), 'level'),
        'voltage': _number(self.voltage_box.text(), 'voltage'),
        'current': _number(self.current_box.text(), 'current'),
        'drive_fraction': _number(self.drive_box.text(),
                                  'drive fraction'),
        'answer': self.answer_box.currentText() or None,
        'channel_answers': channel_answers(self.channels_box.text()),
        'go': True if self.go_box.isChecked() else None,
        'agreed_by': self.agreed_box.text().strip() or None,
    }
    return {name: value for name, value in fields.items()
            if value is not None}

redraw

redraw() -> None

The projection to full level of what the row holds now, and the drive predicted beside the drive read.

Source code in src/visualdynamics/gui/run_log_dock.py

def redraw(self) -> None:
    """The projection to full level of what the row holds now, and
    the drive predicted beside the drive read."""
    try:
        fields = self.fields()
    except RunLogError as error:
        self.projection_label.setText(f'not judged: {error}')
        return
    self.projection_label.setText(projection_text(
        self.header, fields.get('level_db'),
        voltage=fields.get('voltage'), current=fields.get('current'),
        drive_fraction=fields.get('drive_fraction')))
    self.prediction_label.setText(
        self.predicted(fields) if self.predicted is not None else '')

write

write() -> None

Hand the row to on_write with its event time.

Source code in src/visualdynamics/gui/run_log_dock.py

def write(self) -> None:
    """Hand the row to `on_write` with its event time."""
    try:
        fields = self.fields()
    except RunLogError as error:
        self.message.setText(str(error))
        return
    if not fields:
        self.message.setText('the row is empty')
        return
    when = self.started or self._clock()
    if self.on_write is not None:
        self.on_write(fields, when)

clear

clear() -> None

Empty the row after it is written, ready for the next step.

Source code in src/visualdynamics/gui/run_log_dock.py

def clear(self) -> None:
    """Empty the row after it is written, ready for the next step."""
    self.started = None
    self.time_label.setText('')
    self._put_level()
    for line in (self.voltage_box, self.current_box, self.drive_box,
                 self.channels_box, self.agreed_box):
        line.clear()
    self.answer_box.setCurrentIndex(0)
    self.go_box.setChecked(False)
    self.message.setText('')
    self.voltage_box.setFocus()

RunLogPanel

RunLogPanel(parent: QWidget | None = None)

Bases: QWidget

Everything inside the Run Log dock: the status strip, the holds, the tests, the day, the owed lists and the entry.

clock gives the event time an act takes; log_clock and who are handed to the log for its entered-at times and the login and machine every entry records (each the computer's own unless set).

Methods:

Name	Description
show_layout	Name the layout the toggle would switch to.
new_log	Create a log at path and open it for writing, on the worker.
open_log	Open the log at path for writing, on the worker. One that
ask_open_log	Ask for a log to open.
close_log	Close the log and release its lock, on the worker.
reload	Read the log again: a log read-only shows what its writer has
shut_down	Close the log, releasing its lock, after every entry already
settle	Turn the event loop until every job given to the worker has
write_level	Write a level row to the current run of the current test.
abort	The abort: the event time is taken now, at the key, and the
open_form	Open the form name, its event time taken now: a small form
offer_prediction	Show prediction beside what is read: a notch prediction
focus_level_row	Close any open form and put the cursor in the level row.
say	Show message under the entry.
show_test	Show test test in the day and the owed lists (ALL for the
day_texts	The day list's lines as shown, for a script or a test.
owed_texts	The owed lists' lines as shown, each list's title first.

Source code in src/visualdynamics/gui/run_log_dock.py

def __init__(self, parent: QWidget | None = None) -> None:
    super().__init__(parent)
    self.clock: Callable[[], datetime] = _now
    self.log_clock: Callable[[], datetime] | None = None
    self.who: tuple[str, str] | None = None
    self.snapshot: Snapshot | None = None
    self.worker: LogWorker = LogWorker(self)
    self._log: RunLog | None = None
    self._shown_test: Any = ALL
    self._last_current: int | None = None
    self.form: EntryForm | None = None

    self.status: StatusStrip = StatusStrip()
    self.status.holds_button.toggled.connect(self._show_holds)
    self.holds_list: QListWidget = QListWidget()
    self.holds_list.setWordWrap(True)
    self.holds_list.hide()
    _scaled(self.holds_list, TYPE_SCALE)

    self.tests_bar: QTabBar = QTabBar()
    self.tests_bar.setExpanding(False)
    self.tests_bar.currentChanged.connect(self._test_picked)
    _scaled(self.tests_bar, TYPE_SCALE)

    self.day_list: QListWidget = QListWidget()
    self.day_list.setWordWrap(True)
    self.day_list.setAlternatingRowColors(True)
    _scaled(self.day_list, TYPE_SCALE)
    self.owed_tree: QTreeWidget = QTreeWidget()
    self.owed_tree.setHeaderLabels(['Owed (shown, not enforced)'])
    _scaled(self.owed_tree, TYPE_SCALE)

    self.initials_box: QLineEdit = QLineEdit()
    self.initials_box.setPlaceholderText('initials')
    self.initials_box.setMaximumWidth(120)
    self.axis_box: QLineEdit = QLineEdit()
    self.axis_box.setPlaceholderText('axis')
    self.axis_box.setMaximumWidth(120)
    self.run_label: QLabel = QLabel('')
    self.message: QLabel = QLabel('')
    self.message.setWordWrap(True)
    from .run_log_forms import Predictions, drive_text

    #: the predictions the window drew for its project, shown beside
    #: what is read (`offer_prediction`)
    self.predictions: Predictions = Predictions()
    self.level_row: LevelRow = LevelRow(lambda: self.clock())
    self.level_row.on_write = self.write_level
    self.level_row.predicted = lambda fields: drive_text(
        self.predictions, self.level_row.header, fields)
    self.level_row.redraw()
    self.entry_stack: QStackedWidget = QStackedWidget()
    self.entry_stack.addWidget(self.level_row)
    self._form_holder: QScrollArea | None = None

    self.acts: QToolBar = QToolBar('Run log')
    self.forms_bar: QToolBar = QToolBar('Run log readings')
    self.actions_by_name: dict[str, QAction] = {}
    self._build_acts()

    context = QHBoxLayout()
    said = [QLabel('Initials'), QLabel('Axis')]
    context.addWidget(said[0])
    context.addWidget(self.initials_box)
    context.addWidget(said[1])
    context.addWidget(self.axis_box)
    context.addWidget(self.run_label, 1)
    for widget in (*said, self.initials_box, self.axis_box,
                   self.run_label, self.message):
        _scaled(widget, TYPE_SCALE)

    entry = QWidget()
    entry_column = QVBoxLayout(entry)
    entry_column.setContentsMargins(0, 0, 0, 0)
    entry_column.addLayout(context)
    entry_column.addWidget(self.entry_stack, 1)
    entry_column.addWidget(self.message)

    day = QSplitter(Qt.Orientation.Horizontal)
    day.addWidget(self.day_list)
    day.addWidget(self.owed_tree)
    day.setStretchFactor(0, 3)
    day.setStretchFactor(1, 1)
    self.split: QSplitter = QSplitter(Qt.Orientation.Vertical)
    self.split.addWidget(day)
    self.split.addWidget(entry)
    self.split.setStretchFactor(0, 3)
    self.split.setStretchFactor(1, 2)

    column = QVBoxLayout(self)
    column.setContentsMargins(4, 4, 4, 4)
    column.addWidget(self.acts)
    column.addWidget(self.forms_bar)
    column.addWidget(self.status)
    column.addWidget(self.holds_list)
    column.addWidget(self.tests_bar)
    column.addWidget(self.split, 1)

    # the time since start moves once a second; nothing else is
    # read on the tick, so it costs a label
    self._ticker = QTimer(self)
    self._ticker.setInterval(1000)
    self._ticker.timeout.connect(lambda: self.status.tick(self.clock()))
    self._ticker.start()
    self._show(None)

Methods:

show_layout

show_layout(log_only: bool) -> None

Name the layout the toggle would switch to.

Source code in src/visualdynamics/gui/run_log_dock.py

def show_layout(self, log_only: bool) -> None:
    """Name the layout the toggle would switch to."""
    self.layout_action.setText('Whole Window' if log_only else 'Log Only')

new_log

new_log(path: str, title: str = '', mirror: str | None = None) -> None

Create a log at path and open it for writing, on the worker.

Source code in src/visualdynamics/gui/run_log_dock.py

def new_log(self, path: str, title: str = '',
            mirror: str | None = None) -> None:
    """Create a log at `path` and open it for writing, on the worker."""
    log_clock, who = self.log_clock, self.who
    self._swap(lambda: (RunLog.create(path, title=title, mirror=mirror,
                                      clock=log_clock, who=who), ''))

open_log

open_log(path: str, mirror: str | None = None) -> None

Open the log at path for writing, on the worker. One that cannot be written here (another writer holds it, or it is of an older format version) opens read-only, saying why.

Source code in src/visualdynamics/gui/run_log_dock.py

def open_log(self, path: str, mirror: str | None = None) -> None:
    """Open the log at `path` for writing, on the worker. One that
    cannot be written here (another writer holds it, or it is of an
    older format version) opens read-only, saying why."""
    log_clock, who = self.log_clock, self.who

    def opened() -> tuple[RunLog, str]:
        try:
            return RunLog.open(path, mirror=mirror, clock=log_clock,
                               who=who), ''
        except RunLogError as refused:
            # read where it cannot be written; a file that cannot be
            # read either raises what reading it says
            return RunLog.read(path), str(refused)

    self._swap(opened)

ask_open_log

ask_open_log() -> None

Ask for a log to open.

Source code in src/visualdynamics/gui/run_log_dock.py

def ask_open_log(self) -> None:
    """Ask for a log to open."""
    path, _ = QFileDialog.getOpenFileName(self, 'Open Run Log', '',
                                          f'Run log (*{SUFFIX})')
    if path:
        self.open_log(path)

close_log

close_log() -> None

Close the log and release its lock, on the worker.

Source code in src/visualdynamics/gui/run_log_dock.py

def close_log(self) -> None:
    """Close the log and release its lock, on the worker."""
    log, self._log = self._log, None
    if log is None:
        return
    self.worker.submit(log.close, lambda ok, value: (
        self._show(None), self.log_changed.emit('')))

reload

reload() -> None

Read the log again: a log read-only shows what its writer has added since.

Source code in src/visualdynamics/gui/run_log_dock.py

def reload(self) -> None:
    """Read the log again: a log read-only shows what its writer has
    added since."""
    if self.snapshot is None:
        return
    if self._log is not None and self._log.writable:
        log = self._log
        self.worker.submit(lambda: snapshot(log), self._showing)
        return
    path, why = self.snapshot.path, self.snapshot.read_only

    def work() -> tuple[RunLog, Snapshot]:
        log = RunLog.read(path)
        return log, snapshot(log, read_only=why)

    def then(ok: bool, value: Any) -> None:
        if not ok:
            self.say(str(value))
            return
        self._log, seen = value
        self._show(seen)

    self.worker.submit(work, then)

shut_down

shut_down() -> None

Close the log, releasing its lock, after every entry already given is written; then stop the worker. The window calls this as it closes.

Source code in src/visualdynamics/gui/run_log_dock.py

def shut_down(self) -> None:
    """Close the log, releasing its lock, after every entry already
    given is written; then stop the worker. The window calls this as
    it closes."""
    self._ticker.stop()
    log, self._log = self._log, None
    self.worker.shut_down(log.close if log is not None else None)

settle

settle(timeout: float = 10.0) -> bool

Turn the event loop until every job given to the worker has come back, or timeout seconds pass; whether it settled. For a script or a test driving the dock.

Source code in src/visualdynamics/gui/run_log_dock.py

def settle(self, timeout: float = 10.0) -> bool:
    """Turn the event loop until every job given to the worker has
    come back, or `timeout` seconds pass; whether it settled. For a
    script or a test driving the dock."""
    import time as clock

    from PySide6.QtWidgets import QApplication

    deadline = clock.monotonic() + timeout
    while self.worker.pending and clock.monotonic() < deadline:
        QApplication.processEvents()
        clock.sleep(0.005)
    QApplication.processEvents()
    return not self.worker.pending

write_level

write_level(fields: dict[str, Any], when: datetime) -> None

Write a level row to the current run of the current test.

Source code in src/visualdynamics/gui/run_log_dock.py

def write_level(self, fields: dict[str, Any], when: datetime) -> None:
    """Write a level row to the current run of the current test."""
    author = self._author()
    if author is None:
        return
    test, axis, run = self._place()
    written = self._writing(
        lambda log: model_module.append(
            log, 'level_row', author=author, test=test, axis=axis,
            run=run, fields=fields, event_time=when),
        done=lambda _entry: self.level_row.clear(),
        refused=self.level_row.message.setText)
    if written:
        self.level_row.message.setText('')

abort

abort() -> EntryForm | None

The abort: the event time is taken now, at the key, and the form needs only the kind of stop and the author; the rest of the first minute is offered and, left blank, owed.

Source code in src/visualdynamics/gui/run_log_dock.py

def abort(self) -> EntryForm | None:
    """The abort: the event time is taken now, at the key, and the
    form needs only the kind of stop and the author; the rest of the
    first minute is offered and, left blank, owed."""
    taken = self.clock()
    fields = [('stop', 'Kind of stop', 'choice', ('', *STOPS))]
    fields += [(name, name.replace('_', ' ').capitalize(),
                'number' if FIELD_TYPES[name] == 'number' else 'text',
                None)
               for name in KINDS['abort'].optional]
    form = EntryForm('abort', f'ABORT at {_clock(taken)}', fields, taken,
                     required=('stop',), text='What was seen',
                     note='Only the kind of stop is needed now; what is '
                          'left blank is listed as owed.')
    form.on_write = self._write_abort
    self._open(form)
    form.boxes['stop'].setFocus()
    return form

open_form

open_form(name: str) -> EntryForm | None

Open the form name, its event time taken now: a small form ('note', 'clock check', 'person in', 'person out', 'run start', 'run stop', 'cause found', 'correct', 'close test', 'release' or 'new log'), a header form ('open test', 'roll' or 'header'), or a readings form ('sweep', 'shot', 'result', 'notch', 'deviation', 'anomaly', 'disposition', 'setting change' or 'opinion').

Source code in src/visualdynamics/gui/run_log_dock.py

def open_form(self, name: str) -> EntryForm | None:
    """Open the form `name`, its event time taken now: a small form
    ('note', 'clock check', 'person in', 'person out', 'run start',
    'run stop', 'cause found', 'correct', 'close test', 'release' or
    'new log'), a header form ('open test', 'roll' or 'header'), or a
    readings form ('sweep', 'shot', 'result', 'notch', 'deviation',
    'anomaly', 'disposition', 'setting change' or 'opinion')."""
    from . import run_log_forms as forms

    taken = self.clock()
    seen = self.snapshot
    if name == 'header' and (seen is None or seen.status.test is None):
        self.say('No test is open: open one first, and its header is '
                 'here.')
        return None
    builder = {
        'note': lambda: EntryForm('note', 'Note', [], taken,
                                  text='Note'),
        'clock check': lambda: EntryForm(
            'clock check', 'Clock check',
            [('offset_s', 'Recorders less this computer (s)', 'number',
              None), ('recorder', 'Recorder', 'text', None),
             ('method', 'Method', 'text', None)], taken,
            required=('offset_s',)),
        'person in': lambda: EntryForm(
            'person in', 'Person in',
            [('person', 'Person', 'text', None),
             ('role', 'Role', 'text', None)], taken,
            required=('person',)),
        'person out': lambda: EntryForm(
            'person out', 'Person out',
            [('person', 'Person', 'text', None),
             ('role', 'Role', 'text', None)], taken,
            required=('person',)),
        'run start': lambda: EntryForm(
            'run start', 'Run start',
            [('level_db', 'Level (dB)', 'number', None),
             ('counter', 'Counter', 'text', None)], taken),
        'run stop': lambda: EntryForm(
            'run stop', 'Run stop (not an abort)',
            [('counter', 'Counter', 'text', None),
             ('scheduled', 'Scheduled stop', 'flag', None),
             ('result', 'Result', 'text', None),
             ('by', 'Run by', 'text', None),
             ('reason', 'Reason', 'text', None)], taken),
        'cause found': lambda: EntryForm(
            'cause found', 'Cause found',
            [('cause', 'Cause', 'text', None),
             ('refers_to', 'Of entry #', 'entry', None)], taken,
            required=('cause',),
            defaults={'refers_to': seen.open_aborts[-1]
                      if seen and seen.open_aborts else None}),
        'open test': lambda: forms.header_form('open test', taken, seen),
        'roll': lambda: forms.header_form('roll', taken, seen),
        'header': lambda: forms.header_form('header', taken, seen),
        'sweep': lambda: forms.sweep_form(taken, seen),
        'shot': lambda: forms.shot_form(taken, seen, self._axis()),
        'result': lambda: forms.result_form(taken),
        'notch': lambda: forms.notch_form(taken, seen, self._axis(),
                                          lambda: self.predictions),
        'deviation': lambda: forms.deviation_form(taken, seen),
        'anomaly': lambda: forms.anomaly_form(taken),
        'disposition': lambda: forms.disposition_form(taken, seen),
        'setting change': lambda: forms.setting_form(taken),
        'opinion': lambda: forms.opinion_form(taken),
        'close test': lambda: EntryForm(
            'close test', 'Close the test to runs',
            [('reason', 'Reason', 'text', None)], taken, text=''),
        'release': lambda: EntryForm(
            'release', 'Release a test',
            [('test', 'Test', 'whole', None)], taken,
            required=('test',), text='Released by',
            defaults={'test': seen.unreleased[-1]
                      if seen and seen.unreleased else None}),
        'new log': lambda: EntryForm(
            'new log', 'New run log',
            [('path', 'File', 'path', None),
             ('title', 'Article or program', 'text', None),
             ('mirror', 'Second copy (optional)', 'path', None)], taken,
            required=('path',), text='', with_time=False),
        'correct': lambda: self._correct_form(taken),
    }.get(name)
    if builder is None:
        raise ValueError(f'{name!r} is not a form of the run log')
    form = builder()
    if form is None:
        return None
    form.on_write = (lambda values, text, when, name=name:
                     self._write_form(name, values, text, when))
    self._open(form)
    form.focus_first()
    return form

offer_prediction

offer_prediction(kind: str, prediction: Any, source: str) -> None

Show prediction beside what is read: a notch prediction (kind 'limit', core.limit_prediction.LimitPrediction) beside a notch entry, a drive prediction ('drive', core.drive_prediction.DrivePrediction) beside the level row's drive. source names the objects it was made from. The window offers each prediction it draws; None takes it away.

Source code in src/visualdynamics/gui/run_log_dock.py

def offer_prediction(self, kind: str, prediction: Any,
                     source: str) -> None:
    """Show `prediction` beside what is read: a notch prediction
    (`kind` 'limit', `core.limit_prediction.LimitPrediction`) beside
    a notch entry, a drive prediction ('drive',
    `core.drive_prediction.DrivePrediction`) beside the level row's
    drive. `source` names the objects it was made from. The window
    offers each prediction it draws; None takes it away."""
    from dataclasses import replace

    if kind == 'limit':
        self.predictions = replace(self.predictions, limit=prediction,
                                   limit_from=source)
    elif kind == 'drive':
        self.predictions = replace(self.predictions, drive=prediction,
                                   drive_from=source)
    else:
        raise ValueError(f'{kind!r} is not a kind of prediction; it is '
                         'limit or drive')
    self.level_row.redraw()
    if self.form is not None:
        self.form.refresh()

focus_level_row

focus_level_row() -> None

Close any open form and put the cursor in the level row.

Source code in src/visualdynamics/gui/run_log_dock.py

def focus_level_row(self) -> None:
    """Close any open form and put the cursor in the level row."""
    self._close_form()
    self.level_row.voltage_box.setFocus()

say

say(message: str) -> None

Show message under the entry.

Source code in src/visualdynamics/gui/run_log_dock.py

def say(self, message: str) -> None:
    """Show `message` under the entry."""
    self.message.setText(message)

show_test

show_test(test: Any) -> None

Show test test in the day and the owed lists (ALL for the whole log).

Source code in src/visualdynamics/gui/run_log_dock.py

def show_test(self, test: Any) -> None:
    """Show test `test` in the day and the owed lists (`ALL` for the
    whole log)."""
    for index in range(self.tests_bar.count()):
        if self.tests_bar.tabData(index) == test:
            self.tests_bar.setCurrentIndex(index)
            return
    raise KeyError(f'the log has no test {test}')

day_texts

day_texts() -> list[str]

The day list's lines as shown, for a script or a test.

Source code in src/visualdynamics/gui/run_log_dock.py

def day_texts(self) -> list[str]:
    """The day list's lines as shown, for a script or a test."""
    return [self.day_list.item(i).text()
            for i in range(self.day_list.count())]

owed_texts

owed_texts() -> list[str]

The owed lists' lines as shown, each list's title first.

Source code in src/visualdynamics/gui/run_log_dock.py

def owed_texts(self) -> list[str]:
    """The owed lists' lines as shown, each list's title first."""
    texts: list[str] = []
    for i in range(self.owed_tree.topLevelItemCount()):
        top = self.owed_tree.topLevelItem(i)
        texts.append(top.text(0))
        texts.extend(top.child(j).text(0)
                     for j in range(top.childCount()))
    return texts

RunLogDock

RunLogDock(parent: QWidget | None = None)

Bases: QDockWidget

The Run Log dock: movable, floatable to a second monitor, and closable, holding a RunLogPanel. Its place is remembered between launches (remember_place, restore_place).

Source code in src/visualdynamics/gui/run_log_dock.py

def __init__(self, parent: QWidget | None = None) -> None:
    super().__init__('Run Log', parent)
    # the name a main window's saved state knows a dock by
    self.setObjectName('run_log_dock')
    self.setFeatures(QDockWidget.DockWidgetFeature.DockWidgetMovable
                     | QDockWidget.DockWidgetFeature.DockWidgetFloatable
                     | QDockWidget.DockWidgetFeature.DockWidgetClosable)
    self.panel: RunLogPanel = RunLogPanel()
    self.setWidget(self.panel)
    #: set while `restore_place` moves the dock, so the moves it makes
    #: are not remembered as the keeper's
    self.placing: bool = False

Functions:

snapshot

snapshot(log: RunLog, *, read_only: str = '') -> Snapshot

The window's view of log: its model read once (RunLogModel) and everything the dock shows taken from it, in plain values.

Run on the log's worker, never on the window's thread: it reads the whole corrected view and every hold and owed rule.

Source code in src/visualdynamics/gui/run_log_dock.py

def snapshot(log: RunLog, *, read_only: str = '') -> Snapshot:
    """The window's view of `log`: its model read once (`RunLogModel`)
    and everything the dock shows taken from it, in plain values.

    Run on the log's worker, never on the window's thread: it reads the
    whole corrected view and every hold and owed rule.
    """
    model = RunLogModel(log)
    rows = tuple(
        Row(entry, log.entry(entry.id),
            tuple(f'{_correction_text(log, correction)} (#{correction.id}, '
                  f'{correction.author}, {_clock(correction.event_time)})'
                  for correction in log.corrections_of(entry.id)),
            entry.is_late())
        for entry in model.entries)
    tests = tuple(TestTab(test.number, test.title, test.kind, test.state)
                  for test in model.tests())
    problems = [str(problem) for problem in log.problems]
    for test in model.tests():
        problems.extend(test.problems)
    current = model.current
    status = Status()
    header: Mapping[str, Any] = {}
    run_number = None
    levels: tuple[float, ...] = ()
    sweeps: tuple[Mapping[str, Any], ...] = ()
    if current is not None:
        header = dict(current.header.values)
        latest = max(current.runs, key=lambda run: run.number, default=None)
        if latest is None:
            status = Status(current.number, current.title, current.kind,
                            current.axes[-1] if current.axes else None)
            run_number = model.next_run(current.number)
        else:
            level = None
            for entry in latest.entries:
                if isinstance(entry.fields.get('level_db'), (int, float)):
                    level = float(entry.fields['level_db'])
            running = latest.outcome == 'running'
            stop = latest.stop
            status = Status(current.number, current.title, current.kind,
                            latest.axis, latest.number, level, latest.outcome,
                            latest.entries[0].event_time,
                            None if running or stop is None
                            else stop.event_time)
            run_number = (latest.number if running
                          else model.next_run(current.number))
            if running:
                levels = tuple(float(entry.fields['level_db'])
                               for entry in latest.levels
                               if 'level_db' in entry.fields)
                sweeps = tuple(dict(entry.fields) for entry in latest.sweeps)
    holds = tuple(model.holds())
    return Snapshot(
        path=log.path, title=log.header.title, writable=log.writable,
        read_only=read_only, mirror=log.mirror,
        mirror_problem=log.mirror_problem, problems=tuple(problems),
        rows=rows, tests=tests, status=status, header=header, holds=holds,
        owed=tuple(model.owed()), run=run_number, levels=levels,
        open_aborts=tuple(hold.entry.id for hold in holds
                          if hold.rule == 'abort-cause'),
        unreleased=tuple(test.number for test in model.tests()
                         if test.state == 'closed to runs'),
        carried=tuple(current.header.carried) if current else (),
        sweeps=sweeps,
        pulses=model.pulses(current.number) if current else None,
        deviation_wanted=tuple(hold.entry.id for hold in holds
                               if hold.rule in ('unplanned-notch',
                                                'plan-change')),
        open_anomalies=tuple(found.number
                             for found in model.open_anomalies()))

entry_line

entry_line(entry: Entry) -> str

An entry on one line, as the day list shows it: its time, kind, place, fields as the plain text export writes them, free text and author.

Source code in src/visualdynamics/gui/run_log_dock.py

def entry_line(entry: Entry) -> str:
    """An entry on one line, as the day list shows it: its time, kind,
    place, fields as the plain text export writes them, free text and
    author."""
    parts = [_clock(entry.event_time), entry.label, _where(entry)]
    fields = _fields_text(entry, entry.fields)
    if fields:
        parts.append(fields)
    if entry.text:
        parts.append(f'"{entry.text}"')
    parts.append(entry.author)
    return '   '.join(parts)

next_level

next_level(steps: Any, written: Sequence[float]) -> float | None

The level a level row offers: the first step of the header's step schedule above the last level written in the run, the first step when none is, the last written when the schedule is used up or absent, and None with neither.

Source code in src/visualdynamics/gui/run_log_dock.py

def next_level(steps: Any, written: Sequence[float]) -> float | None:
    """The level a level row offers: the first step of the header's step
    schedule above the last level written in the run, the first step
    when none is, the last written when the schedule is used up or
    absent, and None with neither."""
    schedule = sorted(float(step) for step in steps or ()
                      if isinstance(step, (int, float))
                      and not isinstance(step, bool))
    last = written[-1] if written else None
    if not schedule:
        return last
    if last is None:
        return schedule[0]
    higher = [step for step in schedule if step > last + LEVEL_MATCH_DB]
    return higher[0] if higher else last

projection_text

projection_text(header: Mapping[str, Any], level: Any, *, voltage: Any = None, current: Any = None, drive_fraction: Any = None) -> str

The projection of a level row's readings to full level, in words, against the test's header (run_log_model.project): the factor, the verdict and every check, a check that cannot be made saying why.

Source code in src/visualdynamics/gui/run_log_dock.py

def projection_text(header: Mapping[str, Any], level: Any, *,
                    voltage: Any = None, current: Any = None,
                    drive_fraction: Any = None) -> str:
    """The projection of a level row's readings to full level, in words,
    against the test's header (`run_log_model.project`): the factor, the
    verdict and every check, a check that cannot be made saying why."""
    if level is None:
        return 'not judged: no level is given'
    try:
        projection = model_module.project(
            level, voltage=voltage, current=current,
            drive_fraction=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'))
    except RunLogError as error:
        return f'not judged: {error}'
    return str(projection)

channel_answers

channel_answers(text: str) -> dict[str, str] | None

Answers by channel as typed in a level row: 'channel: answer' items separated by commas or semicolons, such as '12: not explained, 4: as predicted'. None for nothing typed; RunLogError naming the item for one without a colon or with an answer that is not one of run_log.ANSWERS.

Source code in src/visualdynamics/gui/run_log_dock.py

def channel_answers(text: str) -> dict[str, str] | None:
    """Answers by channel as typed in a level row: 'channel: answer'
    items separated by commas or semicolons, such as '12: not
    explained, 4: as predicted'. None for nothing typed; `RunLogError`
    naming the item for one without a colon or with an answer that is
    not one of `run_log.ANSWERS`."""
    answers: dict[str, str] = {}
    for item in text.replace(';', ',').split(','):
        if not item.strip():
            continue
        name, colon, answer = item.partition(':')
        answer = answer.strip().lower()
        if not colon or not name.strip():
            raise RunLogError(f'channel answers: {item.strip()!r} is not '
                              '"channel: answer"')
        if answer not in ANSWERS:
            raise RunLogError(f'channel {name.strip()}: {answer!r} is not an '
                              'answer; it is one of: ' + ', '.join(ANSWERS))
        answers[name.strip()] = answer
    return answers or None

planned_notches

planned_notches(text: str) -> list[dict[str, Any]] | None

A header's planned notches as typed: items separated by semicolons, each its band with the axis before a colon and the depth after a comma where the plan gives them, such as 'Z: 180-220 Hz, 6 dB; 400-450 Hz'. None for nothing typed; RunLogError naming the item for a band run_log.band_hz does not read or a depth that is not a number of dB.

Source code in src/visualdynamics/gui/run_log_dock.py

def planned_notches(text: str) -> list[dict[str, Any]] | None:
    """A header's planned notches as typed: items separated by
    semicolons, each its band with the axis before a colon and the depth
    after a comma where the plan gives them, such as 'Z: 180-220 Hz, 6
    dB; 400-450 Hz'. None for nothing typed; `RunLogError` naming the
    item for a band `run_log.band_hz` does not read or a depth that is
    not a number of dB."""
    notches: list[dict[str, Any]] = []
    for item in text.split(';'):
        if not item.strip():
            continue
        axis, colon, rest = item.partition(':')
        if not colon:
            axis, rest = '', item
        band, comma, depth = rest.partition(',')
        notch: dict[str, Any] = {}
        if axis.strip():
            notch['axis'] = axis.strip()
        if not band.strip():
            raise RunLogError(f'planned notches: {item.strip()!r} gives no '
                              'band')
        band_hz(band.strip(), f'planned notches: {item.strip()!r}: its band')
        notch['band'] = band.strip()
        if comma:
            number = depth.strip()
            if number.lower().endswith('db'):
                number = number[:-2].strip()
            notch['depth_db'] = _number(number, f'planned notches: '
                                                f'{item.strip()!r}: its depth')
            if notch['depth_db'] is None:
                raise RunLogError(f'planned notches: {item.strip()!r}: the '
                                  'depth after the comma is blank')
        notches.append(notch)
    return notches or None

notches_text

notches_text(notches: Any) -> str

Planned notches as planned_notches reads them back.

Source code in src/visualdynamics/gui/run_log_dock.py

def notches_text(notches: Any) -> str:
    """Planned notches as `planned_notches` reads them back."""
    items = []
    for notch in notches or ():
        band = notch.get('band')
        if not isinstance(band, str):
            band = f'{_number_text(band[0])}-{_number_text(band[1])} Hz'
        text = f'{notch["axis"]}: {band}' if notch.get('axis') else band
        if notch.get('depth_db') is not None:
            text += f', {_number_text(notch["depth_db"])} dB'
        items.append(text)
    return '; '.join(items)

hashed

hashed(paths: Sequence[str]) -> list[FileRef]

Each file at paths as a FileRef, read and hashed now: called on the log's worker as the entry naming them is written, never on the window's thread. RunLogError naming a file that cannot be read.

Source code in src/visualdynamics/gui/run_log_dock.py

def hashed(paths: Sequence[str]) -> list[FileRef]:
    """Each file at `paths` as a `FileRef`, read and hashed now: called
    on the log's worker as the entry naming them is written, never on
    the window's thread. `RunLogError` naming a file that cannot be
    read."""
    refs = []
    for path in paths:
        try:
            refs.append(file_ref(path))
        except OSError as error:
            raise RunLogError(f'{path} could not be read to attach it: '
                              f'{error.strerror or error}') from None
    return refs

event_time

event_time(text: str, taken: datetime) -> datetime

The event time a form gives: taken (the moment the form was opened) while its box still reads that time, otherwise the time typed (HH:MM or HH:MM:SS) on taken's date and zone, so a keeper entering a late event writes when it happened.

Source code in src/visualdynamics/gui/run_log_dock.py

def event_time(text: str, taken: datetime) -> datetime:
    """The event time a form gives: `taken` (the moment the form was
    opened) while its box still reads that time, otherwise the time
    typed (HH:MM or HH:MM:SS) on `taken`'s date and zone, so a keeper
    entering a late event writes when it happened."""
    text = text.strip()
    if not text or text == _clock(taken):
        return taken
    for pattern in ('%H:%M:%S', '%H:%M'):
        try:
            typed = datetime.strptime(text, pattern).time()
            break
        except ValueError:
            continue
    else:
        raise RunLogError(f'event time: {text!r} is not a time of day '
                          '(HH:MM or HH:MM:SS)')
    return datetime.combine(taken.date(), time(typed.hour, typed.minute,
                                               typed.second),
                            tzinfo=taken.tzinfo)

remember_place

remember_place(dock: QDockWidget, area: DockWidgetArea) -> None

Store where dock is: floating, with its geometry, or docked in area.

Source code in src/visualdynamics/gui/run_log_dock.py

def remember_place(dock: QDockWidget, area: Qt.DockWidgetArea) -> None:
    """Store where `dock` is: floating, with its geometry, or docked in
    `area`."""
    from .preferences import settings

    store = settings()
    store.setValue(PLACE_FLOATING, bool(dock.isFloating()))
    store.setValue(PLACE_AREA, int(area.value))
    if dock.isFloating():
        store.setValue(PLACE_GEOMETRY, dock.saveGeometry())
    store.sync()

restore_place

restore_place(dock: QDockWidget, window: Any) -> None

Put dock where it was last remembered (remember_place): back to its floating geometry, or into its dock area of window; the right edge when nothing is remembered.

Source code in src/visualdynamics/gui/run_log_dock.py

def restore_place(dock: QDockWidget, window: Any) -> None:
    """Put `dock` where it was last remembered (`remember_place`): back to
    its floating geometry, or into its dock area of `window`; the right
    edge when nothing is remembered."""
    from .preferences import settings

    store = settings()
    floating = str(store.value(PLACE_FLOATING, 'false')).lower() == 'true'
    geometry = store.value(PLACE_GEOMETRY)
    try:
        area = Qt.DockWidgetArea(int(store.value(
            PLACE_AREA, Qt.DockWidgetArea.RightDockWidgetArea.value)))
    except (TypeError, ValueError):
        area = Qt.DockWidgetArea.RightDockWidgetArea
    if area not in (Qt.DockWidgetArea.LeftDockWidgetArea,
                    Qt.DockWidgetArea.RightDockWidgetArea,
                    Qt.DockWidgetArea.TopDockWidgetArea,
                    Qt.DockWidgetArea.BottomDockWidgetArea):
        area = Qt.DockWidgetArea.RightDockWidgetArea
    placing, dock.placing = getattr(dock, 'placing', False), True
    try:
        window.addDockWidget(area, dock)
        if floating:
            dock.setFloating(True)
            if isinstance(geometry, QByteArray) and not geometry.isEmpty():
                with contextlib.suppress(Exception):
                    dock.restoreGeometry(geometry)
    finally:
        dock.placing = placing

log_argument

log_argument(argv: Sequence[str]) -> tuple[list[str], bool, str | None]

Take --run-log [file] out of the command line: the arguments left, whether the flag was given, and the log named after it (an argument ending in .vdlog), or None.

Source code in src/visualdynamics/gui/run_log_dock.py

def log_argument(argv: Sequence[str]) -> tuple[list[str], bool, str | None]:
    """Take `--run-log [file]` out of the command line: the arguments
    left, whether the flag was given, and the log named after it (an
    argument ending in `.vdlog`), or None."""
    rest: list[str] = []
    wanted = False
    path = None
    skip = False
    for index, argument in enumerate(argv):
        if skip:
            skip = False
            continue
        if argument == '--run-log':
            wanted = True
            following = argv[index + 1] if index + 1 < len(argv) else None
            if following is not None and following.lower().endswith(SUFFIX):
                path = following
                skip = True
        elif argument.startswith('--run-log='):
            wanted = True
            path = argument.split('=', 1)[1] or None
        else:
            rest.append(argument)
    if path is not None:
        path = os.path.abspath(path)
    return rest, wanted, path