visualdynamics.core.run_log

run_log

The run log: what happened on the floor, in order, as it happened.

A run log is kept while a test runs, by a log keeper who is not at the controller. Each thing written down (a level reached, an abort, a clock check, a person arriving, a note) is one entry. An entry is never edited: a mistake is put right by a correction, a later entry that names the one it corrects and carries the new values, so the reader sees the corrected view and the record keeps both.

The file is JSON Lines (.vdlog): UTF-8 text, one JSON object per line, each line written, flushed and synced to the disk before the writer returns, so a crash loses at most the entry being written. The first line is the header, with the format's stamp and version. A file cut off in the middle of its last line is read up to that line, and the torn line is named (RunLog.problems); any other line that cannot be read is skipped and named the same way, and the rest is read.

Versions. FORMAT_VERSION is the newest version this build reads, and the one it writes. Each version has its own reader, so a later build reads every earlier version, and a file from a newer version is refused with a message to update. While Visual Dynamics is an alpha the format may still change; from the first release that is not an alpha, an older file always opens.

Integrity. The entries of each test are chained by SHA-256: every entry carries the hash of the entry before it in the same test and the hash of its own canonical form, which includes that previous hash. The entries between tests have a chain of their own. RunLog.verify checks every chain and names the first broken link by test and entry. The chain detects accidental damage, a file cut short and casual edits. It is not proof against deliberate alteration: anyone who edits an entry can compute every hash after it again, and the chain then checks. A file cut off at the end of an entry also leaves a valid chain; it is found only against a head hash recorded outside the file (RunLog.head, which a report's data index can carry). Every entry records the computer's login and machine name beside the author's typed initials; those are recorded, not authenticated.

One writer. A log open for writing holds a lock file beside it (<log>.lock); a second writer is refused with a message naming who holds it, and readers open the file read-only (RunLog.read). The lock is the operating system's, so a writer that crashes leaves no lock behind. Each entry can also be copied to a second path (mirror, the lab's share or the data folder); a mirror that cannot be written is reported (RunLog.mirror_problem) and never costs the entry its place in the log itself.

The entry kinds and the fields each may carry are one registry (KINDS). Each kind requires as little as it can: an abort needs only its time, the kind of stop and the author, because the first minute after a stop is the worst moment to be refused. What is still owed of an entry is a derived view, not a refusal.

Classes:

Name	Description
EntryKind	One kind of entry: its name in the file, its label in words, the
RunLogError	A run log, or an entry for one, that cannot be read or written
RunLogLocked	The log is already open for writing elsewhere; the message names
FileRef	A file an entry names: its name, its SHA-256 as 64 lower-case hex
Header	A log's first line: its format version, its id, when and where it
Entry	One entry, as stored, or as corrected (RunLog.view).
ReadProblem	A line of the file that was not read as an entry: its number (the
Break	A broken link found by RunLog.verify: the chain ('test N', or
Verification	What RunLog.verify found: every break, in the order of the file,
RunLog	A run log: its header, its entries in the order written, and, when

Functions:

Name	Description
file_ref	A FileRef for the file at path, hashed in chunks
chain_name	The name of the chain an entry with test reference test belongs
identity	(login, machine name) of this computer, each '' where the system
canonical	The canonical text of a stored entry (or header) that its hash is
band_hz	A band as (low, high) in Hz, from two frequencies or from text
check_fields	The fields of a new entry of kind, checked against KINDS.

Classes

EntryKind dataclass

EntryKind(name: str, label: str, required: tuple[str, ...] = (), optional: tuple[str, ...] = (), needs_test: bool = False, machine: bool = False)

One kind of entry: its name in the file, its label in words, the fields it needs and the fields it may carry.

needs_test is set for the kinds that mean nothing without a test (a test opened or closed). machine marks the kinds a lab's own machine log keeps: the trips, the setting changes, the clock checks, people in and out, the restarts, and the tests opened and closed that say what was running.

Attributes:

Name	Type	Description
fields	tuple[str, ...]	Every field the kind may carry, the required first.

Attributes

fields property

fields: tuple[str, ...]

Every field the kind may carry, the required first.

RunLogError

Bases: ValueError

A run log, or an entry for one, that cannot be read or written one way; the message names the cause.

RunLogLocked

Bases: RunLogError

The log is already open for writing elsewhere; the message names who holds it.

FileRef dataclass

FileRef(name: str, sha256: str, size: int | None = None, format: str = '', holds: str = '', path: str = '')

A file an entry names: its name, its SHA-256 as 64 lower-case hex digits, and its size in bytes when known.

format, holds and path say, when known, the file's format, what it holds (its run and channels) and where it was when it was hashed, in the words a data index records them (data_index.IndexEntry), so the log's files go into a report's data index as they are. Each is left out of the file when empty.

Methods:

Name	Description
to_dict	The file as the log stores it.

Methods:

to_dict

to_dict() -> dict[str, Any]

The file as the log stores it.

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

def to_dict(self) -> dict[str, Any]:
    """The file as the log stores it."""
    stored: dict[str, Any] = {'name': self.name, 'sha256': self.sha256}
    if self.size is not None:
        stored['size'] = self.size
    for label in ('format', 'holds', 'path'):
        if getattr(self, label):
            stored[label] = getattr(self, label)
    return stored

Header dataclass

Header(version: int, log_id: str, created: datetime, title: str = '', login: str = '', computer: str = '')

A log's first line: its format version, its id, when and where it was created, and its title.

Entry dataclass

Entry(id: int, kind: str, event_time: datetime, entered_at: datetime, author: str, login: str = '', computer: str = '', test: int | None = None, axis: str | None = None, run: int | None = None, fields: Mapping[str, Any] = dict(), text: str = '', files: tuple[FileRef, ...] = (), corrects: int | None = None, prev: str = '', hash: str = '', corrected_by: tuple[int, ...] = (), line: int = 0)

One entry, as stored, or as corrected (RunLog.view).

event_time is when the thing happened and entered_at when it was written down, each with its date and its UTC offset. test, axis and run are plain references (a test and run by number, an axis by name), any of them absent. corrects is the id of the entry a correction corrects. prev and hash are the entry's links in its test's chain. In the corrected view, corrected_by lists the corrections applied, in order. line is the entry's line in the file, the header being line 1.

Methods:

Name	Description
is_late	Whether the entry was written down more than after after
to_dict	The entry as the log stores it, its hash included.

Attributes:

Name	Type	Description
label	str	The kind in words, as KINDS names it.
chain	str	The chain the entry belongs to: 'test N', or 'log' for an

Attributes

label property

label: str

The kind in words, as KINDS names it.

chain property

chain: str

The chain the entry belongs to: 'test N', or 'log' for an entry between tests.

Methods:

is_late

is_late(after: timedelta = LATE_AFTER) -> bool

Whether the entry was written down more than after after the event it records. Both times carry their offsets, so a change of daylight saving time between them counts as the real interval, not the difference of the wall clocks.

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

def is_late(self, after: timedelta = LATE_AFTER) -> bool:
    """Whether the entry was written down more than `after` after
    the event it records. Both times carry their offsets, so a
    change of daylight saving time between them counts as the real
    interval, not the difference of the wall clocks."""
    return self.entered_at - self.event_time > after

to_dict

to_dict() -> dict[str, Any]

The entry as the log stores it, its hash included.

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

def to_dict(self) -> dict[str, Any]:
    """The entry as the log stores it, its hash included."""
    stored: dict[str, Any] = {
        'id': self.id, 'kind': self.kind,
        'event_time': _time_text(self.event_time),
        'entered_at': _time_text(self.entered_at),
        'author': self.author, 'login': self.login,
        'computer': self.computer, 'test': self.test,
        'axis': self.axis, 'run': self.run,
        'fields': dict(self.fields), 'text': self.text,
        'files': [ref.to_dict() for ref in self.files],
    }
    if self.corrects is not None:
        stored['corrects'] = self.corrects
    stored['prev'] = self.prev
    stored['hash'] = self.hash
    return stored

ReadProblem dataclass

ReadProblem(line: int, reason: str)

A line of the file that was not read as an entry: its number (the header is line 1) and why.

Break dataclass

Break(chain: str, test: int | None, entry: int | None, line: int | None, reason: str)

A broken link found by RunLog.verify: the chain ('test N', or 'log' for the entries between tests), the test's number, the entry's id and line where they are known, and what was found.

Verification dataclass

Verification(breaks: tuple[Break, ...], entries: int, chains: int)

What RunLog.verify found: every break, in the order of the file, and how many entries and chains were checked. ok only when nothing is broken; first is the first break.

Attributes:

Name	Type	Description
ok	bool	Whether every chain checked.
first	Break | None	The first broken link, or None.

Attributes

ok property

ok: bool

Whether every chain checked.

first property

first: Break | None

The first broken link, or None.

RunLog

RunLog(path: str, header: Header, header_raw: Mapping[str, Any], *, clock: Callable[[], datetime] | None = None, who: tuple[str, str] | None = None)

A run log: its header, its entries in the order written, and, when opened for writing, the way to add to it.

Make one with RunLog.create (a new file), RunLog.open (an existing file, for writing) or RunLog.read (any file, read-only). A log open for writing holds the lock until close; use it in a with block.

Methods:

Name	Description
create	A new run log at path, open for writing.
open	The run log at path, open for writing.
read	The run log at path, read-only, without the lock, so it can be
parse	A run log read from data, the bytes of a .vdlog, read-only:
close	Release the lock and the file. A read-only log has neither;
entry	The entry with id entry_id, as stored.
corrections_of	The corrections of entry entry_id, in the order written.
current	Entry entry_id with its corrections applied in order, each
view	The corrected view: every entry but the corrections, in the
tests	The test numbers the log's entries refer to, in the order
chains	The names of the log's chains, in the order first written:
head	The head of a test's chain (test None for the entries
next_number	The number the next finding of kind ('deviation' or
clock_check_for	The clock check in force at at (an entry, by its corrected
recorder_time	at on the recorders' clock: its time plus the offset of the
verify	Check every chain, and the heads recorded outside the file.
append	Write one entry, synced to the disk before this returns.
correct	Write a correction of an earlier entry, which stays as it was.

Attributes:

Name	Type	Description
writable	bool	Whether entries can be added: opened for writing and not yet
version	int	The format version the file is written in.
entries	tuple[Entry, ...]	Every entry as stored, corrections included, in the order
problems	tuple[ReadProblem, ...]	Every line that was not read as an entry, and why: a torn last

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

def __init__(self, path: str, header: Header,
             header_raw: Mapping[str, Any], *,
             clock: Callable[[], datetime] | None = None,
             who: tuple[str, str] | None = None) -> None:
    self.path: str = path
    self.header: Header = header
    self._header_raw = dict(header_raw)
    self._reader = _READERS[header.version]
    self._clock = clock or _local_now
    self._who = who if who is not None else identity()
    self._records: list[tuple[int, dict[str, Any]]] = []
    self._entries: list[Entry] = []
    self._by_id: dict[int, Entry] = {}
    self._corrections: dict[int, list[Entry]] = {}
    self._heads: dict[str, str] = {}
    self._problems: list[ReadProblem] = []
    self._lines = 1
    self._next_id = 1
    self._handle: Any = None
    self._lock: _Lock | None = None
    self._torn_tail = False
    self.mirror: str | None = None
    self.mirror_problem: str = ''
    self._mirror_in_step = False

Attributes

writable property

writable: bool

Whether entries can be added: opened for writing and not yet closed.

version property

version: int

The format version the file is written in.

entries property

entries: tuple[Entry, ...]

Every entry as stored, corrections included, in the order written.

problems property

problems: tuple[ReadProblem, ...]

Every line that was not read as an entry, and why: a torn last line, a line that is not JSON, a line that is not an entry.

Methods:

create classmethod

create(path: str | PathLike, *, title: str = '', mirror: str | PathLike | None = None, clock: Callable[[], datetime] | None = None, who: tuple[str, str] | None = None) -> RunLog

A new run log at path, open for writing.

Parameters:

Name	Type	Description	Default
path	str or PathLike	Where the log is written. Nothing may be there yet.	required
title	str	The log's title (the article or program it is kept for).	''
mirror	str or PathLike	A second path every entry is copied to as well.	None
clock	callable	Returns the time entries are entered at, with its zone; the computer's own clock unless given.	None
who	(str, str)	The (login, machine name) every entry records; read from the computer unless given.	None

Returns:

Type	Description
RunLog	The new log, holding only its header.

Raises:

Type	Description
RunLogError	When a file is already at path, or the title is not text.
RunLogLocked	When another writer holds the lock beside path.

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

@classmethod
def create(cls, path: str | os.PathLike, *, title: str = '',
           mirror: str | os.PathLike | None = None,
           clock: Callable[[], datetime] | None = None,
           who: tuple[str, str] | None = None) -> RunLog:
    """A new run log at `path`, open for writing.

    Parameters
    ----------
    path : str or os.PathLike
        Where the log is written. Nothing may be there yet.
    title : str, optional
        The log's title (the article or program it is kept for).
    mirror : str or os.PathLike, optional
        A second path every entry is copied to as well.
    clock : callable, optional
        Returns the time entries are entered at, with its zone; the
        computer's own clock unless given.
    who : (str, str), optional
        The (login, machine name) every entry records; read from the
        computer unless given.

    Returns
    -------
    RunLog
        The new log, holding only its header.

    Raises
    ------
    RunLogError
        When a file is already at `path`, or the title is not text.
    RunLogLocked
        When another writer holds the lock beside `path`.
    """
    path = os.fspath(path)
    if os.path.exists(path):
        raise RunLogError(f'a file is already at {path}; open it to add '
                          'to it, or choose another name')
    if not isinstance(title, str):
        raise RunLogError(f'a title is text, not {title!r}')
    login, computer = who if who is not None else identity()
    lock = _Lock(path, login, computer)
    try:
        raw = cls._new_header(title, login, computer, clock)
        with open(path, 'xb') as handle:
            _durable_append(handle, cls._line(raw))
        _sync_folder(path)
        log = cls(path, _READERS[FORMAT_VERSION].header(raw), raw,
                  clock=clock, who=(login, computer))
        log._take(lock, mirror)
    except BaseException:
        lock.release()
        raise
    return log

open classmethod

open(path: str | PathLike, *, mirror: str | PathLike | None = None, clock: Callable[[], datetime] | None = None, who: tuple[str, str] | None = None) -> RunLog

The run log at path, open for writing.

A torn last line (a crash in the middle of an entry) is named in problems and ended, so the next entry starts on a line of its own; nothing in the file is removed. An empty file (a crash while it was being created) is given its header.

Parameters:

Name	Type	Description	Default
path	str or PathLike	The log.	required
mirror	str or PathLike	A second path every entry is copied to as well; the whole log is copied there first.	None
clock	callable	Returns the time entries are entered at, with its zone.	None
who	(str, str)	The (login, machine name) every entry records.	None

Returns:

Type	Description
RunLog	The log, with every entry read.

Raises:

Type	Description
RunLogError	When the file is not a run log, is of a newer format version, or is of an older version (which opens read-only, since a log is only ever written in one version).
RunLogLocked	When another writer holds the lock.
FileNotFoundError	When there is no file at path.

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

@classmethod
def open(cls, path: str | os.PathLike, *,
         mirror: str | os.PathLike | None = None,
         clock: Callable[[], datetime] | None = None,
         who: tuple[str, str] | None = None) -> RunLog:
    """The run log at `path`, open for writing.

    A torn last line (a crash in the middle of an entry) is named in
    `problems` and ended, so the next entry starts on a line of its
    own; nothing in the file is removed. An empty file (a crash
    while it was being created) is given its header.

    Parameters
    ----------
    path : str or os.PathLike
        The log.
    mirror : str or os.PathLike, optional
        A second path every entry is copied to as well; the whole
        log is copied there first.
    clock : callable, optional
        Returns the time entries are entered at, with its zone.
    who : (str, str), optional
        The (login, machine name) every entry records.

    Returns
    -------
    RunLog
        The log, with every entry read.

    Raises
    ------
    RunLogError
        When the file is not a run log, is of a newer format version,
        or is of an older version (which opens read-only, since a log
        is only ever written in one version).
    RunLogLocked
        When another writer holds the lock.
    FileNotFoundError
        When there is no file at `path`.
    """
    path = os.fspath(path)
    if not os.path.isfile(path):
        raise FileNotFoundError(f'no run log at {path}')
    login, computer = who if who is not None else identity()
    lock = _Lock(path, login, computer)
    try:
        with open(path, 'rb') as handle:
            data = handle.read()
        if not data.strip():
            raw = cls._new_header('', login, computer, clock)
            with open(path, 'wb') as handle:
                _durable_append(handle, cls._line(raw))
            log = cls(path, _READERS[FORMAT_VERSION].header(raw), raw,
                      clock=clock, who=(login, computer))
            log._problems.append(ReadProblem(
                1, 'the file was empty, as a crash while it was being '
                'created leaves it; its header was written'))
        else:
            log = cls._parse(path, data, clock=clock,
                             who=(login, computer))
            if log.header.version != FORMAT_VERSION:
                raise RunLogError(
                    f'{path} is in run log format version '
                    f'{log.header.version}; this build adds entries only '
                    f'to version {FORMAT_VERSION}. Read it with '
                    'RunLog.read, and start a new log to go on.')
        log._take(lock, mirror)
    except BaseException:
        lock.release()
        raise
    return log

read classmethod

read(path: str | PathLike) -> RunLog

The run log at path, read-only, without the lock, so it can be read while another program writes it.

Parameters:

Name	Type	Description	Default
path	str or PathLike	The log.	required

Returns:

Type	Description
RunLog	The log, every entry that could be read in it, and the lines that could not in problems.

Raises:

Type	Description
RunLogError	When the file is empty, is not a run log, or is of a newer format version than this build reads.
FileNotFoundError	When there is no file at path.

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

@classmethod
def read(cls, path: str | os.PathLike) -> RunLog:
    """The run log at `path`, read-only, without the lock, so it can be
    read while another program writes it.

    Parameters
    ----------
    path : str or os.PathLike
        The log.

    Returns
    -------
    RunLog
        The log, every entry that could be read in it, and the lines
        that could not in `problems`.

    Raises
    ------
    RunLogError
        When the file is empty, is not a run log, or is of a newer
        format version than this build reads.
    FileNotFoundError
        When there is no file at `path`.
    """
    path = os.fspath(path)
    if not os.path.isfile(path):
        raise FileNotFoundError(f'no run log at {path}')
    with open(path, 'rb') as handle:
        data = handle.read()
    return cls.parse(data, path)

parse classmethod

parse(data: bytes, path: str = '<memory>') -> RunLog

A run log read from data, the bytes of a .vdlog, read-only: read without the file, for a log held in memory (a snapshot kept in a project) or bytes read once and examined more than one way.

Parameters:

Name	Type	Description	Default
data	bytes	The log's bytes, its header line first.	required
path	str	What the log is called in messages, and its path.	'<memory>'

Returns:

Type	Description
RunLog	The log, every entry that could be read in it, and the lines that could not in problems.

Raises:

Type	Description
RunLogError	When data is empty, is not a run log, or is of a newer format version than this build reads.

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

@classmethod
def parse(cls, data: bytes, path: str = '<memory>') -> RunLog:
    """A run log read from `data`, the bytes of a `.vdlog`, read-only:
    `read` without the file, for a log held in memory (a snapshot
    kept in a project) or bytes read once and examined more than
    one way.

    Parameters
    ----------
    data : bytes
        The log's bytes, its header line first.
    path : str, optional
        What the log is called in messages, and its `path`.

    Returns
    -------
    RunLog
        The log, every entry that could be read in it, and the lines
        that could not in `problems`.

    Raises
    ------
    RunLogError
        When `data` is empty, is not a run log, or is of a newer
        format version than this build reads.
    """
    if not data.strip():
        raise RunLogError(f'{path} is empty: it has no header line, so it '
                          'holds no log yet')
    return cls._parse(path, data)

close

close() -> None

Release the lock and the file. A read-only log has neither; closing it does nothing.

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

def close(self) -> None:
    """Release the lock and the file. A read-only log has neither;
    closing it does nothing."""
    if self._handle is not None:
        self._handle.close()
        self._handle = None
    if self._lock is not None:
        self._lock.release()
        self._lock = None

entry

entry(entry_id: int) -> Entry

The entry with id entry_id, as stored.

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

def entry(self, entry_id: int) -> Entry:
    """The entry with id `entry_id`, as stored."""
    if entry_id not in self._by_id:
        raise KeyError(f'the log has no entry #{entry_id}')
    return self._by_id[entry_id]

corrections_of

corrections_of(entry_id: int) -> tuple[Entry, ...]

The corrections of entry entry_id, in the order written.

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

def corrections_of(self, entry_id: int) -> tuple[Entry, ...]:
    """The corrections of entry `entry_id`, in the order written."""
    return tuple(self._corrections.get(entry_id, ()))

current

current(entry_id: int) -> Entry

Entry entry_id with its corrections applied in order, each one's values replacing the entry's and a value of None striking a field; corrected_by lists them. The stored entry is unchanged (entry).

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

def current(self, entry_id: int) -> Entry:
    """Entry `entry_id` with its corrections applied in order, each
    one's values replacing the entry's and a value of None striking
    a field; `corrected_by` lists them. The stored entry is
    unchanged (`entry`)."""
    entry = self.entry(entry_id)
    corrections = self.corrections_of(entry_id)
    if not corrections:
        return entry
    fields = dict(entry.fields)
    changed: dict[str, Any] = {}
    for correction in corrections:
        for name, value in correction.fields.items():
            if name in CORRECTABLE:
                changed[name] = value
            elif value is None:
                fields.pop(name, None)
            else:
                fields[name] = value
    kept = {}
    for name, value in changed.items():
        try:
            if name == 'event_time':
                kept[name] = _parse_time(value, 'a corrected event time')
            elif name == 'text':
                kept[name] = '' if value is None else str(value)
            elif name == 'axis':
                kept[name] = _axis(value)
            else:
                kept[name] = _reference(value, name)
        except RunLogError:
            # an unreadable value in a hand-edited file: the stored
            # value stands, and the correction is still listed
            continue
    return replace(entry, fields=fields,
                   corrected_by=tuple(c.id for c in corrections), **kept)

view

view() -> list[Entry]

The corrected view: every entry but the corrections, in the order written, each with its corrections applied (current).

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

def view(self) -> list[Entry]:
    """The corrected view: every entry but the corrections, in the
    order written, each with its corrections applied
    (`current`)."""
    return [self.current(entry.id) for entry in self._entries
            if entry.kind != 'correction']

tests

tests() -> list[int]

The test numbers the log's entries refer to, in the order first referred to.

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

def tests(self) -> list[int]:
    """The test numbers the log's entries refer to, in the order
    first referred to."""
    seen: list[int] = []
    for entry in self._entries:
        if entry.test is not None and entry.test not in seen:
            seen.append(entry.test)
    return seen

chains

chains() -> list[str]

The names of the log's chains, in the order first written: 'log' for the entries between tests, 'test N' for each test.

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

def chains(self) -> list[str]:
    """The names of the log's chains, in the order first written:
    'log' for the entries between tests, 'test N' for each test."""
    seen: list[str] = []
    for _, stored in self._records:
        name = chain_name(stored.get('test'))
        if name not in seen:
            seen.append(name)
    return seen

head

head(test: int | None = None) -> str

The head of a test's chain (test None for the entries between tests): the hash of its last entry, which the next entry will carry as its previous hash. A chain with no entries yet has its starting hash. A head recorded outside the file (printed at the test's close, in a report's data index) is what verify(heads=...) finds a file cut short against.

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

def head(self, test: int | None = None) -> str:
    """The head of a test's chain (`test` None for the entries
    between tests): the hash of its last entry, which the next entry
    will carry as its previous hash. A chain with no entries yet has
    its starting hash. A head recorded outside the file (printed at
    the test's close, in a report's data index) is what
    `verify(heads=...)` finds a file cut short against."""
    name = chain_name(test)
    return self._heads.get(name) or _genesis(self._header_raw, name)

next_number

next_number(kind: str) -> str

The number the next finding of kind ('deviation' or 'anomaly') takes, numbered across the whole log: D-1, D-2, A-1.

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

def next_number(self, kind: str) -> str:
    """The number the next finding of `kind` ('deviation' or
    'anomaly') takes, numbered across the whole log: D-1, D-2, A-1."""
    return findings.next_number(self._numbers(), kind)

clock_check_for

clock_check_for(at: Entry | datetime, recorder: str | None = None) -> Entry | None

The clock check in force at at (an entry, by its corrected event time, or a time with its zone): the latest clock check at or before it, of recorder when named. None when no clock check comes before it.

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

def clock_check_for(self, at: Entry | datetime,
                    recorder: str | None = None) -> Entry | None:
    """The clock check in force at `at` (an entry, by its corrected
    event time, or a time with its zone): the latest clock check at
    or before it, of `recorder` when named. None when no clock check
    comes before it."""
    when = self._instant(at)
    best: Entry | None = None
    for stored in self._entries:
        if stored.kind != 'clock_check':
            continue
        entry = self.current(stored.id)
        if entry.event_time > when:
            continue
        if recorder is not None and entry.fields.get('recorder') != recorder:
            continue
        if best is None or entry.event_time >= best.event_time:
            best = entry
    return best

recorder_time

recorder_time(at: Entry | datetime, recorder: str | None = None) -> datetime | None

at on the recorders' clock: its time plus the offset of the clock check in force (clock_check_for), which records the recorders' clock less this computer's, in seconds. None when no clock check comes before it.

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

def recorder_time(self, at: Entry | datetime,
                  recorder: str | None = None) -> datetime | None:
    """`at` on the recorders' clock: its time plus the offset of the
    clock check in force (`clock_check_for`), which records the
    recorders' clock less this computer's, in seconds. None when no
    clock check comes before it."""
    when = self._instant(at)
    check = self.clock_check_for(when, recorder)
    if check is None:
        return None
    return when + timedelta(seconds=float(check.fields['offset_s']))

verify

verify(heads: Mapping[int | None, str] | None = None, tests: Iterable[int | None] | None = None) -> Verification

Check every chain, and the heads recorded outside the file.

Each entry's previous hash must be the hash of the entry before it in its chain (the chain's starting hash for the first), and its hash must be the hash of its own canonical form. Entry ids must run from 1 without a gap. A head in heads (by test number, None for the entries between tests) must still be in its chain; one that is not means the file was cut short or changed after the head was recorded.

This detects accidental damage, a file cut short (mid-entry on reading; at an entry's end only against heads) and casual edits. It is not proof against deliberate alteration: an edit whose author computes every following hash again is not detected.

Parameters:

Name	Type	Description	Default
heads	mapping	Heads recorded outside the file, by test number.	None
tests	iterable	Check only these tests' chains (None for the entries between tests), as a file holding one test's entries is checked; the ids then need not run without a gap.	None

Returns:

Type	Description
Verification	Every break in the order of the file, and the first.

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

def verify(self, heads: Mapping[int | None, str] | None = None,
           tests: Iterable[int | None] | None = None) -> Verification:
    """Check every chain, and the heads recorded outside the file.

    Each entry's previous hash must be the hash of the entry before
    it in its chain (the chain's starting hash for the first), and
    its hash must be the hash of its own canonical form. Entry ids
    must run from 1 without a gap. A head in `heads` (by test number,
    None for the entries between tests) must still be in its chain;
    one that is not means the file was cut short or changed after
    the head was recorded.

    This detects accidental damage, a file cut short (mid-entry on
    reading; at an entry's end only against `heads`) and casual
    edits. It is not proof against deliberate alteration: an edit
    whose author computes every following hash again is not
    detected.

    Parameters
    ----------
    heads : mapping, optional
        Heads recorded outside the file, by test number.
    tests : iterable, optional
        Check only these tests' chains (None for the entries between
        tests), as a file holding one test's entries is checked;
        the ids then need not run without a gap.

    Returns
    -------
    Verification
        Every break in the order of the file, and the first.
    """
    breaks: list[tuple[float, Break]] = []
    last: dict[str, str] = {}
    seen: dict[str, set[str]] = {}
    only = None if tests is None else {chain_name(t) for t in tests}
    expected_id = 1
    for line, stored in self._records:
        test = stored.get('test')
        name = chain_name(test)
        if only is not None and name not in only:
            continue
        test = test if isinstance(test, int) else None
        entry_id = stored.get('id')
        entry_id = entry_id if isinstance(entry_id, int) else None
        previous = last.get(name) or _genesis(self._header_raw, name)
        if only is None and entry_id != expected_id:
            if entry_id is not None and entry_id > expected_id:
                gap = (f'entry #{expected_id} is missing' if
                       entry_id == expected_id + 1 else
                       f'entries #{expected_id} to #{entry_id - 1} are '
                       'missing')
                reason = f'the ids jump: {gap}'
            else:
                reason = (f'the id is out of order (#{expected_id} was '
                          'next)')
            breaks.append((line, Break(name, test, entry_id, line,
                                       reason)))
        if entry_id is not None:
            expected_id = max(expected_id, entry_id + 1)
        if stored.get('prev') != previous:
            breaks.append((line, Break(
                name, test, entry_id, line,
                'its previous hash is not the hash of the entry before '
                'it in the chain: an entry before it was removed, moved '
                'or changed')))
        try:
            computed = self._reader.hash(stored)
        except (TypeError, ValueError):
            computed = ''
        if stored.get('hash') != computed:
            breaks.append((line, Break(
                name, test, entry_id, line,
                'its hash does not match its contents: it was changed '
                'after it was written')))
        current = stored.get('hash') if isinstance(stored.get('hash'),
                                                   str) else computed
        last[name] = current
        seen.setdefault(name, set()).add(current)
    end = math.inf                               # after every line
    for test, recorded in (heads or {}).items():
        name = chain_name(test)
        if recorded != _genesis(self._header_raw, name) \
                and recorded not in seen.get(name, set()):
            breaks.append((end, Break(
                name, test, None, None,
                f'the head recorded for it ({recorded[:12]}...) is not '
                'in the log: the log was cut short, or changed, after '
                'the head was recorded')))
    breaks.sort(key=lambda found: found[0])
    return Verification(tuple(found for _, found in breaks),
                        len(self._records), len(last))

append

append(kind: str, *, author: str, fields: Mapping[str, Any] | None = None, text: str = '', event_time: datetime | None = None, test: int | None = None, axis: str | None = None, run: int | None = None, files: Iterable[str | PathLike | FileRef] = ()) -> Entry

Write one entry, synced to the disk before this returns.

A deviation or anomaly opened without a number takes the next of its kind across the log (next_number); a number given must be of its kind and not already taken in the log. A correction is written with correct, not here.

Parameters:

Name	Type	Description	Default
kind	str	One of KINDS.	required
author	str	The author's typed initials.	required
fields	mapping	The kind's fields (check_fields); a field set to None or blank is left out.	None
text	str	Free text.	''
event_time	datetime	When the thing happened, with its zone; now, if not given.	None
test	int	The test and run it belongs to, by number from 1.	None
run	int	The test and run it belongs to, by number from 1.	None
axis	str	The axis it belongs to, by name.	None
files	iterable	Files it names, each a path (hashed now) or a FileRef.	()

Returns:

Type	Description
Entry	The entry as written, with its id and hashes.

Raises:

Type	Description
RunLogError	For a log not open for writing, and for anything the entry cannot hold one way, the message naming the field.

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

def append(self, kind: str, *, author: str,
           fields: Mapping[str, Any] | None = None, text: str = '',
           event_time: datetime | None = None, test: int | None = None,
           axis: str | None = None, run: int | None = None,
           files: Iterable[str | os.PathLike | FileRef] = (),
           ) -> Entry:
    """Write one entry, synced to the disk before this returns.

    A deviation or anomaly opened without a number takes the next of
    its kind across the log (`next_number`); a number given must be
    of its kind and not already taken in the log. A correction is written with
    `correct`, not here.

    Parameters
    ----------
    kind : str
        One of `KINDS`.
    author : str
        The author's typed initials.
    fields : mapping, optional
        The kind's fields (`check_fields`); a field set to None or
        blank is left out.
    text : str, optional
        Free text.
    event_time : datetime, optional
        When the thing happened, with its zone; now, if not given.
    test, run : int, optional
        The test and run it belongs to, by number from 1.
    axis : str, optional
        The axis it belongs to, by name.
    files : iterable, optional
        Files it names, each a path (hashed now) or a `FileRef`.

    Returns
    -------
    Entry
        The entry as written, with its id and hashes.

    Raises
    ------
    RunLogError
        For a log not open for writing, and for anything the entry
        cannot hold one way, the message naming the field.
    """
    if kind == 'correction':
        raise RunLogError('a correction is written with correct(), which '
                          'names the entry it corrects')
    entry_kind = _kind(kind)
    kept = check_fields(kind, fields)
    if entry_kind.needs_test and test is None:
        raise RunLogError(f'a {entry_kind.label} entry needs the test it '
                          'belongs to')
    self._check_entry_refs(entry_kind, kept)
    if kind in NUMBERED:
        kept = self._numbered(kind, kept)
    if kind == 'anomaly_dispositioned':
        opened = {item['number'] for item in
                  self._numbers('anomaly_opened')}
        if kept['number'] not in opened:
            raise RunLogError(f'{kept["number"]} is not an anomaly opened '
                              'in this log')
    return self._write(kind, author=author, fields=kept,
                       text=text, event_time=event_time, test=test,
                       axis=axis, run=run, files=files, corrects=None)

correct

correct(target: int | Entry, *, author: str, fields: Mapping[str, Any] | None = None, text: str = '', event_time: datetime | None = None) -> Entry

Write a correction of an earlier entry, which stays as it was.

Parameters:

Name	Type	Description	Default
target	int or Entry	The entry corrected, by id. A correction is not corrected itself; correct the original entry again.	required
author	str	The author's typed initials.	required
fields	mapping	The new values: any of the corrected entry's fields (None strikes an optional one), and any of CORRECTABLE (its event time, free text, test, axis and run).	None
text	str	Why it is corrected.	''
event_time	datetime	When the correction was made; now, if not given.	None

Returns:

Type	Description
Entry	The correction as written. It belongs to the corrected entry's test, so it is in that test's chain.

Raises:

Type	Description
RunLogError	When the entry is not in the log or is a correction, when no value is given, when a value is one the entry cannot hold, when a required field would be struck, or when a finding's number would change.

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

def correct(self, target: int | Entry, *, author: str,
            fields: Mapping[str, Any] | None = None, text: str = '',
            event_time: datetime | None = None) -> Entry:
    """Write a correction of an earlier entry, which stays as it was.

    Parameters
    ----------
    target : int or Entry
        The entry corrected, by id. A correction is not corrected
        itself; correct the original entry again.
    author : str
        The author's typed initials.
    fields : mapping
        The new values: any of the corrected entry's fields (None
        strikes an optional one), and any of `CORRECTABLE` (its
        event time, free text, test, axis and run).
    text : str, optional
        Why it is corrected.
    event_time : datetime, optional
        When the correction was made; now, if not given.

    Returns
    -------
    Entry
        The correction as written. It belongs to the corrected
        entry's test, so it is in that test's chain.

    Raises
    ------
    RunLogError
        When the entry is not in the log or is a correction, when no
        value is given, when a value is one the entry cannot hold,
        when a required field would be struck, or when a finding's
        number would change.
    """
    entry_id = target.id if isinstance(target, Entry) else target
    if entry_id not in self._by_id:
        raise RunLogError(f'the log has no entry #{entry_id} to correct')
    original = self._by_id[entry_id]
    if original.kind == 'correction':
        raise RunLogError(f'entry #{entry_id} is a correction; correct '
                          f'the original, #{original.corrects}, again')
    if not fields:
        raise RunLogError('a correction needs the values it corrects')
    if not isinstance(fields, Mapping):
        raise RunLogError(f'a correction\'s values are a mapping, not '
                          f'{type(fields).__name__}')
    entry_kind = KINDS[original.kind]
    if original.kind == 'header_change' and 'field' in fields:
        raise RunLogError('a header change\'s field is never changed; '
                          'write a header change of the other field')
    merged = {**self.current(entry_id).fields, **fields}
    changes: dict[str, Any] = {}
    for name, value in fields.items():
        if name in CORRECTABLE:
            changes[name] = self._corrected_reference(name, value)
        elif name not in entry_kind.fields:
            carried = ', '.join(entry_kind.fields + CORRECTABLE)
            raise RunLogError(f'a {entry_kind.label} entry has no field '
                              f'{name!r}; a correction may change: '
                              f'{carried}')
        elif name == 'number' and original.kind in (
                *NUMBERED, 'anomaly_dispositioned'):
            raise RunLogError('a finding\'s number is never changed; open '
                              'a new finding instead')
        elif value is None:
            changes[name] = None
        else:
            changes[name] = _value(entry_kind, name, value,
                                   _field_type(entry_kind, name, merged))
            self._check_entry_refs(entry_kind, {name: changes[name]})
    corrected = dict(self.current(entry_id).fields)
    for name, value in changes.items():
        if name in CORRECTABLE:
            continue
        if value is None:
            corrected.pop(name, None)
        else:
            corrected[name] = value
    _check_required(entry_kind, corrected)
    if entry_kind.needs_test and 'test' in changes \
            and changes['test'] is None:
        raise RunLogError(f'a {entry_kind.label} entry needs the test it '
                          'belongs to')
    return self._write('correction', author=author,
                       fields=changes, text=text, event_time=event_time,
                       test=original.test, axis=original.axis,
                       run=original.run, files=(), corrects=entry_id)

Functions:

file_ref

file_ref(path: str | PathLike, name: str | None = None) -> FileRef

A FileRef for the file at path, hashed in chunks (data_index.digest); name is the file's own name unless given.

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

def file_ref(path: str | os.PathLike, name: str | None = None) -> FileRef:
    """A `FileRef` for the file at `path`, hashed in chunks
    (`data_index.digest`); `name` is the file's own name unless given."""
    size, sha = data_index.digest(path)
    return FileRef(name or os.path.basename(os.fspath(path)), sha, size)

chain_name

chain_name(test: Any) -> str

The name of the chain an entry with test reference test belongs to: 'test N', or 'log' for none.

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

def chain_name(test: Any) -> str:
    """The name of the chain an entry with test reference `test` belongs
    to: 'test N', or 'log' for none."""
    return 'log' if test is None else f'test {test}'

identity

identity() -> tuple[str, str]

(login, machine name) of this computer, each '' where the system does not say.

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

def identity() -> tuple[str, str]:
    """(login, machine name) of this computer, each '' where the system
    does not say."""
    try:
        login = getpass.getuser()
    except (OSError, KeyError, ImportError):         # pragma: no cover
        # getuser raises when no variable names the user and the
        # password database has no entry; the log still opens
        login = ''
    return login, platform.node()

canonical

canonical(stored: Mapping[str, Any]) -> str

The canonical text of a stored entry (or header) that its hash is taken over: keys sorted, no spaces, UTF-8 characters as themselves, and the entry's own hash left out. The previous hash is a key of the entry, so it is part of what is hashed.

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

def canonical(stored: Mapping[str, Any]) -> str:
    """The canonical text of a stored entry (or header) that its hash is
    taken over: keys sorted, no spaces, UTF-8 characters as themselves,
    and the entry's own `hash` left out. The previous hash is a key of
    the entry, so it is part of what is hashed."""
    body = {key: value for key, value in stored.items() if key != 'hash'}
    return json.dumps(body, sort_keys=True, separators=(',', ':'),
                      ensure_ascii=False, allow_nan=False)

band_hz

band_hz(value: Any, what: str = 'a band') -> tuple[float, float]

A band as (low, high) in Hz, from two frequencies or from text such as '180-220 Hz', '180 to 220', '1.2-2 kHz' or '200 Hz' (one frequency, both edges).

Refused with RunLogError naming what: text that is not one of those forms, and anything findings.band_of refuses (not two finite, non-negative frequencies, low edge above high).

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

def band_hz(value: Any, what: str = 'a band') -> tuple[float, float]:
    """A band as (low, high) in Hz, from two frequencies or from text
    such as '180-220 Hz', '180 to 220', '1.2-2 kHz' or '200 Hz' (one
    frequency, both edges).

    Refused with `RunLogError` naming `what`: text that is not one of
    those forms, and anything `findings.band_of` refuses (not two
    finite, non-negative frequencies, low edge above high).
    """
    if isinstance(value, str):
        found = _BAND.match(value)
        if found is None:
            raise RunLogError(f'{what} {value!r} is not read as a band: give '
                              'it as low-high in Hz, such as 180-220 Hz')
        scale = 1000.0 if (found.group(3) or '').lower() == 'khz' else 1.0
        try:
            low = float(found.group(1)) * scale
            high = float(found.group(2) or found.group(1)) * scale
        except ValueError:
            raise RunLogError(f'{what} {value!r} is not read as a band: give '
                              'it as low-high in Hz, such as 180-220 Hz'
                              ) from None
        value = (low, high)
    try:
        edges = findings.band_of(value, what)
    except (TypeError, ValueError) as error:
        raise RunLogError(str(error)) from None
    if edges is None:
        raise RunLogError(f'{what} is two frequencies in Hz, not None')
    return edges[0], edges[1]

check_fields

check_fields(kind: str, fields: Mapping[str, Any] | None) -> dict[str, Any]

The fields of a new entry of kind, checked against KINDS.

A field set to None or to blank text is left out, so a form can pass every box it has. Refused with RunLogError, each naming the kind and the field: a kind that is not in KINDS, a field the kind does not carry, a value of the wrong type or not finite, a required field missing, an abort's kind of stop not one of STOPS, and a sweep's span that band_hz does not read as a band.

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

def check_fields(kind: str, fields: Mapping[str, Any] | None,
                 ) -> dict[str, Any]:
    """The fields of a new entry of `kind`, checked against `KINDS`.

    A field set to None or to blank text is left out, so a form can
    pass every box it has. Refused with `RunLogError`, each naming the
    kind and the field: a kind that is not in `KINDS`, a field the kind
    does not carry, a value of the wrong type or not finite, a required
    field missing, an abort's kind of stop not one of `STOPS`, and a
    sweep's span that `band_hz` does not read as a band.
    """
    entry_kind = _kind(kind)
    if fields is None:
        fields = {}
    if not isinstance(fields, Mapping):
        raise RunLogError(f'an entry\'s fields are a mapping of names to '
                          f'values, not {type(fields).__name__}')
    kept: dict[str, Any] = {}
    for name, value in fields.items():
        if name not in entry_kind.fields:
            carried = ', '.join(entry_kind.fields) or 'none'
            raise RunLogError(f'a {entry_kind.label} entry has no field '
                              f'{name!r}; its fields are: {carried}')
        if value is None or (isinstance(value, str) and not value.strip()):
            continue
        kept[name] = _value(entry_kind, name, value,
                            _field_type(entry_kind, name, fields))
    _check_required(entry_kind, kept)
    return kept