Add logbook entries, data export page, and JSON import
This commit is contained in:
@@ -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/<int:battery_id>/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/<int:battery_id>/logbook/<int:entry_id>/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/<int:device_id>/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/<int:device_id>/logbook/<int:entry_id>/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
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user