Writing results safely¶

Two small modules that everything else depends on being correct. Neither is part of the user interface; both exist because of failure modes that are invisible until they have already cost someone their results.

Atomic writes¶

Analysis steps stream output row by row, and steps skip work whose output already exists — that is what makes a long pipeline resumable. Together those two facts make an interrupted run dangerous: a half-written file keeps its final name, and the next run accepts it as finished.

Writing under a scratch name and renaming at the end removes that. A rename is indivisible as far as the filesystem is concerned, so the real name only ever refers to a complete file.

taters.helpers.atomic ¶

Write a file under a scratch name, and give it its real name only when finished.

Analysis steps stream their output row by row, which means the output file exists -- with its final name -- from the first row onwards. Anything that interrupts the run leaves that half-written file behind looking exactly like a completed one.

That would be survivable on its own. What makes it a real hazard is the other half of the design: steps skip work whose output already exists, because that is what makes a long pipeline resumable. So a truncated file is not merely wrong, it is sticky -- the next run sees it, decides the step is done, and returns it. You get 3 rows where you asked for 300,000, with no error.

The fix is to write to <name>.part and rename it at the end. A rename is indivisible as far as the filesystem is concerned: there is no moment at which the destination exists half-renamed. Every other part of the write can be interrupted; that step cannot. So the real name only ever refers to a complete file, and an interrupted run leaves nothing for the next one to mistake for finished work.

This covers interruption generally -- a cancelled run, a full disk, a power cut, Ctrl-C -- not just any one of them.

atomic_write ¶

atomic_write(path, mode='w', **open_kwargs)

Open a file for writing that only appears at path once complete.

Parameters:

Name	Type	Description	Default
`path`	`str or Path`	Where the finished file should end up. Parent directories are created.	required
`mode`	`str`	As :func:`open`. Must be a writing mode.	`"w"`
`**open_kwargs`	`Any`	Passed straight through to :func:`open` -- `newline`, `encoding` and so on.	`{}`

Yields:

Type	Description
`IO`	The handle to write to. It refers to the scratch file, not to `path`.

Notes

On an exception the scratch file is removed and path is left exactly as it was -- which for a first run means absent, so the step is retried rather than resumed from a partial file.

Not safe for two processes writing the same destination at once: they would share a scratch name. Nothing in Taters does that, since a GLOBAL step runs once and ITEM steps write per-input paths.

Source code in src\taters\helpers\atomic.py

@contextmanager
def atomic_write(
    path: Union[str, Path],
    mode: str = "w",
    **open_kwargs: Any,
) -> Iterator[IO]:
    """
    Open a file for writing that only appears at ``path`` once complete.

    Parameters
    ----------
    path : str or pathlib.Path
        Where the finished file should end up. Parent directories are created.
    mode : str, default "w"
        As :func:`open`. Must be a writing mode.
    **open_kwargs
        Passed straight through to :func:`open` -- ``newline``, ``encoding``
        and so on.

    Yields
    ------
    IO
        The handle to write to. It refers to the scratch file, not to ``path``.

    Notes
    -----
    On an exception the scratch file is removed and ``path`` is left exactly as
    it was -- which for a first run means absent, so the step is retried rather
    than resumed from a partial file.

    Not safe for two processes writing the same destination at once: they would
    share a scratch name. Nothing in Taters does that, since a GLOBAL step runs
    once and ITEM steps write per-input paths.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    scratch = path.with_name(path.name + SCRATCH_SUFFIX)

    handle = scratch.open(mode, **open_kwargs)
    try:
        yield handle
    except BaseException:
        # BaseException, not Exception: KeyboardInterrupt is the most likely
        # way we land here, and that's exactly the case this exists for.
        handle.close()
        scratch.unlink(missing_ok=True)
        raise
    else:
        handle.close()
        os.replace(scratch, path)

Progress reporting¶

A GLOBAL pipeline step is a single call, so the runner cannot count it from outside. A step that can count itself says so by declaring an on_progress parameter, which the runner then injects automatically — a signature check rather than a registry, so a new analyzer opts in without anything else needing to know.

taters.helpers.progress ¶

Row-level progress reporting for analysis steps.

A GLOBAL pipeline step is a single call: the runner hands over and gets control back at the end, so it cannot count anything from outside. A step that can count itself says so by declaring an on_progress parameter, which the pipeline runner then injects automatically.

The contract

on_progress(done, total, message=None, unit=None)

total is an int once the size of the job is known, and done is a position within it.
total is None while the size is still being worked out. done is then a running tally, and message says which pass is running -- reading the input, counting its rows. On a large file those passes take long enough to look like a hang, and a number climbing is the only visible proof that anything is happening.
unit names what is being counted when it is not rows. The only value with a meaning today is "seconds", used by transcription, where the numbers are a position in the recording: 252/771 would be read as segments, which is not what it is, while 4:12/12:51 cannot be misread.

The last two arguments are optional in both directions -- a sink that predates them still works, because nothing is obliged to send them.

This module exists so the five text analyzers share one implementation of that pattern rather than five copies that drift.

Ticker ¶

Ticker(on_progress, total=0)

Counts work as it happens and reports it.

Safe to use unconditionally: with no on_progress every method is a no-op, so the analysis code reads the same whether anyone is watching.

Source code in src\taters\helpers\progress.py

def __init__(self, on_progress: Optional[Callable[..., None]], total: int = 0) -> None:
    self._on_progress = on_progress
    self._total = int(total)
    self._done = 0
    if on_progress is not None:
        on_progress(0, self._total or None, None)

tick ¶

tick(n=1, message=None)

Record n more units of work done.

message names the unit -- "scoring " -- for steps whose rows are wildly uneven: one two-million-character document can take minutes where its neighbors take milliseconds, and a bar that just sits there unnamed reads as a hang rather than as one slow paper.

Source code in src\taters\helpers\progress.py

def tick(self, n: int = 1, message: Optional[str] = None) -> None:
    """
    Record ``n`` more units of work done.

    ``message`` names the unit -- "scoring <text_id>" -- for steps whose
    rows are wildly uneven: one two-million-character document can take
    minutes where its neighbors take milliseconds, and a bar that just
    sits there unnamed reads as a hang rather than as one slow paper.
    """
    if self._on_progress is None:
        return
    self._done += n
    self._on_progress(self._done, self._total or None, message)

FlightReporter ¶

FlightReporter(on_progress, total, message)

Progress for a phase whose work is spread over parallel workers.

Reports the usual (done, total, message) plus -- to sinks that accept it -- inflight: the names currently being worked on, so a display can show one sub-bar per file the way item steps do for ffmpeg or Whisper. start/finish may be called from executor threads; consumed is the parent's in-order tally. All three re-report immediately.

Source code in src\taters\helpers\progress.py

def __init__(self, on_progress: Optional[Callable[..., None]],
             total: int, message: str) -> None:
    import threading

    self._on_progress = on_progress
    self._total = int(total)
    self._message = message
    self._done = 0
    self._inflight: list = []
    self._lock = threading.Lock()
    self._takes_inflight = (on_progress is not None
                            and _sink_takes(on_progress, "inflight"))
    self._report()

announce ¶

announce(on_progress, message)

Name the phase that is about to run, with no size known yet.

Source code in src\taters\helpers\progress.py

def announce(on_progress: Optional[Callable[..., None]], message: str) -> None:
    """Name the phase that is about to run, with no size known yet."""
    if on_progress is not None:
        on_progress(0, None, message)

count_rows ¶

count_rows(
    path,
    *,
    on_progress=None,
    encoding="utf-8-sig",
    every=1000
)

Count the data records in a CSV, reporting progress as it goes.

Records, not lines: document text carries embedded newlines inside its quoted field, and counting physical lines told a 2,300-paper run it had 3.2 million rows to do -- a denominator so wrong the bar read as broken.

Always counts, and only reports when something is watching. It used to return 0 with no sink, on the theory that the count was only ever a progress bar's denominator -- but :func:taters.helpers.row_map.map_text_rows sizes its worker pool from it, so every direct API call (no sink) ran on one worker while the same call from the wizard ran on twelve. One csv.reader sweep is far cheaper than the work it sizes.

The header is not counted. A file that cannot be read comes back as 0 rather than raising -- a progress bar is not worth failing a run over.

Source code in src\taters\helpers\progress.py

def count_rows(
    path: Path,
    *,
    on_progress: Optional[Callable[..., None]] = None,
    encoding: str = "utf-8-sig",
    every: int = 1_000,
) -> int:
    """
    Count the data *records* in a CSV, reporting progress as it goes.

    Records, not lines: document text carries embedded newlines inside its
    quoted field, and counting physical lines told a 2,300-paper run it had
    3.2 million rows to do -- a denominator so wrong the bar read as broken.

    Always counts, and only *reports* when something is watching. It used
    to return 0 with no sink, on the theory that the count was only ever a
    progress bar's denominator -- but :func:`taters.helpers.row_map.map_text_rows`
    sizes its worker pool from it, so every direct API call (no sink) ran
    on one worker while the same call from the wizard ran on twelve. One
    ``csv.reader`` sweep is far cheaper than the work it sizes.

    The header is not counted. A file that cannot be read comes back as 0
    rather than raising -- a progress bar is not worth failing a run over.
    """
    import csv

    widen_csv_field_limit()

    seen = 0
    try:
        with Path(path).open("r", encoding=encoding, newline="") as fh:
            for seen, _ in enumerate(csv.reader(fh), 1):
                if on_progress is not None and seen % every == 0:
                    on_progress(seen, None, "counting rows")
    except (OSError, UnicodeDecodeError, csv.Error):
        return 0

    return max(0, seen - 1)