9858738e82
- Pass pets list to dashboard template (Critical 3)
- Sanitize upload filenames with werkzeug secure_filename (Important 6)
- Validate image extension against allowlist (Important 7)
- 404 check for pet existence in upload and update endpoints (Important 8)
- Save uploads to training/{pet_name}/ not training/{pet_id}/ (Important 11)
- Wire /train to PetTrainer with background thread (Important 12)
- Wire /api/training-status to stored trainer status (Important 13)
- Implement /api/highlights from sightings + wildlife queries (Important 14)
- Cap limit param at 500 on sightings and wildlife endpoints (Important 15)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
302 lines
9.4 KiB
Python
302 lines
9.4 KiB
Python
"""Pets blueprint — pet registration, sightings, labeling, training."""
|
|
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from flask import Blueprint, current_app, jsonify, render_template, request
|
|
from werkzeug.utils import secure_filename
|
|
|
|
pets_bp = Blueprint("pets", __name__, url_prefix="/pets")
|
|
|
|
ALLOWED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"}
|
|
|
|
|
|
def _engine():
|
|
return current_app.config.get("DB_ENGINE")
|
|
|
|
|
|
@pets_bp.route("/")
|
|
def dashboard():
|
|
engine = _engine()
|
|
all_pets = []
|
|
if engine:
|
|
from vigilar.storage.queries import get_all_pets
|
|
all_pets = get_all_pets(engine)
|
|
return render_template("pets/dashboard.html", pets=all_pets)
|
|
|
|
|
|
@pets_bp.route("/register", methods=["POST"])
|
|
def register_pet():
|
|
data = request.get_json(silent=True) or {}
|
|
name = data.get("name")
|
|
species = data.get("species")
|
|
if not name or not species:
|
|
return jsonify({"error": "name and species required"}), 400
|
|
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"error": "database not available"}), 503
|
|
|
|
from vigilar.storage.queries import insert_pet
|
|
pet_id = insert_pet(
|
|
engine,
|
|
name=name,
|
|
species=species,
|
|
breed=data.get("breed"),
|
|
color_description=data.get("color_description"),
|
|
photo_path=data.get("photo_path"),
|
|
)
|
|
return jsonify({"id": pet_id, "name": name, "species": species})
|
|
|
|
|
|
@pets_bp.route("/api/status")
|
|
def pet_status():
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"pets": []})
|
|
|
|
from vigilar.storage.queries import get_all_pets, get_pet_last_location
|
|
pets_list = get_all_pets(engine)
|
|
result = []
|
|
for pet in pets_list:
|
|
last = get_pet_last_location(engine, pet["id"])
|
|
result.append({
|
|
**pet,
|
|
"last_seen_ts": last["ts"] if last else None,
|
|
"last_camera": last["camera_id"] if last else None,
|
|
})
|
|
return jsonify({"pets": result})
|
|
|
|
|
|
@pets_bp.route("/api/sightings")
|
|
def pet_sightings():
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"sightings": []})
|
|
|
|
pet_id = request.args.get("pet_id")
|
|
camera_id = request.args.get("camera_id")
|
|
since_ts = request.args.get("since", type=float)
|
|
limit = min(request.args.get("limit", 100, type=int), 500)
|
|
|
|
from vigilar.storage.queries import get_pet_sightings
|
|
sightings = get_pet_sightings(engine, pet_id=pet_id, camera_id=camera_id,
|
|
since_ts=since_ts, limit=limit)
|
|
return jsonify({"sightings": sightings})
|
|
|
|
|
|
@pets_bp.route("/api/wildlife")
|
|
def wildlife_sightings():
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"sightings": []})
|
|
|
|
threat_level = request.args.get("threat_level")
|
|
camera_id = request.args.get("camera_id")
|
|
since_ts = request.args.get("since", type=float)
|
|
limit = min(request.args.get("limit", 100, type=int), 500)
|
|
|
|
from vigilar.storage.queries import get_wildlife_sightings
|
|
sightings = get_wildlife_sightings(engine, threat_level=threat_level, camera_id=camera_id,
|
|
since_ts=since_ts, limit=limit)
|
|
return jsonify({"sightings": sightings})
|
|
|
|
|
|
@pets_bp.route("/api/unlabeled")
|
|
def unlabeled_crops():
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"crops": []})
|
|
|
|
species = request.args.get("species")
|
|
limit = request.args.get("limit", 50, type=int)
|
|
|
|
from vigilar.storage.queries import get_unlabeled_sightings
|
|
crops = get_unlabeled_sightings(engine, species=species, limit=limit)
|
|
return jsonify({"crops": crops})
|
|
|
|
|
|
@pets_bp.route("/<sighting_id>/label", methods=["POST"])
|
|
def label_crop(sighting_id: str):
|
|
data = request.get_json(silent=True) or {}
|
|
pet_id = data.get("pet_id")
|
|
if not pet_id:
|
|
return jsonify({"error": "pet_id required"}), 400
|
|
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"error": "database not available"}), 503
|
|
|
|
from vigilar.storage.queries import label_sighting
|
|
try:
|
|
label_sighting(engine, int(sighting_id), pet_id)
|
|
except (ValueError, TypeError):
|
|
return jsonify({"error": "invalid sighting_id"}), 400
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@pets_bp.route("/<pet_id>/upload", methods=["POST"])
|
|
def upload_photo(pet_id: str):
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"error": "database not available"}), 503
|
|
|
|
from vigilar.storage.queries import get_pet, insert_training_image
|
|
pet = get_pet(engine, pet_id)
|
|
if not pet:
|
|
return jsonify({"error": "Pet not found"}), 404
|
|
|
|
files = request.files.getlist("photos")
|
|
if not files:
|
|
return jsonify({"error": "no files provided"}), 400
|
|
|
|
cfg = current_app.config.get("VIGILAR_CONFIG")
|
|
pets_cfg = current_app.config.get("PETS_CONFIG")
|
|
if pets_cfg:
|
|
training_dir = pets_cfg.training_dir
|
|
else:
|
|
training_dir = os.path.join(cfg.system.data_dir if cfg else "/tmp/vigilar", "training")
|
|
|
|
pet_dir = Path(training_dir) / pet["name"].lower()
|
|
pet_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
saved = []
|
|
for f in files:
|
|
if not f.filename:
|
|
continue
|
|
safe_name = secure_filename(f.filename)
|
|
if not safe_name:
|
|
continue
|
|
if Path(safe_name).suffix.lower() not in ALLOWED_IMAGE_EXTENSIONS:
|
|
continue
|
|
dest = pet_dir / f"{int(time.time() * 1000)}_{safe_name}"
|
|
f.save(str(dest))
|
|
insert_training_image(engine, pet_id=pet_id, image_path=str(dest), source="upload")
|
|
saved.append(str(dest))
|
|
|
|
return jsonify({"ok": True, "saved": len(saved)})
|
|
|
|
|
|
@pets_bp.route("/<pet_id>/update", methods=["POST"])
|
|
def update_pet(pet_id: str):
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"error": "database not available"}), 503
|
|
|
|
from vigilar.storage.queries import get_pet
|
|
pet = get_pet(engine, pet_id)
|
|
if not pet:
|
|
return jsonify({"error": "Pet not found"}), 404
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
from sqlalchemy import update as sa_update
|
|
|
|
from vigilar.storage.schema import pets as pets_table
|
|
|
|
allowed = {"name", "species", "breed", "color_description", "photo_path"}
|
|
values = {k: v for k, v in data.items() if k in allowed}
|
|
if not values:
|
|
return jsonify({"error": "no valid fields to update"}), 400
|
|
|
|
with engine.begin() as conn:
|
|
conn.execute(sa_update(pets_table).where(pets_table.c.id == pet_id).values(**values))
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@pets_bp.route("/<pet_id>/delete", methods=["DELETE"])
|
|
def delete_pet(pet_id: str):
|
|
engine = _engine()
|
|
if engine is None:
|
|
return jsonify({"error": "database not available"}), 503
|
|
|
|
from vigilar.storage.schema import pets as pets_table
|
|
with engine.begin() as conn:
|
|
conn.execute(pets_table.delete().where(pets_table.c.id == pet_id))
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@pets_bp.route("/train", methods=["POST"])
|
|
def train_model():
|
|
engine = _engine()
|
|
pets_cfg = current_app.config.get("PETS_CONFIG")
|
|
if not pets_cfg:
|
|
return jsonify({"error": "Pet detection not configured"}), 400
|
|
|
|
from vigilar.detection.trainer import PetTrainer
|
|
trainer = PetTrainer(
|
|
training_dir=pets_cfg.training_dir,
|
|
model_output_path=pets_cfg.pet_id_model_path,
|
|
)
|
|
|
|
ready, msg = trainer.check_readiness(min_images=pets_cfg.min_training_images)
|
|
if not ready:
|
|
return jsonify({"error": msg}), 400
|
|
|
|
# Store trainer on app for status polling
|
|
current_app.extensions["pet_trainer"] = trainer
|
|
|
|
import threading
|
|
thread = threading.Thread(target=trainer.train, daemon=True)
|
|
thread.start()
|
|
|
|
return jsonify({"ok": True, "status": "training_started", "message": msg})
|
|
|
|
|
|
@pets_bp.route("/api/training-status")
|
|
def training_status():
|
|
trainer = current_app.extensions.get("pet_trainer")
|
|
if not trainer:
|
|
return jsonify({"status": "idle"})
|
|
s = trainer.status
|
|
return jsonify({
|
|
"status": "training" if s.is_training else "complete" if s.progress >= 1.0 else "idle",
|
|
"progress": round(s.progress, 2),
|
|
"epoch": s.epoch,
|
|
"total_epochs": s.total_epochs,
|
|
"accuracy": round(s.accuracy, 4),
|
|
"error": s.error,
|
|
})
|
|
|
|
|
|
@pets_bp.route("/api/highlights")
|
|
def highlights():
|
|
engine = _engine()
|
|
if not engine:
|
|
return jsonify({"highlights": []})
|
|
|
|
import time as time_mod
|
|
since = time_mod.time() - 86400
|
|
|
|
from vigilar.storage.queries import get_pet_sightings, get_wildlife_sightings
|
|
|
|
result = []
|
|
|
|
# High-confidence pet sightings
|
|
sightings = get_pet_sightings(engine, since_ts=since, limit=50)
|
|
for s in sightings:
|
|
if s.get("confidence", 0) > 0.8 and s.get("pet_id"):
|
|
result.append({
|
|
"type": "pet_sighting",
|
|
"pet_id": s.get("pet_id"),
|
|
"camera_id": s["camera_id"],
|
|
"timestamp": s["ts"],
|
|
"confidence": s.get("confidence"),
|
|
})
|
|
|
|
# Wildlife events
|
|
wildlife = get_wildlife_sightings(engine, since_ts=since, limit=20)
|
|
for w in wildlife:
|
|
result.append({
|
|
"type": "wildlife",
|
|
"species": w["species"],
|
|
"threat_level": w["threat_level"],
|
|
"camera_id": w["camera_id"],
|
|
"timestamp": w["ts"],
|
|
"confidence": w.get("confidence"),
|
|
})
|
|
|
|
# Sort by timestamp descending, limit to 20
|
|
result.sort(key=lambda x: x["timestamp"], reverse=True)
|
|
return jsonify({"highlights": result[:20]})
|