visualdynamics.core.run_log_export

run_log_export

Views of a run log for other readers: plain text, CSV, and the machine log.

The .vdlog file is its own JSON export, and it is the one format the log is read back from. These views are written out only:

• Plain text (to_text): every entry in the order written, for printing or attaching to a report. A corrected entry is marked with the corrections that changed it, and each correction says what it changed and what the value was; a late entry is marked with when it was entered. The chains' heads and what RunLog.verify found close it.
• CSV (to_csv): one row per entry as stored, corrections included, with the corrected-by and corrects columns, a late column, the time on the recorders' clock, and the fields as JSON.
• The machine log (machine_log_csv): the entries a lab's own machine log keeps (trips and aborts, setting changes, clock checks, people in and out, restarts, and the tests opened and closed, which say what was running), corrected, one row each, in columns a spreadsheet or a machine log takes as they are.

Each takes test: 'all' (the whole log, the default), a test's number, or None for the entries between tests.

Functions:

Name	Description
to_text	The log as plain text, every entry in the order written.
to_csv	The log as CSV: a header row of CSV_COLUMNS, then one row per
machine_log_csv	The trips and machine events of the log, for the lab's own
export	Write one view of the log to path, in UTF-8, and return the

Classes

Functions:

to_text

to_text(log: RunLog, *, test: int | str | None = ALL, late_after: timedelta = LATE_AFTER) -> str

The log as plain text, every entry in the order written.

Parameters:

Name	Type	Description	Default
log	RunLog	The log.	required
test	(int, None or 'all')	Which entries: the whole log, one test, or (None) those between tests.	ALL
late_after	timedelta	How long after its event an entry is entered to be marked late.	LATE_AFTER

Returns:

Type	Description
str	The text, a line or more per entry, then the lines that could not be read, the chains' heads, and what verifying found.

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

def to_text(log: RunLog, *, test: int | str | None = ALL,
            late_after: timedelta = LATE_AFTER) -> str:
    """The log as plain text, every entry in the order written.

    Parameters
    ----------
    log : RunLog
        The log.
    test : int, None or 'all', optional
        Which entries: the whole log, one test, or (None) those between
        tests.
    late_after : timedelta, optional
        How long after its event an entry is entered to be marked late.

    Returns
    -------
    str
        The text, a line or more per entry, then the lines that could
        not be read, the chains' heads, and what verifying found.
    """
    entries = _selected(list(log.entries), test)
    header = log.header
    lines = [f'Run log: {header.title or "(untitled)"}',
             (f'Log {header.log_id}, run log format {header.version}, created '
             f'{_clock(header.created)}'),
             f'{len(entries)} entr{"y" if len(entries) == 1 else "ies"}'
             + ('' if test == ALL else
                f', {"between tests" if test is None else f"test {test}"}'),
             '']
    for entry in entries:
        recorder = log.recorder_time(entry)
        lines.append(f'#{entry.id}  {_clock(entry.event_time)}  {entry.author}'
                     f' ({entry.login or "?"} on {entry.computer or "?"})  '
                     f'{_where(entry)}')
        if entry.kind == 'correction':
            body = _correction_text(log, entry)
        else:
            fields = _fields_text(entry, entry.fields)
            body = entry.label + (f': {fields}' if fields else '')
        lines.append(f'    {body}')
        if entry.text:
            lines.append(f'    "{entry.text}"')
        for ref in entry.files:
            size = '' if ref.size is None else f', {ref.size} bytes'
            lines.append(f'    file: {ref.name} (SHA-256 {ref.sha256}{size})')
        marks = []
        if recorder is not None:
            marks.append(f'recorders\' clock {_clock(recorder)}')
        if entry.is_late(late_after):
            marks.append(f'LATE: entered {_clock(entry.entered_at)}')
        corrections = log.corrections_of(entry.id)
        if corrections:
            marks.append('CORRECTED by '
                         + ', '.join(f'#{c.id}' for c in corrections))
        if marks:
            lines.append('    [' + '] ['.join(marks) + ']')
    if log.problems:
        lines += ['', 'Lines not read:']
        lines += [f'  {problem}' for problem in log.problems]
    chains = log.chains()
    if test != ALL:
        chains = [name for name in chains
                  if name == ('log' if test is None else f'test {test}')]
    if chains:
        lines += ['', 'Chain heads (SHA-256):']
        for name in chains:
            number = None if name == 'log' else int(name.split()[1])
            lines.append(f'  {name}: {log.head(number)}')
    lines += ['', f'Integrity: {log.verify()}']
    return '\n'.join(lines) + '\n'

to_csv

to_csv(log: RunLog, *, test: int | str | None = ALL, late_after: timedelta = LATE_AFTER) -> str

The log as CSV: a header row of CSV_COLUMNS, then one row per entry as stored, corrections included.

Parameters:

Name	Type	Description	Default
log	RunLog	The log.	required
test	(int, None or 'all')	Which entries: the whole log, one test, or (None) those between tests.	ALL
late_after	timedelta	How long after its event an entry is entered to be marked late.	LATE_AFTER

Returns:

Type	Description
str	The CSV text. Times are ISO 8601 with their UTC offsets; late is 'late' or blank; fields is the entry's fields as JSON; files lists each file as its name and SHA-256; a corrected entry lists its corrections in corrected_by, and a correction names its entry in corrects.

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

def to_csv(log: RunLog, *, test: int | str | None = ALL,
           late_after: timedelta = LATE_AFTER) -> str:
    """The log as CSV: a header row of `CSV_COLUMNS`, then one row per
    entry as stored, corrections included.

    Parameters
    ----------
    log : RunLog
        The log.
    test : int, None or 'all', optional
        Which entries: the whole log, one test, or (None) those between
        tests.
    late_after : timedelta, optional
        How long after its event an entry is entered to be marked late.

    Returns
    -------
    str
        The CSV text. Times are ISO 8601 with their UTC offsets; `late`
        is 'late' or blank; `fields` is the entry's fields as JSON;
        `files` lists each file as its name and SHA-256; a corrected
        entry lists its corrections in `corrected_by`, and a correction
        names its entry in `corrects`.
    """
    rows = []
    for entry in _selected(list(log.entries), test):
        recorder = log.recorder_time(entry)
        rows.append([
            entry.id, entry.kind, entry.label,
            entry.event_time.isoformat(timespec='milliseconds'),
            '' if recorder is None else recorder.isoformat(
                timespec='milliseconds'),
            entry.entered_at.isoformat(timespec='milliseconds'),
            'late' if entry.is_late(late_after) else '',
            entry.author, entry.login, entry.computer,
            _optional(entry.test), _optional(entry.axis),
            _optional(entry.run),
            json.dumps(dict(entry.fields), ensure_ascii=False,
                       sort_keys=True),
            entry.text,
            '; '.join(f'{ref.name} {ref.sha256}' for ref in entry.files),
            _optional(entry.corrects),
            ' '.join(str(c.id) for c in log.corrections_of(entry.id)),
            entry.hash])
    return _csv(rows, CSV_COLUMNS)

machine_log_csv

machine_log_csv(log: RunLog, *, test: int | str | None = ALL) -> str

The trips and machine events of the log, for the lab's own machine log: a header row of MACHINE_COLUMNS, then one row per entry of a kind KINDS marks machine, in the corrected view.

Parameters:

Name	Type	Description	Default
log	RunLog	The log.	required
test	(int, None or 'all')	Which entries: the whole log, one test, or (None) those between tests.	ALL

Returns:

Type	Description
str	The CSV text: the event's local date, time and UTC offset, its time on the recorders' clock where a clock check is in force, the event in words ('Abort or trip'), its details, the test that was running and its title, the axis and run, the author, the entry's id, and 'corrected' when a correction changed it.

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

def machine_log_csv(log: RunLog, *, test: int | str | None = ALL) -> str:
    """The trips and machine events of the log, for the lab's own
    machine log: a header row of `MACHINE_COLUMNS`, then one row per
    entry of a kind `KINDS` marks `machine`, in the corrected view.

    Parameters
    ----------
    log : RunLog
        The log.
    test : int, None or 'all', optional
        Which entries: the whole log, one test, or (None) those between
        tests.

    Returns
    -------
    str
        The CSV text: the event's local date, time and UTC offset, its
        time on the recorders' clock where a clock check is in force,
        the event in words ('Abort or trip'), its details, the test
        that was running and its title, the axis and run, the author,
        the entry's id, and 'corrected' when a correction changed it.
    """
    view = log.view()
    titles: dict[int, str] = {}
    for entry in view:
        if entry.kind == 'test_opened' and entry.test is not None:
            titles[entry.test] = str(entry.fields.get('title', ''))
    rows = []
    for entry in _selected(view, test):
        kind = KINDS.get(entry.kind)
        if kind is None or not kind.machine:
            continue
        recorder = log.recorder_time(entry)
        offset = entry.event_time.strftime('%z')
        detail = _fields_text(entry, entry.fields)
        if entry.text:
            detail = f'{detail}; {entry.text}' if detail else entry.text
        rows.append([
            entry.event_time.strftime('%Y-%m-%d'),
            entry.event_time.strftime('%H:%M:%S'),
            f'{offset[:3]}:{offset[3:]}',
            '' if recorder is None else recorder.isoformat(
                timespec='milliseconds'),
            kind.label, detail, _optional(entry.test),
            titles.get(entry.test, '') if entry.test is not None else '',
            _optional(entry.axis), _optional(entry.run), entry.author,
            entry.id, 'corrected' if entry.corrected_by else ''])
    return _csv(rows, MACHINE_COLUMNS)

export

export(log: RunLog, path: str | PathLike, view: str = 'text', *, test: int | str | None = ALL) -> str

Write one view of the log to path, in UTF-8, and return the path written.

Parameters:

Name	Type	Description	Default
log	RunLog	The log.	required
path	str or PathLike	Where to write; given the view's extension (VIEWS) when it has none.	required
view	str	'text' (to_text), 'csv' (to_csv) or 'machine' (machine_log_csv).	'text'
test	(int, None or 'all')	Which entries.	ALL

Returns:

Type	Description
str	The path written.

Raises:

Type	Description
RunLogError	For a view that is not one of VIEWS, or a path that is the log itself.

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

def export(log: RunLog, path: str | os.PathLike, view: str = 'text', *,
           test: int | str | None = ALL) -> str:
    """Write one view of the log to `path`, in UTF-8, and return the
    path written.

    Parameters
    ----------
    log : RunLog
        The log.
    path : str or os.PathLike
        Where to write; given the view's extension (`VIEWS`) when it has
        none.
    view : str, optional
        'text' (`to_text`), 'csv' (`to_csv`) or 'machine'
        (`machine_log_csv`).
    test : int, None or 'all', optional
        Which entries.

    Returns
    -------
    str
        The path written.

    Raises
    ------
    RunLogError
        For a view that is not one of `VIEWS`, or a path that is the
        log itself.
    """
    if view not in VIEWS:
        raise RunLogError(f'{view!r} is not a view of a run log; the views '
                          'are: ' + ', '.join(VIEWS))
    path = os.fspath(path)
    if not os.path.splitext(path)[1]:
        path += VIEWS[view]
    if os.path.normcase(os.path.abspath(path)) == os.path.normcase(
            os.path.abspath(log.path)):
        raise RunLogError('an export is written beside the log, never over it')
    if view == 'text':
        text = to_text(log, test=test)
    elif view == 'csv':
        text = to_csv(log, test=test)
    else:
        text = machine_log_csv(log, test=test)
    with open(path, 'w', encoding='utf-8', newline='') as handle:
        handle.write(text)
    return path