iterminate
b7e2d54bd2
Pass a 'next' hidden field from the device detail unassign form so the route redirects back to the device page instead of the dashboard.
570 lines
23 KiB
Python
570 lines
23 KiB
Python
from flask import Flask, render_template, redirect, url_for, request, flash, abort
|
|
from sqlalchemy import create_engine, func
|
|
from sqlalchemy.orm import scoped_session, sessionmaker
|
|
|
|
from models import Base, Battery, Device
|
|
|
|
|
|
def create_app(config_object="config"):
|
|
app = Flask(__name__)
|
|
app.config.from_object(config_object)
|
|
|
|
engine = create_engine(
|
|
app.config["SQLALCHEMY_DATABASE_URI"],
|
|
pool_pre_ping=True,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
|
|
db = scoped_session(sessionmaker(bind=engine))
|
|
|
|
@app.teardown_appcontext
|
|
def remove_session(exc=None):
|
|
db.remove()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Dashboard
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/")
|
|
def dashboard():
|
|
batteries = db.query(Battery).order_by(Battery.label).all()
|
|
storage_locations = [
|
|
r[0] for r in db.query(Battery.storage_location)
|
|
.filter(Battery.storage_location.isnot(None))
|
|
.distinct().order_by(Battery.storage_location).all()
|
|
]
|
|
devices = db.query(Device).order_by(Device.name).all()
|
|
return render_template("dashboard.html", batteries=batteries,
|
|
storage_locations=storage_locations, devices=devices)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — add
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/add", methods=["GET", "POST"])
|
|
def battery_add():
|
|
if request.method == "POST":
|
|
f = request.form
|
|
brand = f.get("brand", "").strip()
|
|
notes = f.get("notes", "").strip() or None
|
|
try:
|
|
count = max(1, min(50, int(f.get("count", 1) or 1)))
|
|
except (ValueError, TypeError):
|
|
count = 1
|
|
|
|
if not brand:
|
|
flash("Brand is required.", "error")
|
|
brands = [r[0] for r in db.query(Battery.brand).distinct().order_by(Battery.brand).all()]
|
|
return render_template("battery_add.html",
|
|
form_brand="", form_count=1, form_notes=notes or "",
|
|
brands=brands), 400
|
|
|
|
def _int(key):
|
|
v = f.get(key, "").strip()
|
|
try:
|
|
return int(v) if v else None
|
|
except ValueError:
|
|
return None
|
|
|
|
size = f.get("size", "").strip() or None
|
|
chemistry = f.get("chemistry", "").strip() or None
|
|
capacity_mah = _int("capacity_mah")
|
|
purchase_date = f.get("purchase_date", "").strip() or None
|
|
storage_location = f.get("storage_location", "").strip() or None
|
|
|
|
existing = db.query(func.count(Battery.id)).filter_by(brand=brand).scalar()
|
|
for i in range(count):
|
|
label = f"{brand} {existing + i + 1:03d}"
|
|
db.add(Battery(label=label, brand=brand, status="available", notes=notes,
|
|
size=size, chemistry=chemistry, capacity_mah=capacity_mah,
|
|
purchase_date=purchase_date, storage_location=storage_location))
|
|
db.commit()
|
|
flash(f"Added {count} {brand} batter{'y' if count == 1 else 'ies'}.", "success")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
brands = [r[0] for r in db.query(Battery.brand).distinct().order_by(Battery.brand).all()]
|
|
storage_locations = [
|
|
r[0] for r in db.query(Battery.storage_location)
|
|
.filter(Battery.storage_location.isnot(None))
|
|
.distinct().order_by(Battery.storage_location).all()
|
|
]
|
|
return render_template("battery_add.html", form_count=1, brands=brands,
|
|
storage_locations=storage_locations)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — detail
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/<int:battery_id>")
|
|
def battery_detail(battery_id):
|
|
battery = db.get(Battery, battery_id)
|
|
if battery is None:
|
|
abort(404)
|
|
storage_locations = [
|
|
r[0] for r in db.query(Battery.storage_location)
|
|
.filter(Battery.storage_location.isnot(None))
|
|
.distinct().order_by(Battery.storage_location).all()
|
|
]
|
|
return render_template("battery_detail.html", battery=battery,
|
|
storage_locations=storage_locations)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — edit notes
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/<int:battery_id>/edit-details", methods=["POST"])
|
|
def battery_edit_details(battery_id):
|
|
battery = db.get(Battery, battery_id)
|
|
if battery is None:
|
|
abort(404)
|
|
f = request.form
|
|
|
|
def _int(key):
|
|
v = f.get(key, "").strip()
|
|
try:
|
|
return int(v) if v else None
|
|
except ValueError:
|
|
return None
|
|
|
|
battery.notes = f.get("notes", "").strip() or None
|
|
battery.size = f.get("size", "").strip() or None
|
|
battery.chemistry = f.get("chemistry", "").strip() or None
|
|
battery.capacity_mah = _int("capacity_mah")
|
|
battery.tested_capacity_mah = _int("tested_capacity_mah")
|
|
battery.tested_date = f.get("tested_date", "").strip() or None
|
|
battery.charge_cycles = _int("charge_cycles")
|
|
battery.purchase_date = f.get("purchase_date", "").strip() or None
|
|
battery.storage_location = f.get("storage_location", "").strip() or None
|
|
|
|
db.commit()
|
|
flash("Details updated.", "success")
|
|
return redirect(url_for("battery_detail", battery_id=battery_id))
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — assign
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/<int:battery_id>/assign", methods=["GET", "POST"])
|
|
def battery_assign(battery_id):
|
|
battery = db.get(Battery, battery_id)
|
|
if battery is None:
|
|
abort(404)
|
|
|
|
if battery.is_retired():
|
|
flash("Cannot assign a retired battery.", "error")
|
|
return redirect(url_for("battery_detail", battery_id=battery_id))
|
|
|
|
devices = db.query(Device).order_by(Device.name).all()
|
|
|
|
if request.method == "POST":
|
|
device_id = request.form.get("device_id", type=int)
|
|
device = db.get(Device, device_id)
|
|
if device is None:
|
|
flash("Device not found.", "error")
|
|
return render_template("assign.html", battery=battery, devices=devices)
|
|
|
|
if device.installed_count() >= device.battery_slots:
|
|
flash(
|
|
f"{device.name} is already full "
|
|
f"({device.battery_slots}/{device.battery_slots} slots used).",
|
|
"error",
|
|
)
|
|
return render_template("assign.html", battery=battery, devices=devices)
|
|
|
|
# Warn on brand mix (non-blocking)
|
|
existing_brands = device.installed_brands()
|
|
if existing_brands and battery.brand not in existing_brands:
|
|
flash(
|
|
f"Warning: {device.name} already has batteries from "
|
|
f"{', '.join(existing_brands)}. Mixing brands is not recommended.",
|
|
"warning",
|
|
)
|
|
|
|
# Unassign from previous device if needed
|
|
if battery.device_id is not None:
|
|
battery.device_id = None
|
|
|
|
battery.status = "installed"
|
|
battery.device_id = device.id
|
|
db.commit()
|
|
flash(f"{battery.label} assigned to {device.name}.", "success")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
return render_template("assign.html", battery=battery, devices=devices)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — unassign
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/<int:battery_id>/unassign", methods=["POST"])
|
|
def battery_unassign(battery_id):
|
|
battery = db.get(Battery, battery_id)
|
|
if battery is None:
|
|
abort(404)
|
|
battery.status = "available"
|
|
battery.device_id = None
|
|
db.commit()
|
|
flash(f"{battery.label} unassigned and marked available.", "success")
|
|
next_url = request.form.get("next", "")
|
|
return redirect(next_url if next_url.startswith("/") else url_for("dashboard"))
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — retire
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/<int:battery_id>/retire", methods=["POST"])
|
|
def battery_retire(battery_id):
|
|
battery = db.get(Battery, battery_id)
|
|
if battery is None:
|
|
abort(404)
|
|
if battery.is_retired():
|
|
flash(f"{battery.label} is already retired.", "error")
|
|
return redirect(url_for("battery_detail", battery_id=battery_id))
|
|
battery.status = "retired"
|
|
battery.device_id = None
|
|
db.commit()
|
|
flash(f"{battery.label} has been retired.", "success")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — unretire
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/<int:battery_id>/unretire", methods=["POST"])
|
|
def battery_unretire(battery_id):
|
|
battery = db.get(Battery, battery_id)
|
|
if battery is None:
|
|
abort(404)
|
|
if not battery.is_retired():
|
|
flash(f"{battery.label} is not retired.", "error")
|
|
return redirect(url_for("battery_detail", battery_id=battery_id))
|
|
battery.status = "available"
|
|
db.commit()
|
|
flash(f"{battery.label} is now available again.", "success")
|
|
return redirect(url_for("battery_detail", battery_id=battery_id))
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — delete
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/<int:battery_id>/delete", methods=["GET", "POST"])
|
|
def battery_delete(battery_id):
|
|
battery = db.get(Battery, battery_id)
|
|
if battery is None:
|
|
abort(404)
|
|
if request.method == "POST":
|
|
label = battery.label
|
|
db.delete(battery)
|
|
db.commit()
|
|
flash(f"Battery {label} permanently deleted.", "success")
|
|
return redirect(url_for("dashboard"))
|
|
return render_template("battery_delete.html", battery=battery)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Battery — bulk action
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/battery/bulk-action", methods=["POST"])
|
|
def battery_bulk_action():
|
|
ids = request.form.getlist("battery_ids", type=int)
|
|
if not ids:
|
|
flash("No batteries selected.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
batteries = db.query(Battery).filter(Battery.id.in_(ids)).all()
|
|
action = request.form.get("action")
|
|
n = len(batteries)
|
|
|
|
if action == "retire":
|
|
for b in batteries:
|
|
b.status = "retired"
|
|
b.device_id = None
|
|
db.commit()
|
|
flash(f"Retired {n} batter{'y' if n == 1 else 'ies'}.", "success")
|
|
elif action == "delete":
|
|
for b in batteries:
|
|
db.delete(b)
|
|
db.commit()
|
|
flash(f"Deleted {n} batter{'y' if n == 1 else 'ies'}.", "success")
|
|
elif action == "unassign":
|
|
count = sum(1 for b in batteries if b.is_installed())
|
|
for b in batteries:
|
|
if b.is_installed():
|
|
b.status = "available"
|
|
b.device_id = None
|
|
db.commit()
|
|
flash(f"Unassigned {count} batter{'y' if count == 1 else 'ies'}.", "success")
|
|
elif action == "set_brand":
|
|
new_brand = request.form.get("new_brand", "").strip()
|
|
if not new_brand:
|
|
flash("Brand name is required.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
for b in batteries:
|
|
b.brand = new_brand
|
|
db.commit()
|
|
flash(f"Updated brand to '{new_brand}' for {n} batter{'y' if n == 1 else 'ies'}.", "success")
|
|
elif action == "install_device":
|
|
device_id = request.form.get("device_id", type=int)
|
|
if not device_id:
|
|
flash("Please select a device.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
device = db.get(Device, device_id)
|
|
if device is None:
|
|
flash("Device not found.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
already_here = [b for b in batteries if b.device_id == device.id]
|
|
retired_sel = [b for b in batteries if b.is_retired()]
|
|
to_process = [b for b in batteries
|
|
if b.device_id != device.id and not b.is_retired()]
|
|
|
|
if not to_process:
|
|
flash("No eligible batteries to install.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
free_slots = device.battery_slots - device.installed_count()
|
|
if len(to_process) > free_slots:
|
|
flash(
|
|
f"{device.name} only has {free_slots} free slot(s), "
|
|
f"but {len(to_process)} need installing.",
|
|
"error",
|
|
)
|
|
return redirect(url_for("dashboard"))
|
|
|
|
existing_brands = device.installed_brands()
|
|
new_brands = set(b.brand for b in to_process)
|
|
if existing_brands and (new_brands - existing_brands):
|
|
flash(f"Warning: mixing brands in {device.name}.", "warning")
|
|
|
|
for b in to_process:
|
|
b.status = "installed"
|
|
b.device_id = device.id
|
|
db.commit()
|
|
|
|
msg = (f"Installed {len(to_process)} "
|
|
f"batter{'y' if len(to_process) == 1 else 'ies'} in {device.name}.")
|
|
notes = []
|
|
if already_here:
|
|
notes.append(f"{len(already_here)} already there")
|
|
if retired_sel:
|
|
notes.append(f"{len(retired_sel)} retired skipped")
|
|
if notes:
|
|
msg += f" ({', '.join(notes)}.)"
|
|
flash(msg, "success")
|
|
|
|
elif action == "set_field":
|
|
field_name = request.form.get("field_name", "").strip()
|
|
field_value = request.form.get("field_value", "").strip() or None
|
|
allowed = {"brand", "storage_location"}
|
|
if field_name not in allowed:
|
|
flash("Invalid field.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
if field_name == "brand" and not field_value:
|
|
flash("Brand name is required.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
for b in batteries:
|
|
setattr(b, field_name, field_value)
|
|
db.commit()
|
|
label = field_name.replace("_", " ").title()
|
|
flash(f"Set {label} on {n} batter{'y' if n == 1 else 'ies'}.", "success")
|
|
else:
|
|
flash("Unknown action.", "error")
|
|
|
|
return redirect(url_for("dashboard"))
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Devices — list
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/device/")
|
|
def device_list():
|
|
devices = db.query(Device).order_by(Device.name).all()
|
|
return render_template("device_list.html", devices=devices)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Devices — add
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/device/add", methods=["GET", "POST"])
|
|
def device_add():
|
|
if request.method == "POST":
|
|
name = request.form.get("name", "").strip()
|
|
slots_raw = request.form.get("battery_slots", "1").strip()
|
|
notes = request.form.get("notes", "").strip() or None
|
|
|
|
if not name:
|
|
flash("Device name is required.", "error")
|
|
return render_template("device_add.html"), 400
|
|
|
|
try:
|
|
slots = int(slots_raw)
|
|
if slots < 1:
|
|
raise ValueError
|
|
except ValueError:
|
|
flash("Battery slots must be a positive integer.", "error")
|
|
return render_template("device_add.html",
|
|
form_name=name, form_notes=notes or ""), 400
|
|
|
|
if db.query(Device).filter_by(name=name).first():
|
|
flash(f"A device named '{name}' already exists.", "error")
|
|
return render_template("device_add.html",
|
|
form_name=name, form_slots=slots,
|
|
form_notes=notes or ""), 400
|
|
|
|
device = Device(name=name, battery_slots=slots, notes=notes)
|
|
db.add(device)
|
|
db.commit()
|
|
flash(f"Device '{name}' added.", "success")
|
|
return redirect(url_for("device_list"))
|
|
|
|
return render_template("device_add.html")
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Devices — detail
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/device/<int:device_id>")
|
|
def device_detail(device_id):
|
|
device = db.get(Device, device_id)
|
|
if device is None:
|
|
abort(404)
|
|
brands = [r[0] for r in db.query(Battery.brand).distinct().order_by(Battery.brand).all()]
|
|
available_batteries = (db.query(Battery)
|
|
.filter_by(status="available")
|
|
.order_by(Battery.label).all())
|
|
return render_template("device_detail.html", device=device, brands=brands,
|
|
available_batteries=available_batteries)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Devices — install batteries
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/device/<int:device_id>/install", methods=["POST"])
|
|
def device_install(device_id):
|
|
device = db.get(Device, device_id)
|
|
if device is None:
|
|
abort(404)
|
|
|
|
brands = request.form.getlist("brand[]")
|
|
qtys_raw = request.form.getlist("qty[]")
|
|
|
|
pairs = []
|
|
for brand, qty_raw in zip(brands, qtys_raw):
|
|
brand = brand.strip()
|
|
try:
|
|
qty = int(qty_raw)
|
|
except (ValueError, TypeError):
|
|
qty = 0
|
|
if brand and qty > 0:
|
|
pairs.append((brand, qty))
|
|
|
|
if not pairs:
|
|
flash("No batteries specified.", "error")
|
|
return redirect(url_for("device_detail", device_id=device_id))
|
|
|
|
free_slots = device.battery_slots - device.installed_count()
|
|
total_requested = sum(qty for _, qty in pairs)
|
|
if total_requested > free_slots:
|
|
flash(
|
|
f"Only {free_slots} slot(s) free, but {total_requested} requested.",
|
|
"error",
|
|
)
|
|
return redirect(url_for("device_detail", device_id=device_id))
|
|
|
|
# Validate availability before writing anything
|
|
for brand, qty in pairs:
|
|
available_count = (
|
|
db.query(func.count(Battery.id))
|
|
.filter_by(brand=brand, status="available")
|
|
.scalar()
|
|
)
|
|
if available_count < qty:
|
|
flash(
|
|
f"Need {qty} {brand}, but only {available_count} available.",
|
|
"error",
|
|
)
|
|
return redirect(url_for("device_detail", device_id=device_id))
|
|
|
|
# All checks passed — perform installs
|
|
total_installed = 0
|
|
for brand, qty in pairs:
|
|
batch = (
|
|
db.query(Battery)
|
|
.filter_by(brand=brand, status="available")
|
|
.order_by(Battery.id)
|
|
.limit(qty)
|
|
.all()
|
|
)
|
|
for b in batch:
|
|
b.status = "installed"
|
|
b.device_id = device.id
|
|
total_installed += 1
|
|
|
|
db.commit()
|
|
flash(
|
|
f"Installed {total_installed} batter{'y' if total_installed == 1 else 'ies'}"
|
|
f" into {device.name}.",
|
|
"success",
|
|
)
|
|
return redirect(url_for("device_detail", device_id=device_id))
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Devices — install one specific battery
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/device/<int:device_id>/install-one", methods=["POST"])
|
|
def device_install_one(device_id):
|
|
device = db.get(Device, device_id)
|
|
if device is None:
|
|
abort(404)
|
|
battery_id = request.form.get("battery_id", type=int)
|
|
battery = db.get(Battery, battery_id)
|
|
if battery is None or not battery.is_available():
|
|
flash("Battery not found or not available.", "error")
|
|
return redirect(url_for("device_detail", device_id=device_id))
|
|
if device.installed_count() >= device.battery_slots:
|
|
flash(
|
|
f"{device.name} is full "
|
|
f"({device.battery_slots}/{device.battery_slots} slots used).",
|
|
"error",
|
|
)
|
|
return redirect(url_for("device_detail", device_id=device_id))
|
|
existing_brands = device.installed_brands()
|
|
if existing_brands and battery.brand not in existing_brands:
|
|
flash(
|
|
f"Warning: {device.name} already has batteries from "
|
|
f"{', '.join(existing_brands)}. Mixing brands is not recommended.",
|
|
"warning",
|
|
)
|
|
battery.status = "installed"
|
|
battery.device_id = device.id
|
|
db.commit()
|
|
flash(f"{battery.label} installed in {device.name}.", "success")
|
|
return redirect(url_for("device_detail", device_id=device_id))
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Devices — delete
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@app.route("/device/<int:device_id>/delete", methods=["POST"])
|
|
def device_delete(device_id):
|
|
device = db.get(Device, device_id)
|
|
if device is None:
|
|
abort(404)
|
|
for battery in device.batteries:
|
|
battery.status = "available"
|
|
battery.device_id = None
|
|
name = device.name
|
|
db.delete(device)
|
|
db.commit()
|
|
flash(f"Device '{name}' deleted. All batteries marked available.", "success")
|
|
return redirect(url_for("device_list"))
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
if __name__ == "__main__":
|
|
app.run(debug=True)
|