karung-masuk_frigate-counter/web_app.py

#!/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/<date_str>')
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/<camera_name>')
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=8080, debug=True)