homassistant_chrony_addon/chrony-addon/rootfs/var/www/app.py

#!/usr/bin/env python3
"""Chrony NTP Server Web UI for Home Assistant."""

import json
import os
import subprocess
from flask import Flask, render_template, jsonify

app = Flask(__name__)

OPTIONS_FILE = "/data/options.json"


def get_options():
    """Load current options from Home Assistant."""
    try:
        with open(OPTIONS_FILE, 'r') as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {
            "ntp_pools": [
                {"pool": "pool.ntp.org", "iburst": True}
            ],
            "ntp_servers": [
                {"server": "time.google.com", "iburst": True}
            ],
            "allow_clients": ["192.168.0.0/16", "10.0.0.0/8", "172.16.0.0/12"],
            "enable_ntp_server": True,
            "detailed_logging": False
        }


def get_chrony_status():
    """Get current chrony synchronization status."""
    status = {
        "tracking": {},
        "sources": [],
        "clients": []
    }

    try:
        # Get tracking info
        result = subprocess.run(
            ["chronyc", "-c", "tracking"],
            capture_output=True, text=True, timeout=5
        )
        if result.returncode == 0:
            fields = result.stdout.strip().split(',')
            if len(fields) >= 14:
                status["tracking"] = {
                    "reference_id": fields[0],
                    "reference_name": fields[1],
                    "stratum": fields[2],
                    "ref_time": fields[3],
                    "system_time": fields[4],
                    "last_offset": fields[5],
                    "rms_offset": fields[6],
                    "frequency": fields[7],
                    "residual_freq": fields[8],
                    "skew": fields[9],
                    "root_delay": fields[10],
                    "root_dispersion": fields[11],
                    "update_interval": fields[12],
                    "leap_status": fields[13]
                }

        # Get sources
        result = subprocess.run(
            ["chronyc", "-c", "sources"],
            capture_output=True, text=True, timeout=5
        )
        if result.returncode == 0:
            for line in result.stdout.strip().split('\n'):
                if line:
                    fields = line.split(',')
                    if len(fields) >= 10:
                        status["sources"].append({
                            "mode": fields[0],
                            "state": fields[1],
                            "name": fields[2],
                            "stratum": fields[3],
                            "poll": fields[4],
                            "reach": fields[5],
                            "last_rx": fields[6],
                            "last_sample_offset": fields[7],
                            "last_sample_error": fields[8]
                        })

        # Get clients (if server mode enabled)
        result = subprocess.run(
            ["chronyc", "-c", "clients"],
            capture_output=True, text=True, timeout=5
        )
        if result.returncode == 0:
            for line in result.stdout.strip().split('\n'):
                if line:
                    fields = line.split(',')
                    if len(fields) >= 5:
                        status["clients"].append({
                            "hostname": fields[0],
                            "ntp_requests": fields[1],
                            "drop": fields[2],
                            "last_rx": fields[4]
                        })

    except subprocess.TimeoutExpired:
        status["error"] = "Timeout getting chrony status"
    except Exception as e:
        status["error"] = str(e)

    return status


@app.route('/')
def index():
    """Main dashboard page."""
    options = get_options()
    status = get_chrony_status()
    return render_template('index.html', options=options, status=status)


@app.route('/api/status')
def api_status():
    """API endpoint for current status."""
    return jsonify(get_chrony_status())


@app.route('/api/options')
def api_options():
    """API endpoint for current options."""
    return jsonify(get_options())


@app.route('/api/force-sync', methods=['POST'])
def force_sync():
    """Force immediate time synchronization."""
    try:
        result = subprocess.run(
            ["chronyc", "makestep"],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode == 0:
            return jsonify({"success": True, "message": "Time step applied"})
        else:
            return jsonify({"success": False, "message": result.stderr}), 500
    except Exception as e:
        return jsonify({"success": False, "message": str(e)}), 500


@app.route('/api/burst', methods=['POST'])
def burst_sync():
    """Trigger burst mode for faster synchronization."""
    try:
        result = subprocess.run(
            ["chronyc", "burst", "4/4"],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode == 0:
            return jsonify({"success": True, "message": "Burst mode activated"})
        else:
            return jsonify({"success": False, "message": result.stderr}), 500
    except Exception as e:
        return jsonify({"success": False, "message": str(e)}), 500


if __name__ == '__main__':
    # Get ingress port from environment
    port = int(os.environ.get('INGRESS_PORT', 8099))
    app.run(host='0.0.0.0', port=port, debug=False)