Heidelberg AI Curriculum

1. The number nobody can rebuild

Your colleague does not need another screenshot. They need the next version of the result, on a morning when you are not there.

You open a report before a review meeting and find a chart labelled "Mean value by group." Someone asks which rows produced the taller bar. The analyst who made it is on leave. Their shared folder contains final.csv, final-fixed.csv, a screenshot, and a script that refers to C:\Users\name\Desktop\August. Nobody knows whether the screenshot came before or after the manual fix.

The number may be right, but the team cannot demonstrate that. Re-running the script fails on another computer. Replacing its path makes it run, yet the result differs because one parameter lived in an old notebook cell. In a lab, this blocks review of a result. In a company, it delays a monthly report and forces someone to reverse-engineer undocumented choices.

Continue the project you chose in Clean a messy export, or choose another permitted analysis with a real recipient. Perhaps a colleague needs a weekly instrument summary, a recurring capacity comparison, or a table for a monthly report. Those are possibilities, not assignments. Name the result they will actually use, when they need it again, and what they cannot do if it is missing or wrong.

Your job is to build the short, inspectable route from raw evidence to that result. This book's concrete workbench is a plain-text editor, a local terminal, Python's standard library, and Git. The optional six-row examples make the machinery easy to inspect; they are not a reason to abandon a useful project or convert every analysis into a CSV bar chart.

The boundary matters. T10-L02 made cleanup repeatable. Here you package the analysis, its checks, and its instructions so someone else can depend on it. T10-L04 adds connections to changing systems and live-data quality gates. You do not need a database, scheduler, container, or deployment to make this handoff valuable.

2. After this you can

Separate raw data, analysis steps, and generated output so edits cannot be mistaken for computation.
Centralise paths, labels, and acceptance values in one explicit parameter file.
Build an analysis that reproduces its declared result from recorded inputs.
Test a clean-room run, raw-file preservation, deliberate failure, and the agreed rerun comparison.
Hand one versioned pipeline to a colleague who can regenerate, use, and recover the result without your memory.

3. Prerequisites

T10-L02 - Clean a messy export, including preservation of raw input and explicit transformations.
T08-L02 - Code without being a developer, including bounded code review, known-input checks, and deliberate failures.
For the runnable example: Python 3 with the standard library, Git, and a plain-text editor or approved coding assistant. Record the exact Python version used.
Permission to create a local Git repository and run a script in a dedicated project folder.
A colleague who can perform the final handoff from written instructions alone. Until that happens, your handoff is prepared, not independently demonstrated.

Use only public, synthetic, course-provided, or explicitly approved data. Both fixtures below are synthetic. Do not put patient, participant, employee, customer, credential, unpublished, or production data into an exercise repository. For an approved project, permission to analyse a file is not automatically permission to put it in Git or send it to an assistant. Record the approved storage location and recipient access; keep restricted inputs outside shared history and document their authorised retrieval instead.

Git retains history: deleting a sensitive file in a later commit does not remove it from earlier commits. If unsuitable data enters history, stop sharing the repository and follow your organisation's incident and history-remediation process.

4. The idea in one page

Start with the recipient, not the file format

Before opening the editor, finish this sentence in your project notes: "When [approved input] arrives, [recipient] needs [result] to [do something], and will reject it if [specific condition]." Replace the brackets with your own task. A research collaborator may need counts alongside a mean to spot an incomplete batch. A planning colleague may need totals and units, not a polished chart. Let that use determine what you build.

Agree on the smallest result that removes a recurring manual step. Do not add a dashboard if a reopened table is enough. Equally, do not call a table sufficient if the recipient needs a labelled figure and cannot reconstruct it. The deliverable is the result plus the means to regenerate it, not simply code that terminates successfully.

Give every kind of file one job

Use three top-level folders and do not blur their roles:

Folder	Contains	Rule
`raw/`	The received input used for this run	Never edit it in place. Introduce a new input through a documented new version.
`steps/`	Code or an exported visual workflow, plus automated checks	Version it. Every transformation must be visible here.
`output/`	Tables, figures, and other derived files	Regenerate it. Never correct it by hand.

Keep parameters.json and README.md at the project root for the Python route. The parameter file is the one place for input and output paths, column choices, labels, group order, expected row count, and the hand-calculated acceptance result. The README states the environment, exact commands, generated files, expected evidence, and limits. Git identifies the complete version a colleague tested.

Think of the flow as raw input + parameters -> visible steps -> checks -> generated result -> colleague's use. A correction belongs before the generated result, in a named transformation or an approved new source version. A human editing the last box breaks the chain.

This separation creates a useful test: remove only the disposable output/ folder, follow the README, and see whether the pipeline reconstructs it from raw/, steps/, and parameters.json. If removal destroys information needed for the analysis, hidden work was living in the output folder. Keep handoff evidence and irreplaceable notes outside that folder.

Define "the same result"

Do not settle for "the bars look similar." For this bounded Python example, the same result means:

The CSV has the exact group order, counts, totals, and means declared before the run.
The SVG is generated from that same summary, not edited separately.
The raw input hash is unchanged by execution.
A second run with the same commit and parameters produces byte-identical CSV and SVG files.
A malformed input stops before any plausible replacement output is written.

Byte identity is practical here because the outputs contain no timestamps, random values, machine paths, locale-dependent formatting, or unstable row order. Other analyses may legitimately produce equivalent rather than byte-identical files. Then the README must define the comparison in advance, such as a numeric tolerance, sorted row set, or image-data checksum. Never weaken the comparison only after observing a mismatch.

A hash answers "Are these bytes the same?", not "Is this analysis sound?" Two identical files can repeat the same mistake. The independently calculated fixture result checks arithmetic; the recipient and data owner still need to approve units, grouping, exclusions, and the statistic's use. A bar chart of means does not establish causation, uncertainty, or comparability between groups.

Remove hidden state

Hidden state is anything the pipeline needs but the project does not declare: an absolute path, an unexecuted notebook cell, a package installed months ago, a manually selected spreadsheet range, an environment variable, a locale-specific decimal rule, or an output left from the previous run. A successful run on the author's machine does not reveal these dependencies.

The example uses only the Python standard library, resolves paths relative to parameters.json, orders groups and decimals explicitly, recreates outputs, and tests in a temporary directory without old output. These behaviours make a clean handoff possible.

If you are continuing a saved KNIME or notebook workflow, keep it when it already serves the task. Export or version the actual workflow, expose its input and output locations, remove dependence on cached execution state, and write the exact reset-and-run instructions for that interface. You must still demonstrate the same file roles, acceptance checks, and independent handoff. The Python listing below is a complete worked route, not a compulsory migration from your existing tool.

Separate a fixture from the next delivery

The six-row fixture has a result you can calculate by hand. Keep it as a stable regression example: a later code change must not silently change its answer. Your next approved export is a different input, and its real result will often be different. Reproducibility means rebuilding the answer for that input, not forcing every month to equal the first month.

The example deliberately requires an exact expected_summary. It is therefore a checked snapshot analysis, not a general unattended monthly service. For a small new snapshot, review and independently calculate its new expected values before running it. Do not paste a failing script's output into the acceptance file merely to make it pass. For a larger continuing project, retain a small known-answer fixture and define separate justified checks for delivery data, such as reconciled totals, allowed units, or approved ranges. That adaptation needs its own tests; changing six rows to sixty thousand does not make the current fixture contract sufficient.

Treat reproducibility as an independent test

The builder runs the checks first, but the exit check belongs to a colleague. Give them a commit identifier and the README, then stay silent unless the instructions themselves tell them to ask for a documented prerequisite. They should start from a fresh clone or exported copy, run the checks, remove disposable generated output, run the analysis, compare the declared result, and record what happened.

Then ask them to use it: open the table in their actual receiving application, or place the generated figure in a draft of the intended report. Can they find the input version, units, and counts needed to interpret it? A matching hash does not rescue an unreadable label or a column whose meaning they have to guess.

If they need an oral instruction, the handoff is not yet complete. Add the missing instruction to the README, commit it, and ask them to restart from the new commit. Do not quietly coach them through the old version and call it reproducible.

5. The worked example: rebuild one result from a clean checkout

The optional Lab and Company framings use the same project shape, script, tests, commands, and evidence. Only the synthetic rows, labels, and declared acceptance result change. Choose one if you need a small inspectable example; you do not need to complete both. For your own project, use the same handoff logic without adopting irrelevant groups, statistics, or formats.

Assemble the shared project

In your file manager or editor, create a new dedicated folder named reproducible-analysis. Open that folder in the editor. Create the files below as plain text, not as a notebook or a word-processing document. Save the Python, JSON, and CSV contents without the surrounding Markdown fences; ensure your editor has not appended .txt to their names.

reproducible-analysis/
|-- .gitignore
|-- README.md
|-- parameters.json
|-- raw/
|   `-- input.csv
|-- steps/
|   |-- analyze.py
|   `-- test_pipeline.py
`-- output/                 # generated; may be absent before a run

Open a terminal in this project root, using your editor's terminal or your operating system's terminal. Confirm its displayed location ends in reproducible-analysis before any run or deletion command. The script resolves its data paths itself, but the relative command paths below assume this terminal location.

Create .gitignore before running Python so interpreter artifacts cannot enter the handoff:

__pycache__/
*.py[cod]
.venv/

Do not ignore output/ in this small example: its visible untracked or changed state helps the runner see generated files, though status alone does not prove a successful run. Do not store anything irreplaceable there. The writer replaces the whole directory, not just two selected files. Use one runner at a time, ordinary project directories rather than linked folders, and no other application writing into output/ during execution.

The code below rejects absolute paths, paths that escape the project, an input outside raw/, and outputs outside output/. Save it as the new file steps/analyze.py:

import argparse
import csv
import io
import json
import os
import shutil
import tempfile
from decimal import Decimal, InvalidOperation
from html import escape
from pathlib import Path

def project_path(root, value, required_folder):
    if not isinstance(value, str) or not value:
        raise ValueError("Paths must be non-empty strings")
    if Path(value).is_absolute():
        raise ValueError(f"Path must be relative: {value}")
    project_root = root.resolve()
    candidate = (root / value).resolve()
    boundary = (root / required_folder).resolve()
    try:
        boundary.relative_to(project_root)
        candidate.relative_to(boundary)
    except ValueError as error:
        raise ValueError(f"Path must be inside {required_folder}/: {value}") from error
    return candidate

def read_parameters(path):
    with path.open(encoding="utf-8") as source:
        params = json.load(source)
    if not isinstance(params, dict):
        raise ValueError("parameters.json must contain one JSON object")
    required = {
        "input", "output_csv", "output_svg", "id_column", "group_column",
        "value_column", "group_order", "group_labels", "expected_row_count",
        "expected_summary",
    }
    missing = required - set(params)
    if missing:
        raise ValueError(f"Missing parameters: {', '.join(sorted(missing))}")
    group_order = params["group_order"]
    labels = params["group_labels"]
    if (
        not isinstance(group_order, list)
        or not group_order
        or any(not isinstance(group, str) or not group for group in group_order)
        or len(group_order) != len(set(group_order))
    ):
        raise ValueError("group_order must contain unique non-empty strings")
    if not isinstance(labels, dict) or set(labels) != set(group_order):
        raise ValueError("group_labels must contain exactly one label per group")
    if any(not isinstance(label, str) or not label for label in labels.values()):
        raise ValueError("group labels must be non-empty strings")
    if not isinstance(params["expected_row_count"], int) or params["expected_row_count"] < 1:
        raise ValueError("expected_row_count must be a positive integer")
    if not isinstance(params["expected_summary"], list):
        raise ValueError("expected_summary must be a list")
    return params

def read_rows(input_path, params):
    with input_path.open(newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source, strict=True)
        required = {
            params["id_column"], params["group_column"], params["value_column"]
        }
        missing = required - set(reader.fieldnames or [])
        if missing:
            raise ValueError(f"Missing columns: {', '.join(sorted(missing))}")

        rows = []
        seen_ids = set()
        allowed_groups = set(params["group_order"])
        for line_number, row in enumerate(reader, start=2):
            if None in row:
                raise ValueError(
                    f"Line {line_number}: malformed CSV or extra fields"
                )
            record_id = (row[params["id_column"]] or "").strip()
            group = " ".join((row[params["group_column"]] or "").split()).casefold()
            value_text = (row[params["value_column"]] or "").strip()
            if not record_id or not group or not value_text:
                raise ValueError(f"Line {line_number}: blank required value")
            if record_id in seen_ids:
                raise ValueError(f"Line {line_number}: duplicate ID {record_id}")
            if group not in allowed_groups:
                raise ValueError(f"Line {line_number}: unexpected group {group}")
            try:
                value = Decimal(value_text)
            except InvalidOperation as error:
                raise ValueError(f"Line {line_number}: invalid number {value_text}") from error
            if not value.is_finite():
                raise ValueError(f"Line {line_number}: number must be finite")
            seen_ids.add(record_id)
            rows.append((group, value))

    if len(rows) != params["expected_row_count"]:
        raise ValueError(
            f"Expected {params['expected_row_count']} rows, received {len(rows)}"
        )
    return rows

def summarize(rows, group_order):
    counts = {group: 0 for group in group_order}
    totals = {group: Decimal("0") for group in group_order}
    for group, value in rows:
        counts[group] += 1
        totals[group] += value

    summary = []
    for group in group_order:
        if counts[group] == 0:
            raise ValueError(f"Group has no rows: {group}")
        total = totals[group]
        mean = total / counts[group]
        summary.append({
            "group": group,
            "count": counts[group],
            "total": f"{total:.2f}",
            "mean": f"{mean:.2f}",
        })
    return summary

def verify_expected(summary, expected):
    if summary != expected:
        raise ValueError(f"Summary differs from declared acceptance result: {summary}")

def render_csv(summary):
    target = io.StringIO(newline="")
    writer = csv.DictWriter(
        target,
        fieldnames=["group", "count", "total", "mean"],
        lineterminator="\n",
    )
    writer.writeheader()
    writer.writerows(summary)
    return target.getvalue()

def render_svg(summary, labels):
    means = [Decimal(row["mean"]) for row in summary]
    if any(mean < 0 for mean in means) or max(means) == 0:
        raise ValueError("SVG requires non-negative means and at least one positive mean")
    maximum = max(means)
    bars = []
    for index, row in enumerate(summary):
        y = 28 + index * 48
        width = int(Decimal(row["mean"]) / maximum * 360)
        label = escape(labels[row["group"]])
        text = escape(f"{label}: mean {row['mean']} (n={row['count']})")
        bars.append(f'<text x="10" y="{y + 16}">{text}</text>')
        bars.append(
            f'<rect x="220" y="{y}" width="{width}" height="24" fill="#315f72"/>'
        )
    height = 52 + len(summary) * 48
    svg = (
        f'<svg xmlns="http://www.w3.org/2000/svg" width="620" height="{height}" '
        'role="img" aria-labelledby="figure-title figure-description">\n'
        '<title id="figure-title">Mean value by group</title>\n'
        '<desc id="figure-description">Horizontal bars compare the mean value '
        'for each group; labels also give each mean and sample count.</desc>\n'
        '<style>text { font: 14px sans-serif; }</style>\n'
        + "\n".join(bars)
        + "\n</svg>\n"
    )
    return svg

def write_outputs(csv_text, svg_text, csv_path, svg_path, output_root):
    staging_root = Path(tempfile.mkdtemp(dir=output_root.parent, prefix=".output-stage-"))
    backup_root = Path(tempfile.mkdtemp(dir=output_root.parent, prefix=".output-backup-"))
    backup_root.rmdir()
    had_output = output_root.exists()
    try:
        staged_csv = staging_root / csv_path.relative_to(output_root)
        staged_svg = staging_root / svg_path.relative_to(output_root)
        for path, text in ((staged_csv, csv_text), (staged_svg, svg_text)):
            path.parent.mkdir(parents=True, exist_ok=True)
            with path.open("w", newline="\n", encoding="utf-8") as target:
                target.write(text)

        if had_output:
            os.replace(output_root, backup_root)
        try:
            os.replace(staging_root, output_root)
        except OSError:
            if had_output and backup_root.exists():
                os.replace(backup_root, output_root)
            raise
        if had_output:
            shutil.rmtree(backup_root)
    finally:
        if staging_root.exists():
            shutil.rmtree(staging_root)

def run(parameter_path):
    parameter_path = parameter_path.resolve()
    root = parameter_path.parent
    params = read_parameters(parameter_path)
    input_path = project_path(root, params["input"], "raw")
    csv_path = project_path(root, params["output_csv"], "output")
    svg_path = project_path(root, params["output_svg"], "output")
    output_root = (root / "output").resolve()
    if csv_path == svg_path:
        raise ValueError("CSV and SVG output paths must differ")
    rows = read_rows(input_path, params)
    summary = summarize(rows, params["group_order"])
    verify_expected(summary, params["expected_summary"])
    csv_text = render_csv(summary)
    svg_text = render_svg(summary, params["group_labels"])
    write_outputs(csv_text, svg_text, csv_path, svg_path, output_root)
    print(f"PASS: {len(rows)} rows, {len(summary)} groups, outputs regenerated")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--parameters", type=Path, required=True)
    arguments = parser.parse_args()
    try:
        run(arguments.parameters)
    except (
        ArithmeticError, KeyError, OSError, TypeError, ValueError,
        csv.Error, json.JSONDecodeError,
    ) as error:
        raise SystemExit(f"FAIL: {error}") from error

Read the route before trusting the result

Start at run, near the bottom, rather than reading the listing as an alphabet of unfamiliar functions. It names the route through the analysis. read_parameters loads the contract; project_path checks where files may live; read_rows validates and normalises the input; summarize calculates; verify_expected compares against the declared answer; only then do rendering and writing begin.

Notice what the normalisation does and does not do. Trimming an ID does not change its case. Group names have whitespace collapsed and case folded, so Control and control join the same group. That is justified for this fixture because they are spellings of one declared category. It would be wrong if case distinguished real categories in your project. The code rejects an unexpected group rather than guessing its meaning, and rejects duplicate IDs rather than silently choosing a row.

Decimal keeps these decimal inputs out of binary floating-point arithmetic, and the summary explicitly formats two decimal places. That is a declared presentation choice, not a universal precision policy. Check the rounding and precision your own measurement or financial task requires before adapting it. This plot also rejects negative means and an all-zero scale; it is not a general charting implementation.

The script completes every read, validation, calculation, and rendering step before staging a replacement output directory beside the current one. A parsing, acceptance, label, or chart-calculation failure therefore occurs before replacement begins. The file-role checks reject a raw/ or output/ link resolving outside the project; they are not a sandbox for untrusted filesystem changes. Keep the project under the runner's control.

The writer moves the previous output aside and then moves the staged directory into place. If the second move raises an ordinary OSError, it attempts to restore the previous directory. These two moves are not one crash-safe transaction. A process interruption between them can leave output/ absent and a backup beside it; failed restoration or cleanup can also require attention. The failure test below proves one handled-error path, not immunity to power loss, concurrent runners, or every filesystem error. Recovery instructions belong in the handoff precisely because the helper is bounded.

Nothing in the analysis calls a network service or requires a remembered notebook state. group_order controls output order rather than incidental input order. Those small choices do more for this colleague handoff than an elaborate orchestration system would.

Test without yesterday's output

Save this clean-room test as the new file steps/test_pipeline.py:

import csv
import hashlib
import json
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from steps import analyze

SOURCE_ROOT = Path(__file__).resolve().parents[1]

def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()

class PipelineTest(unittest.TestCase):
    def clean_copy(self, destination):
        shutil.copytree(SOURCE_ROOT / "raw", destination / "raw")
        (destination / "steps").mkdir()
        shutil.copy2(SOURCE_ROOT / "steps" / "analyze.py", destination / "steps")
        shutil.copy2(SOURCE_ROOT / "parameters.json", destination)

    def execute(self, root):
        return subprocess.run(
            [
                sys.executable,
                str(root / "steps" / "analyze.py"),
                "--parameters",
                str(root / "parameters.json"),
            ],
            cwd=root,
            capture_output=True,
            text=True,
        )

    def expected_csv(self, root):
        params = json.loads((root / "parameters.json").read_text(encoding="utf-8"))
        lines = ["group,count,total,mean"]
        for row in params["expected_summary"]:
            lines.append(f"{row['group']},{row['count']},{row['total']},{row['mean']}")
        return "\n".join(lines) + "\n"

    def update_parameters(self, root, change):
        path = root / "parameters.json"
        params = json.loads(path.read_text(encoding="utf-8"))
        change(params)
        path.write_text(json.dumps(params, indent=2) + "\n", encoding="utf-8")

    def test_clean_room_run_matches_contract_and_preserves_raw(self):
        with tempfile.TemporaryDirectory() as folder:
            root = Path(folder)
            self.clean_copy(root)
            raw_path = root / "raw" / "input.csv"
            before = digest(raw_path)
            result = self.execute(root)
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertEqual(before, digest(raw_path))
            self.assertEqual(
                (root / "output" / "summary.csv").read_text(encoding="utf-8"),
                self.expected_csv(root),
            )
            svg = (root / "output" / "figure.svg").read_text(encoding="utf-8")
            self.assertIn('role="img"', svg)
            self.assertIn('<title id="figure-title">', svg)
            self.assertIn('<desc id="figure-description">', svg)

    def test_second_run_is_byte_identical(self):
        with tempfile.TemporaryDirectory() as folder:
            root = Path(folder)
            self.clean_copy(root)
            first = self.execute(root)
            self.assertEqual(first.returncode, 0, first.stderr)
            first_hashes = [
                digest(root / "output" / "summary.csv"),
                digest(root / "output" / "figure.svg"),
            ]
            second = self.execute(root)
            self.assertEqual(second.returncode, 0, second.stderr)
            second_hashes = [
                digest(root / "output" / "summary.csv"),
                digest(root / "output" / "figure.svg"),
            ]
            self.assertEqual(first_hashes, second_hashes)

    def test_extra_csv_field_stops_before_output(self):
        with tempfile.TemporaryDirectory() as folder:
            root = Path(folder)
            self.clean_copy(root)
            input_path = root / "raw" / "input.csv"
            lines = input_path.read_text(encoding="utf-8").splitlines()
            lines[1] += ",unexpected"
            input_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
            result = self.execute(root)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("malformed CSV or extra fields", result.stderr)
            self.assertFalse((root / "output").exists())

    def test_duplicate_stops_before_output(self):
        with tempfile.TemporaryDirectory() as folder:
            root = Path(folder)
            self.clean_copy(root)
            input_path = root / "raw" / "input.csv"
            with input_path.open(newline="", encoding="utf-8") as source:
                rows = list(csv.reader(source))
            with input_path.open("a", newline="", encoding="utf-8") as target:
                csv.writer(target, lineterminator="\n").writerow(rows[1])
            result = self.execute(root)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("duplicate ID", result.stderr)
            self.assertFalse((root / "output").exists())

    def test_missing_label_stops_before_output(self):
        with tempfile.TemporaryDirectory() as folder:
            root = Path(folder)
            self.clean_copy(root)
            self.update_parameters(
                root,
                lambda params: params["group_labels"].pop(params["group_order"][0]),
            )
            result = self.execute(root)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("exactly one label per group", result.stderr)
            self.assertFalse((root / "output").exists())

    def test_zero_chart_scale_stops_before_output(self):
        with tempfile.TemporaryDirectory() as folder:
            root = Path(folder)
            self.clean_copy(root)
            input_path = root / "raw" / "input.csv"
            with input_path.open(newline="", encoding="utf-8") as source:
                reader = csv.DictReader(source)
                rows = list(reader)
                fieldnames = reader.fieldnames
            for row in rows:
                row["value"] = "0"
            with input_path.open("w", newline="", encoding="utf-8") as target:
                writer = csv.DictWriter(target, fieldnames=fieldnames, lineterminator="\n")
                writer.writeheader()
                writer.writerows(rows)

            def expect_zero(params):
                for row in params["expected_summary"]:
                    row["total"] = "0.00"
                    row["mean"] = "0.00"

            self.update_parameters(root, expect_zero)
            result = self.execute(root)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("at least one positive mean", result.stderr)
            self.assertFalse((root / "output").exists())

    def test_output_cannot_overlap_raw_input(self):
        with tempfile.TemporaryDirectory() as folder:
            root = Path(folder)
            self.clean_copy(root)
            raw_path = root / "raw" / "input.csv"
            before = digest(raw_path)
            self.update_parameters(
                root,
                lambda params: params.update(output_csv="raw/input.csv"),
            )
            result = self.execute(root)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("inside output/", result.stderr)
            self.assertEqual(before, digest(raw_path))
            self.assertFalse((root / "output").exists())

    def test_failed_directory_swap_restores_previous_output(self):
        with tempfile.TemporaryDirectory() as folder:
            root = Path(folder)
            self.clean_copy(root)
            first = self.execute(root)
            self.assertEqual(first.returncode, 0, first.stderr)
            output_paths = [
                root / "output" / "summary.csv",
                root / "output" / "figure.svg",
            ]
            before = [digest(path) for path in output_paths]
            real_replace = analyze.os.replace
            replace_calls = 0

            def fail_second_replace(source, target):
                nonlocal replace_calls
                replace_calls += 1
                if replace_calls == 2:
                    raise OSError("injected directory swap failure")
                return real_replace(source, target)

            with patch.object(analyze.os, "replace", side_effect=fail_second_replace):
                with self.assertRaisesRegex(OSError, "injected directory swap failure"):
                    analyze.run(root / "parameters.json")

            self.assertEqual(before, [digest(path) for path in output_paths])
            self.assertFalse(list(root.glob(".output-*")))

if __name__ == "__main__":
    unittest.main()

Old output cannot make these tests pass: each uses a temporary project containing only the fixture, parameters, and script. They check declared values, raw preservation, repeatability, malformed CSV with an extra field, duplicate IDs, missing labels, invalid chart scale, path collision, and restoration after an injected directory-swap error.

Read the duplicate test particularly carefully. It damages a temporary copy, runs the analysis, and expects both a non-zero result and no output directory. "The program complained" is weaker evidence than "the program complained and produced no plausible replacement result." The swap test starts with a good output because it asks a different question: can one handled replacement failure restore the previous generation?

These tests intentionally know the example's filenames and value column. If you adapt those names, change the test setup and README with them. The parameter file does not magically parameterise test code that still names raw/input.csv or output/summary.csv. Likewise, preserving the number eight is not the goal for a different project; preserving meaningful checks is.

Optional Lab framing: a raw export becomes a group figure

Save this synthetic fixture as the new file raw/input.csv:

record_id,group,value
S-101,Control,9
S-102,control,10
S-103, Control ,11
S-104,Treated,13
S-105,treated,15
S-106,TREATED,17

Save this as the new file parameters.json:

{
  "input": "raw/input.csv",
  "output_csv": "output/summary.csv",
  "output_svg": "output/figure.svg",
  "id_column": "record_id",
  "group_column": "group",
  "value_column": "value",
  "group_order": ["control", "treated"],
  "group_labels": {"control": "Control", "treated": "Treated"},
  "expected_row_count": 6,
  "expected_summary": [
    {"group": "control", "count": 3, "total": "30.00", "mean": "10.00"},
    {"group": "treated", "count": 3, "total": "45.00", "mean": "15.00"}
  ]
}

The acceptance values come from a manual calculation made before execution: Control is 9 + 10 + 11 = 30, with mean 10; Treated is 13 + 15 + 17 = 45, with mean 15. They verify arithmetic and wiring only. They do not establish that a mean is scientifically appropriate, that the groups are comparable, or that the observed difference is meaningful.

Save all files, then run the tests from the terminal at the project root. On Windows PowerShell:

py -3 --version
py -3 -m unittest steps/test_pipeline.py -v

On macOS or Linux:

python3 --version
python3 -m unittest steps/test_pipeline.py -v

All eight tests must report ok, with an overall OK. If a command is missing, use your approved Python setup route rather than installing arbitrary packages: this example has no package dependencies. If an import of steps fails, first check that the terminal is at the project root and the filenames are correct. Do not run the test file directly as a substitute for the documented module command.

Only after the tests pass, run the analysis. On Windows PowerShell:

py -3 steps/analyze.py --parameters parameters.json

On macOS or Linux:

python3 steps/analyze.py --parameters parameters.json

The expected message is PASS: 6 rows, 2 groups, outputs regenerated. These are expected results for your run, not a claim that a classroom or colleague has already performed it.

In the editor, open output/summary.csv; confirm the header group,count,total,mean and the two declared summary rows. Open output/figure.svg in a browser using its file-open action. It should show Treated longer than Control. The SVG includes an image role, title, and description, but their presence alone does not prove accessibility in every receiving application. Check that text is readable at the intended size; the inherited example uses small labels and a fixed-width layout, so zoom for inspection and revise the renderer if your actual labels collide or become unreadable. Supply the numeric table and a meaningful report caption alongside the figure. Do not repair a generated SVG by hand.

Optional Company framing: a monthly export becomes a report figure

Use the same project and code, but begin in a separate clean copy if you already ran the Lab example. Replace raw/input.csv in that copy with this synthetic fixture:

record_id,group,value
R-201,North,120
R-202,north,100
R-203, North ,80
R-204,South,90
R-205,south,110
R-206,SOUTH,130

Replace that copy's parameters.json with:

{
  "input": "raw/input.csv",
  "output_csv": "output/summary.csv",
  "output_svg": "output/figure.svg",
  "id_column": "record_id",
  "group_column": "group",
  "value_column": "value",
  "group_order": ["north", "south"],
  "group_labels": {"north": "North", "south": "South"},
  "expected_row_count": 6,
  "expected_summary": [
    {"group": "north", "count": 3, "total": "300.00", "mean": "100.00"},
    {"group": "south", "count": 3, "total": "330.00", "mean": "110.00"}
  ]
}

The checked result is North 120 + 100 + 80 = 300, mean 100; South 90 + 110 + 130 = 330, mean 110. Run the same test and analysis commands from this copy's root. Require the same eight test behaviours and inspect the new files. The figure does not explain the difference or decide whether anyone met a target. A real monthly report also needs the period, units, and source identity in its accompanying context; these invented values carry no business interpretation.

Write the handoff before asking for it

Create README.md for the selected example or adapt it to your own project. Use this complete minimum. Replace the purpose and expected-result brackets before committing; leave the colleague evidence fields blank until the colleague runs it. Do not pre-fill PASS from your own result.

# Reproducible analysis

Purpose: regenerate [the chosen result] from [the approved input version].
Recipient and recurring use: [who needs it, when, and for what next action].

## Requirements

- Python 3 with the standard library; builder's version: [exact version]
- Git for identifying the tested commit
- The included synthetic example needs no account, network, credentials, or packages
- One runner, in an ordinary local project folder; no concurrent output writers

## Files

- `raw/input.csv`: immutable synthetic input; do not edit during a run
- `parameters.json`: all paths, columns, labels, order, counts, and acceptance values
- `steps/analyze.py`: validation, summary, and output generation
- `steps/test_pipeline.py`: clean-room, rerun, and failure tests
- `.gitignore`: excludes Python caches and local environments
- `output/summary.csv` and `output/figure.svg`: generated; never hand-edit
- `output/` belongs exclusively to this pipeline and is replaced on success

## Run from a fresh checkout

1. Open a terminal at the project root and record `git rev-parse HEAD`.
   Record `py -3 --version` on Windows or `python3 --version` on macOS/Linux.
2. Hash `raw/input.csv` with the platform command below and record the before value.
3. Run the eight tests using `py -3 -m unittest steps/test_pipeline.py -v` on
   Windows or `python3 -m unittest steps/test_pipeline.py -v` on macOS/Linux.
   Stop if any test fails.
4. Confirm `output/` holds only disposable generated files and preserve any needed
   prior accepted delivery elsewhere. From this root only, remove `output` with
   `Remove-Item -Recurse -Force output` on PowerShell or `rm -r output` on
   macOS/Linux. If it is already absent, skip removal. Stop on other errors.
5. Run `py -3 steps/analyze.py --parameters parameters.json` on Windows or
   `python3 steps/analyze.py --parameters parameters.json` on macOS/Linux.
6. Stop on failure; do not use a remaining file as evidence of current success.
   On success confirm `PASS`, then inspect both generated files.
7. Hash the raw file and both outputs. Record raw-after and output run-1 values.
8. Run the analysis and hashes again. Raw before/after and each output's run-1/run-2
   hashes must match respectively.
9. Run `git status --short`; only generated output should be untracked or changed
   before you write the evidence record.
10. Open the result in the receiving application and confirm the intended use.

## Hash commands

Use the raw-only command before running, and the three-file command after each run.

PowerShell:
    Get-FileHash -Algorithm SHA256 raw/input.csv
    Get-FileHash -Algorithm SHA256 raw/input.csv, output/summary.csv, output/figure.svg

macOS:
    shasum -a 256 raw/input.csv
    shasum -a 256 raw/input.csv output/summary.csv output/figure.svg

Linux:
    sha256sum raw/input.csv
    sha256sum raw/input.csv output/summary.csv output/figure.svg

## Expected result

[Paste the two expected summary rows from the selected framing.]
The example suite must report eight passing tests. Any failed test, different value,
changed raw hash, oral instruction, or edited output is a failed handoff.

## Failure and recovery

Stop using the attempted delivery on any failure. Preserve the error and input
identity outside `output/`. Do not label a previous accepted result as current.
Fix the responsible source version, step, parameter, or instruction, then repeat
from a clean copy. Never change raw evidence or acceptance values just to pass.
After an interrupted write, do not delete `.output-backup-*` or `.output-stage-*`
directories blindly. Stop all runners, preserve those directories, and rebuild
the recorded accepted input and commit in a separate clean copy. Compare with
the prior accepted hashes before restoring a result to use. Ask the maintainer
if the accepted version or permissions cannot be established.
Maintainer/contact route: [actual person or team and approved contact location].

## Next approved input and limits

This example proves reproducibility for its synthetic fixture and declared contract.
Keep that known-answer fixture. Record a new input identity and independently
review its expected result before adapting the snapshot example to a new delivery.
The data owner must approve schema, grouping, units, statistic, and use.
Do not add live connections or unattended scheduling as part of this handoff.

## Colleague handoff record

Commit: [full tested commit hash]
Runner and date: [name or role, YYYY-MM-DD]
Environment: [operating system and exact Python version]
Input identity and approved location: [version/location]
Test result: [8/8 pass or fail for the unchanged example]
Raw SHA-256 before/after: [hash] / [hash]
Summary SHA-256 run 1/run 2: [hash] / [hash]
Figure SHA-256 run 1/run 2: [hash] / [hash]
Recipient use: [application, action, and any interpretation/readability problem]
Coaching needed: [none, or exact missing instruction]
Decision: [PASS or FAIL, based on the evidence]

For another tool or artifact, replace the example commands and comparison with the actual ones your colleague will use. An exported project without .git cannot run git rev-parse HEAD; supply its source commit in a separate handoff note, and have the runner record that provided identity rather than pretending to query a repository that is absent.

Before handoff, follow the README yourself: remove only disposable output, run the checks, run the analysis twice, and review the result. If this is a new standalone example repository, initialise and commit it using the following commands from its root. If your chosen project already has Git history, use that repository's normal change process; do not initialise another repository inside it.

git init
git add .gitignore README.md parameters.json raw/input.csv steps/analyze.py steps/test_pipeline.py
git diff --cached --check
git diff --cached

Inspect the displayed staged diff. Confirm no cache, environment, sensitive file, unrelated work, or generated output is staged. The explicit raw/input.csv addition is appropriate for the synthetic fixture, not blanket permission to commit a real export. Only after this review:

git commit -m "Build reproducible analysis pipeline"
git status --short
git rev-parse HEAD

Record the full commit. Give the colleague a fresh clone or clean exported project at that version, with approved access to any inputs not carried in the repository. Do not send only the script: without parameters, fixture, tests, and README, it is not the tested pipeline.

The colleague follows the README without a screen-share. For the unchanged example, a pass has identical raw hashes before and after, identical output hashes across runs, eight passing tests, declared summary values, a usable received result, and Coaching needed: none. Keep their completed record outside disposable output, attached to the tested commit. If you later commit that record, its commit is the evidence commit, not a reason to replace the recorded identity of the code they actually ran.

6. What goes wrong

Raw data is edited in place

Symptom: the pipeline now produces the expected result, but the received value that caused the mismatch has disappeared and no transformation explains it.

Fix: restore the recorded raw input from the approved source. Put a justified correction in a visible step or obtain a newly versioned source file from its owner. Rerun the checks and record which input version was used. Preserve the failed evidence rather than rewriting its history.

Generated output is hand-fixed

Symptom: the figure is correct in the report, but removing output/ and rerunning brings the old label or value back.

Fix: repair the responsible parameter or step, remove only disposable generated output, regenerate it, and repeat the clean-room check. Never use drawing or spreadsheet edits as an invisible final pipeline stage. If a caption carries essential units or caveats, include its source in the versioned project too.

A path works on one computer

Symptom: the author can run the analysis, while the colleague receives file not found for a username, drive letter, home folder, or mounted share.

Fix: keep exercise files under one project root and resolve declared relative paths from a stable file such as parameters.json, not from the author's current directory. Test from another location. For an approved external input store, document access separately rather than hard-coding the author's credentials or drive mapping.

Parameters are scattered through code

Symptom: changing a label requires editing the script, a notebook cell, and the figure; one old value survives.

Fix: move run choices to one reviewed parameter file. Search the steps for duplicate path, group, threshold, and label literals. Keep implementation constants in code, but keep analysis choices in the declared contract. Update the test setup if you change the example's filenames; it intentionally uses fixed fixture paths.

Dependencies exist only in the author's environment

Symptom: an import succeeds for the builder and fails for the colleague, or two environments format a result differently.

Fix: declare the runtime and packages. This example avoids external packages; a larger project should use the team's approved lockfile and environment setup. Test from a clean environment rather than installing missing pieces ad hoc during handoff. Record the runtime that actually ran, not just "Python installed."

An old output makes a broken run look successful

Symptom: the command fails, yet a plausible summary.csv remains from yesterday and is mistaken for today's result.

Fix: start the handoff without disposable output, stop on a non-zero command, and verify that deliberate bad input creates no output in a clean room. A file's presence is not proof that the current run made it. In a recurring task, tell the recipient that the new delivery failed; a previous accepted result may remain available only with its old input identity clearly stated.

A write is interrupted

Symptom: output/ is missing, or a .output-backup-* directory remains after a failed run.

Fix: stop other runs and preserve the error and candidate directories before cleaning anything. Rebuild the recorded accepted commit and input in a separate clean copy, then compare with the accepted hashes. Do not assume a staging folder is complete or that the newest-looking folder is correct. The example's handled-error test is not a power-loss recovery guarantee.

The next input fails the old acceptance values

Symptom: a new month arrives, and the script rejects its different totals even though the file looks reasonable.

Fix: first verify input identity, completeness, and meaning. The snapshot example intentionally checks an exact answer. For a small new snapshot, review a new expected answer independently; for a broader project, keep the known-answer regression fixture and design suitable delivery checks separately. Never teach a check to accept whatever the current code produced.

The README requires oral footnotes

Symptom: the colleague succeeds only after being told where to run the command, which file to rename, or which unexplained warning to ignore.

Fix: mark the handoff incomplete and add the missing fact to the README. Restart from the new commit. Do not convert coaching into undocumented pipeline state. Also record questions about using the result: "Which period is this?" is a handoff defect even when every command worked.

7. Do it yourself: make your result dependable

This is a sequence, not a reading-time target. Spend time on the parts your project needs; do not replace a sound existing analysis simply to reproduce the example.

Choose one recurring result. Continue your chosen project and name the recipient's next action. Confirm the input, workspace, output location, and sharing permissions. If you need a safe practice case first, use one optional synthetic fixture.

Expose the route. Identify unchanged raw evidence, visible transformations, parameters, checks, and generated files. In the editor, trace the function calls or workflow connections from input to output. Remove hidden paths and manual finishing steps. Write down any domain decision the code cannot justify, such as units or whether a mean is appropriate.

Establish the answer before trusting the run. Keep a small known-answer case and calculate its expected result independently. Define the rerun comparison before execution. For the supplied Python example, use the two declared summary rows and byte identity; for another format, specify the justified comparison your colleague can actually perform.

Prove useful failure. Run the project's bounded checks. In a disposable copy, introduce one meaningful defect and confirm that it stops before a new result is accepted. Preserve the error and recover without editing the raw source or generated result. The supplied eight tests are an inspectable example, not a mandatory test count for every project.

Rebuild and inspect. Begin without disposable output, run, inspect the actual result, and rerun. Check unchanged input and the agreed comparison. Open the result where the recipient will use it, not only in the development editor. If text is unreadable or context is missing, fix the versioned source of that defect.

Write and version the handoff. State exact commands or interface actions, environment, input identity, expected files, acceptance rules, failure response, and the next-input procedure. Review the intended diff and commit only permitted project material. Keep evidence and previously accepted deliveries outside the replaceable output folder.

Let the colleague take over. Give them the version and instructions, not your terminal history. They obtain a fresh copy, run the checks, rebuild, compare, use the result, and record what happened. If they cannot finish unaided, preserve that evidence and repair the missing instruction or step. If no colleague is available yet, label the result prepared for handoff, not passed.

8. Exit check

Deliver one versioned analysis pipeline with its independent colleague handoff record. The record identifies the tested commit, environment, and input; shows that the raw evidence was preserved; reports the project's checks and agreed rerun comparison; records regeneration from absent disposable output; and describes the recipient's actual use of the result. Include the failure/recovery evidence and instructions for the next approved input.

For the unchanged Python example, that means 8/8 tests, the declared CSV rows, unchanged raw SHA-256, byte-identical CSV and SVG across two runs, and no coaching. For your own workflow, report the checks you actually defined and ran rather than borrowing an 8/8 claim from this example. Do not claim the next-input procedure was exercised unless it was.

The handoff is not passed if the colleague starts from your working directory, relies on old output, edits raw evidence or generated output, receives an unrecorded instruction, installs an undeclared dependency, cannot identify the version, or obtains a different result under the agreed comparison. A screenshot, a figure alone, and your own successful run do not replace this evidence.

The useful end state is simple: the colleague knows which result they can use, how it was rebuilt, and what to do when the next run does not pass.

9. Rule to remember

If a colleague cannot rebuild and recognise the result without you, the handoff is not finished.

10. Further reading & tools

T10-L02 - Clean a messy export establishes unchanged raw input, explicit transformations, validation, and regenerated output. Continue its project rather than cleaning another file for its own sake.
T08-L02 - Code without being a developer provides the bounded coding contract, diff review, known-input test, and deliberate failure used here.
Python pathlib documentation (opens in a new tab) is the inherited primary reference for project-relative path handling.
Python unittest documentation (opens in a new tab) is the inherited primary reference for the standard-library test runner used in the example.
Python os.replace documentation (opens in a new tab) describes the rename primitive used by the output writer; do not confuse one operation with a transaction covering the whole two-move sequence.
Git documentation (opens in a new tab) is the inherited primary command reference for commits, status, diffs, and commit identifiers.
T10-L04 - Pipelines over real data continues with live sources and quality gates that stop schema or data failures.
T08-L04 - Working in a real repo carries versioned evidence into a larger shared codebase.
Tools index provides current tool references. KNIME is an optional visual workflow route; an approved coding assistant such as Cursor is an optional editor aid. Neither removes the need to package and independently check the actual workflow.

11. How this book was made

This edition preserves the published book's complete Python analysis, eight-test example, two synthetic fixtures, hand calculations, and reproducibility explanations. The source record does not identify its original writer, so this edition does not assign that work to an invented author.

Astra (OpenAI, openai/gpt-6-astra) wrote the current revision: the learner-chosen project path, clearer tool actions, explanations of the implementation's boundaries, and a handoff centred on the colleague's use and recovery. There was no delegated writing team in this pass. Documentation retrieval is not an observed execution, and the example's expected outcomes are not claims of classroom or colleague success. Independent review and publication are separate from preparing this draft.