#!/usr/bin/env python3 """ Flask Web Application for Frigate Counter Displays daily karung counts from SQLite database """ from flask import Flask, render_template, jsonify, request import sqlite3 from datetime import datetime, date, timedelta from typing import List, Dict, Optional import os app = Flask(__name__) DB_PATH = 'karung_counts.db' def get_db_connection(): """Create a database connection""" conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def get_daily_counts(selected_date: Optional[str] = None) -> List[Dict]: """Get counts for a specific date or all dates""" conn = get_db_connection() cursor = conn.cursor() if selected_date: cursor.execute(''' SELECT camera_name, date, counter_value, timestamp FROM karung_counts WHERE date = ? ORDER BY timestamp DESC ''', (selected_date,)) else: cursor.execute(''' SELECT camera_name, date, counter_value, timestamp FROM karung_counts ORDER BY date DESC, timestamp DESC ''') rows = cursor.fetchall() conn.close() return [{ 'camera_name': row['camera_name'], 'date': row['date'], 'counter_value': row['counter_value'], 'timestamp': row['timestamp'] } for row in rows] def get_summary_by_date() -> List[Dict]: """Get daily summary grouped by date""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' SELECT date, COUNT(DISTINCT camera_name) as camera_count, SUM(counter_value) as total_count, MAX(timestamp) as last_update FROM karung_counts GROUP BY date ORDER BY date DESC ''') rows = cursor.fetchall() conn.close() return [{ 'date': row['date'], 'camera_count': row['camera_count'], 'total_count': row['total_count'] or 0, 'last_update': row['last_update'] } for row in rows] def get_available_dates() -> List[str]: """Get list of all dates with data""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' SELECT DISTINCT date FROM karung_counts ORDER BY date DESC ''') rows = cursor.fetchall() conn.close() return [row['date'] for row in rows] def get_camera_summary() -> List[Dict]: """Get summary by camera""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' SELECT camera_name, COUNT(DISTINCT date) as day_count, SUM(counter_value) as total_count, MAX(timestamp) as last_update FROM karung_counts GROUP BY camera_name ORDER BY camera_name ''') rows = cursor.fetchall() conn.close() return [{ 'camera_name': row['camera_name'], 'day_count': row['day_count'], 'total_count': row['total_count'] or 0, 'last_update': row['last_update'] } for row in rows] @app.route('/') def index(): """Main page showing daily summary""" summary = get_summary_by_date() cameras = get_camera_summary() available_dates = get_available_dates() today = date.today().isoformat() today_data = get_daily_counts(today) return render_template('index.html', summary=summary, cameras=cameras, today_data=today_data, available_dates=available_dates, today=today) @app.route('/date/') def date_detail(date_str: str): """Show details for a specific date""" counts = get_daily_counts(date_str) available_dates = get_available_dates() # Calculate total for this date total = sum(c['counter_value'] for c in counts) return render_template('date_detail.html', date=date_str, counts=counts, total=total, available_dates=available_dates) @app.route('/camera/') def camera_detail(camera_name: str): """Show details for a specific camera""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' SELECT date, counter_value, timestamp FROM karung_counts WHERE camera_name = ? ORDER BY date DESC ''', (camera_name,)) rows = cursor.fetchall() conn.close() counts = [{ 'date': row['date'], 'counter_value': row['counter_value'], 'timestamp': row['timestamp'] } for row in rows] total = sum(c['counter_value'] for c in counts) return render_template('camera_detail.html', camera_name=camera_name, counts=counts, total=total) @app.route('/api/counts') def api_counts(): """API endpoint to get all counts""" date_filter = request.args.get('date') counts = get_daily_counts(date_filter) return jsonify({ 'counts': counts, 'date_filter': date_filter }) @app.route('/api/summary') def api_summary(): """API endpoint to get daily summary""" summary = get_summary_by_date() return jsonify({'summary': summary}) @app.route('/api/cameras') def api_cameras(): """API endpoint to get camera summary""" cameras = get_camera_summary() return jsonify({'cameras': cameras}) if __name__ == '__main__': # Ensure database exists if not os.path.exists(DB_PATH): print(f"Error: Database {DB_PATH} not found!") exit(1) # Run Flask app on all interfaces, port 5000 app.run(host='0.0.0.0', port=8898, debug=True)