From 3b2029d3b8ca3e4105fdee00831df16cbb16e0d5 Mon Sep 17 00:00:00 2001 From: Darek Date: Sun, 26 Apr 2026 20:03:58 -0500 Subject: [PATCH] Add logbook entries, data export page, and JSON import --- app.py | 414 +++++++++++++++++++++++++++++++++- models.py | 44 +++- templates/base.html | 14 ++ templates/battery_detail.html | 32 +++ templates/device_detail.html | 31 +++ templates/export.html | 70 ++++++ templates/import.html | 73 ++++++ tests/test_acceptance.py | 205 ++++++++++++++++- 8 files changed, 877 insertions(+), 6 deletions(-) create mode 100644 templates/export.html create mode 100644 templates/import.html diff --git a/app.py b/app.py index b11c697..a832bfd 100644 --- a/app.py +++ b/app.py @@ -1,11 +1,15 @@ -from flask import Flask, render_template, redirect, url_for, request, flash, abort, send_from_directory, jsonify +from flask import Flask, render_template, redirect, url_for, request, flash, abort, send_from_directory, jsonify, Response from sqlalchemy import create_engine, func from sqlalchemy.orm import scoped_session, sessionmaker from datetime import datetime, date, timedelta +import csv +import io +import json import re +import zipfile -from models import Base, Battery, BatteryPctLog, Device, CapacityTest, ChargeLog +from models import Base, Battery, BatteryPctLog, Device, CapacityTest, ChargeLog, Logbook def _parse_date(val: str) -> str | None: @@ -231,7 +235,8 @@ def create_app(config_object="config"): pct_logs=pct_logs, charge_logs_data=charge_logs_data, capacity_tests_data=capacity_tests_data, - pct_logs_data=pct_logs_data) + pct_logs_data=pct_logs_data, + logbook_entries=battery.logbook_entries) # ------------------------------------------------------------------ # # Battery — edit notes @@ -757,7 +762,8 @@ def create_app(config_object="config"): device_locations=device_locations, device_battery_sizes=device_battery_sizes, ha_enabled=ha_client.enabled, - ha_live_pct=ha_live_pct) + ha_live_pct=ha_live_pct, + logbook_entries=device.logbook_entries) # ------------------------------------------------------------------ # # Devices — edit @@ -954,6 +960,406 @@ def create_app(config_object="config"): ) return redirect(url_for("device_list")) + # ------------------------------------------------------------------ # + # Export + # ------------------------------------------------------------------ # + + def _batteries_csv(): + rows = db.query(Battery).order_by(Battery.id).all() + buf = io.StringIO() + w = csv.writer(buf) + w.writerow(["id", "label", "brand", "status", "device_id", "device_name", + "size", "chemistry", "capacity_mah", "tested_capacity_mah", + "tested_date", "charge_cycles", "purchase_date", + "storage_location", "battery_percentage", "notes"]) + for b in rows: + w.writerow([b.id, b.label, b.brand, b.status, b.device_id or "", + b.device.name if b.device else "", + b.size or "", b.chemistry or "", b.capacity_mah or "", + b.tested_capacity_mah or "", b.tested_date or "", + b.charge_cycles or "", b.purchase_date or "", + b.storage_location or "", b.battery_percentage or "", b.notes or ""]) + return buf.getvalue() + + def _devices_csv(): + rows = db.query(Device).order_by(Device.id).all() + buf = io.StringIO() + w = csv.writer(buf) + w.writerow(["id", "name", "battery_slots", "installed_count", + "device_type", "battery_size", "location", "ha_entity_id", "notes"]) + for d in rows: + w.writerow([d.id, d.name, d.battery_slots, d.installed_count(), + d.device_type or "", d.battery_size or "", + d.location or "", d.ha_entity_id or "", d.notes or ""]) + return buf.getvalue() + + def _charge_logs_csv(): + rows = db.query(ChargeLog).order_by(ChargeLog.id).all() + buf = io.StringIO() + w = csv.writer(buf) + w.writerow(["id", "battery_id", "battery_label", "charged_date", "increment_cycles", "notes"]) + for l in rows: + w.writerow([l.id, l.battery_id, l.battery.label, + l.charged_date, l.increment_cycles, l.notes or ""]) + return buf.getvalue() + + def _capacity_tests_csv(): + rows = db.query(CapacityTest).order_by(CapacityTest.id).all() + buf = io.StringIO() + w = csv.writer(buf) + w.writerow(["id", "battery_id", "battery_label", "tested_capacity_mah", "tested_date", "notes"]) + for t in rows: + w.writerow([t.id, t.battery_id, t.battery.label, + t.tested_capacity_mah, t.tested_date, t.notes or ""]) + return buf.getvalue() + + def _pct_logs_csv(): + rows = db.query(BatteryPctLog).order_by(BatteryPctLog.id).all() + buf = io.StringIO() + w = csv.writer(buf) + w.writerow(["id", "battery_id", "battery_label", "percentage", "recorded_at", "source"]) + for l in rows: + w.writerow([l.id, l.battery_id, l.battery.label, + l.percentage, l.recorded_at, l.source or ""]) + return buf.getvalue() + + @app.route("/export") + def export_page(): + return render_template("export.html") + + @app.route("/export/batteries.csv") + def export_batteries_csv(): + return Response(_batteries_csv(), mimetype="text/csv", + headers={"Content-Disposition": "attachment; filename=batteries.csv"}) + + @app.route("/export/devices.csv") + def export_devices_csv(): + return Response(_devices_csv(), mimetype="text/csv", + headers={"Content-Disposition": "attachment; filename=devices.csv"}) + + @app.route("/export/charge-logs.csv") + def export_charge_logs_csv(): + return Response(_charge_logs_csv(), mimetype="text/csv", + headers={"Content-Disposition": "attachment; filename=charge-logs.csv"}) + + @app.route("/export/capacity-tests.csv") + def export_capacity_tests_csv(): + return Response(_capacity_tests_csv(), mimetype="text/csv", + headers={"Content-Disposition": "attachment; filename=capacity-tests.csv"}) + + @app.route("/export/pct-logs.csv") + def export_pct_logs_csv(): + return Response(_pct_logs_csv(), mimetype="text/csv", + headers={"Content-Disposition": "attachment; filename=pct-logs.csv"}) + + @app.route("/export/csv.zip") + def export_csv_zip(): + zip_buf = io.BytesIO() + with zipfile.ZipFile(zip_buf, "w", compression=zipfile.ZIP_DEFLATED) as zf: + zf.writestr("batteries.csv", _batteries_csv()) + zf.writestr("devices.csv", _devices_csv()) + zf.writestr("charge-logs.csv", _charge_logs_csv()) + zf.writestr("capacity-tests.csv", _capacity_tests_csv()) + zf.writestr("pct-logs.csv", _pct_logs_csv()) + zip_buf.seek(0) + fname = f"battery-tracker-{datetime.utcnow().strftime('%Y%m%d')}.zip" + return Response(zip_buf.read(), mimetype="application/zip", + headers={"Content-Disposition": f"attachment; filename={fname}"}) + + @app.route("/export/all.json") + def export_json(): + batteries = db.query(Battery).order_by(Battery.id).all() + devices = db.query(Device).order_by(Device.id).all() + charge_logs = db.query(ChargeLog).order_by(ChargeLog.id).all() + capacity_tests = db.query(CapacityTest).order_by(CapacityTest.id).all() + pct_logs = db.query(BatteryPctLog).order_by(BatteryPctLog.id).all() + payload = { + "exported_at": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"), + "batteries": [ + {"id": b.id, "label": b.label, "brand": b.brand, "status": b.status, + "device_id": b.device_id, "device_name": b.device.name if b.device else None, + "size": b.size, "chemistry": b.chemistry, "capacity_mah": b.capacity_mah, + "tested_capacity_mah": b.tested_capacity_mah, "tested_date": b.tested_date, + "charge_cycles": b.charge_cycles, "purchase_date": b.purchase_date, + "storage_location": b.storage_location, + "battery_percentage": b.battery_percentage, "notes": b.notes} + for b in batteries + ], + "devices": [ + {"id": d.id, "name": d.name, "battery_slots": d.battery_slots, + "installed_count": d.installed_count(), "device_type": d.device_type, + "battery_size": d.battery_size, "location": d.location, + "ha_entity_id": d.ha_entity_id, "notes": d.notes} + for d in devices + ], + "charge_logs": [ + {"id": l.id, "battery_id": l.battery_id, "battery_label": l.battery.label, + "charged_date": l.charged_date, "increment_cycles": l.increment_cycles, + "notes": l.notes} + for l in charge_logs + ], + "capacity_tests": [ + {"id": t.id, "battery_id": t.battery_id, "battery_label": t.battery.label, + "tested_capacity_mah": t.tested_capacity_mah, "tested_date": t.tested_date, + "notes": t.notes} + for t in capacity_tests + ], + "pct_logs": [ + {"id": l.id, "battery_id": l.battery_id, "battery_label": l.battery.label, + "percentage": l.percentage, "recorded_at": l.recorded_at, "source": l.source} + for l in pct_logs + ], + } + return Response(json.dumps(payload, indent=2), mimetype="application/json", + headers={"Content-Disposition": "attachment; filename=battery-tracker-export.json"}) + + # ------------------------------------------------------------------ # + # Import + # ------------------------------------------------------------------ # + + @app.route("/import", methods=["GET", "POST"]) + def import_page(): + if request.method == "GET": + return render_template("import.html", results=None) + + # --- file presence --- + if "file" not in request.files or request.files["file"].filename == "": + flash("No file selected.", "error") + return render_template("import.html", results=None), 400 + + f = request.files["file"] + + # --- size check (10 MB) --- + f.seek(0, 2) + size = f.tell() + f.seek(0) + if size > 10 * 1024 * 1024: + flash("File too large (max 10 MB).", "error") + return render_template("import.html", results=None), 400 + + # --- JSON parse --- + try: + data = json.load(f) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + flash(f"Invalid JSON file: {e}", "error") + return render_template("import.html", results=None), 400 + + # --- key validation --- + if not isinstance(data, dict) or "batteries" not in data or "devices" not in data: + flash("Invalid format: JSON must contain 'batteries' and 'devices' keys.", "error") + return render_template("import.html", results=None), 400 + + device_id_map = {} + battery_id_map = {} + devices_created = devices_skipped = 0 + batteries_created = batteries_skipped = 0 + charge_logs_appended = charge_logs_skipped = 0 + capacity_tests_appended = capacity_tests_skipped = 0 + pct_logs_appended = pct_logs_skipped = 0 + + try: + # --- devices --- + for d in data.get("devices", []): + old_id = d.get("id") + name = (d.get("name") or "").strip() + if not name: + devices_skipped += 1 + continue + existing = db.query(Device).filter_by(name=name).first() + if existing: + if old_id is not None: + device_id_map[old_id] = existing.id + devices_skipped += 1 + else: + new_dev = Device( + name = name, + battery_slots = d.get("battery_slots") or 1, + device_type = d.get("device_type") or None, + battery_size = d.get("battery_size") or "", + location = d.get("location") or None, + ha_entity_id = d.get("ha_entity_id") or None, + notes = d.get("notes") or None, + ) + db.add(new_dev) + db.flush() + if old_id is not None: + device_id_map[old_id] = new_dev.id + devices_created += 1 + + # --- batteries --- + for b in data.get("batteries", []): + old_id = b.get("id") + label = (b.get("label") or "").strip() + if not label: + batteries_skipped += 1 + continue + existing = db.query(Battery).filter_by(label=label).first() + if existing: + if old_id is not None: + battery_id_map[old_id] = existing.id + batteries_skipped += 1 + else: + old_device_id = b.get("device_id") + new_device_id = device_id_map.get(old_device_id) if old_device_id is not None else None + source_status = b.get("status", "available") + if new_device_id is not None: + status = "installed" + elif source_status == "retired": + status = "retired" + else: + status = "available" + new_bat = Battery( + label = label, + brand = b.get("brand") or "Unknown", + status = status, + device_id = new_device_id, + size = b.get("size") or None, + chemistry = b.get("chemistry") or None, + capacity_mah = b.get("capacity_mah") or None, + tested_capacity_mah = b.get("tested_capacity_mah") or None, + tested_date = b.get("tested_date") or None, + charge_cycles = b.get("charge_cycles") or None, + purchase_date = b.get("purchase_date") or None, + storage_location = b.get("storage_location") or None, + battery_percentage = b.get("battery_percentage") or None, + notes = b.get("notes") or None, + ) + db.add(new_bat) + db.flush() + if old_id is not None: + battery_id_map[old_id] = new_bat.id + batteries_created += 1 + + # --- charge logs --- + for cl in data.get("charge_logs", []): + old_bat_id = cl.get("battery_id") + new_bat_id = battery_id_map.get(old_bat_id) + if new_bat_id is None: + charge_logs_skipped += 1 + continue + db.add(ChargeLog( + battery_id = new_bat_id, + charged_date = cl.get("charged_date") or date.today().isoformat(), + increment_cycles = cl.get("increment_cycles") or 0, + notes = cl.get("notes") or None, + )) + charge_logs_appended += 1 + + # --- capacity tests --- + for ct in data.get("capacity_tests", []): + old_bat_id = ct.get("battery_id") + new_bat_id = battery_id_map.get(old_bat_id) + if new_bat_id is None: + capacity_tests_skipped += 1 + continue + db.add(CapacityTest( + battery_id = new_bat_id, + tested_capacity_mah = ct.get("tested_capacity_mah") or 0, + tested_date = ct.get("tested_date") or date.today().isoformat(), + notes = ct.get("notes") or None, + )) + capacity_tests_appended += 1 + + # --- pct logs --- + for pl in data.get("pct_logs", []): + old_bat_id = pl.get("battery_id") + new_bat_id = battery_id_map.get(old_bat_id) + if new_bat_id is None: + pct_logs_skipped += 1 + continue + db.add(BatteryPctLog( + battery_id = new_bat_id, + percentage = pl.get("percentage") or 0, + recorded_at = pl.get("recorded_at") or datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"), + source = pl.get("source") or None, + )) + pct_logs_appended += 1 + + db.commit() + + except Exception as e: + db.rollback() + flash(f"Import failed: {e}", "error") + return render_template("import.html", results=None), 500 + + flash( + f"Import complete: {devices_created} device(s) and {batteries_created} battery/ies created.", + "success", + ) + results = { + "devices_created": devices_created, + "devices_skipped": devices_skipped, + "batteries_created": batteries_created, + "batteries_skipped": batteries_skipped, + "charge_logs_appended": charge_logs_appended, + "charge_logs_skipped": charge_logs_skipped, + "capacity_tests_appended": capacity_tests_appended, + "capacity_tests_skipped": capacity_tests_skipped, + "pct_logs_appended": pct_logs_appended, + "pct_logs_skipped": pct_logs_skipped, + } + return render_template("import.html", results=results) + + # ------------------------------------------------------------------ # + # Logbook + # ------------------------------------------------------------------ # + + @app.route("/battery//logbook/add", methods=["POST"]) + def battery_logbook_add(battery_id): + battery = db.get(Battery, battery_id) + if battery is None: + abort(404) + body = request.form.get("body", "").strip() + if not body: + flash("Entry text is required.", "error") + else: + battery.logbook_entries.append( + Logbook(body=body, recorded_at=datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")) + ) + db.commit() + flash("Logbook entry added.", "success") + return redirect(url_for("battery_detail", battery_id=battery_id)) + + @app.route("/battery//logbook//delete", methods=["POST"]) + def battery_logbook_delete(battery_id, entry_id): + battery = db.get(Battery, battery_id) + if battery is None: + abort(404) + entry = db.get(Logbook, entry_id) + if entry and entry in battery.logbook_entries: + battery.logbook_entries.remove(entry) + db.commit() + flash("Logbook entry deleted.", "success") + return redirect(url_for("battery_detail", battery_id=battery_id)) + + @app.route("/device//logbook/add", methods=["POST"]) + def device_logbook_add(device_id): + device = db.get(Device, device_id) + if device is None: + abort(404) + body = request.form.get("body", "").strip() + if not body: + flash("Entry text is required.", "error") + else: + device.logbook_entries.append( + Logbook(body=body, recorded_at=datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")) + ) + db.commit() + flash("Logbook entry added.", "success") + return redirect(url_for("device_detail", device_id=device_id)) + + @app.route("/device//logbook//delete", methods=["POST"]) + def device_logbook_delete(device_id, entry_id): + device = db.get(Device, device_id) + if device is None: + abort(404) + entry = db.get(Logbook, entry_id) + if entry and entry in device.logbook_entries: + device.logbook_entries.remove(entry) + db.commit() + flash("Logbook entry deleted.", "success") + return redirect(url_for("device_detail", device_id=device_id)) + return app diff --git a/models.py b/models.py index 1ad6d25..94b072a 100644 --- a/models.py +++ b/models.py @@ -1,10 +1,25 @@ from datetime import datetime -from sqlalchemy import Column, Integer, String, Text, ForeignKey +from sqlalchemy import Column, Integer, String, Text, ForeignKey, Table from sqlalchemy.orm import declarative_base, relationship Base = declarative_base() +# Association tables for logbook (defined before ORM classes so they can be +# referenced as `secondary` in relationship()). String-based FKs are resolved +# lazily when create_all() binds the metadata. +battery_logbook_table = Table( + "battery_logbook", Base.metadata, + Column("battery_id", Integer, ForeignKey("battery.id", ondelete="CASCADE"), primary_key=True), + Column("logbook_id", Integer, ForeignKey("logbook.id", ondelete="CASCADE"), primary_key=True), +) + +device_logbook_table = Table( + "device_logbook", Base.metadata, + Column("device_id", Integer, ForeignKey("device.id", ondelete="CASCADE"), primary_key=True), + Column("logbook_id", Integer, ForeignKey("logbook.id", ondelete="CASCADE"), primary_key=True), +) + class Device(Base): __tablename__ = "device" @@ -19,6 +34,12 @@ class Device(Base): ha_entity_id = Column(String(100), nullable=True) # e.g. "sensor.tv_remote_battery" batteries = relationship("Battery", back_populates="device") + logbook_entries = relationship( + "Logbook", secondary=device_logbook_table, + order_by="Logbook.recorded_at.desc()", + cascade="all, delete-orphan", + single_parent=True, + ) def installed_count(self): return sum(1 for b in self.batteries if b.status == "installed") @@ -70,6 +91,12 @@ class Battery(Base): order_by="BatteryPctLog.recorded_at.desc()", cascade="all, delete-orphan", ) + logbook_entries = relationship( + "Logbook", secondary=battery_logbook_table, + order_by="Logbook.recorded_at.desc()", + cascade="all, delete-orphan", + single_parent=True, + ) def is_available(self): return self.status == "available" @@ -127,3 +154,18 @@ class BatteryPctLog(Base): def __repr__(self): return f"" + + +class Logbook(Base): + __tablename__ = "logbook" + + id = Column(Integer, primary_key=True, autoincrement=True) + body = Column(Text, nullable=False) + recorded_at = Column(String(19), nullable=False) # "YYYY-MM-DD HH:MM:SS" + # Note: SQLite has no native DATETIME type (stores dates as text). + # On MariaDB this would be a native DATETIME column — swap String(19) for + # DateTime when migrating, since SQLAlchemy's DateTime maps to DATETIME there. + # String(19) is used here to match every other timestamp column in the codebase. + + def __repr__(self): + return f"" diff --git a/templates/base.html b/templates/base.html index 68f3df1..68ab225 100644 --- a/templates/base.html +++ b/templates/base.html @@ -334,6 +334,20 @@ {% block content %}{% endblock %} + + +