ExoMy_Cuno/ExoMy_Software-master/scripts/exomy_admin_api.py

#!/usr/bin/env python3
import http.server
import json
import shutil
import socketserver
import subprocess
import time

PORT = 8082
EXOMY_CONTAINER = 'exomy_autostart'


class ReusableThreadingTCPServer(socketserver.ThreadingTCPServer):
    allow_reuse_address = True


def run_command(command):
    result = subprocess.run(command, capture_output=True, text=True, check=False)
    return result.stdout.strip(), result.stderr.strip(), result.returncode


def read_cpu_temperature():
    try:
        with open('/sys/class/thermal/thermal_zone0/temp', 'r', encoding='utf-8') as handle:
            raw_value = handle.read().strip()
        return f'{int(raw_value) / 1000:.1f} °C'
    except (FileNotFoundError, ValueError):
        return 'unbekannt'


def format_uptime():
    try:
        with open('/proc/uptime', 'r', encoding='utf-8') as handle:
            uptime_seconds = int(float(handle.read().split()[0]))
    except (FileNotFoundError, ValueError, IndexError):
        return 'unbekannt'

    days, remainder = divmod(uptime_seconds, 86400)
    hours, remainder = divmod(remainder, 3600)
    minutes, _ = divmod(remainder, 60)

    parts = []
    if days:
        parts.append(f'{days}d')
    if days or hours:
        parts.append(f'{hours}h')
    parts.append(f'{minutes}m')
    return ' '.join(parts)


def format_disk_free():
    usage = shutil.disk_usage('/')
    free_gb = usage.free / (1024 ** 3)
    total_gb = usage.total / (1024 ** 3)
    return f'{free_gb:.1f} GB frei / {total_gb:.1f} GB'


def get_ip_addresses():
    stdout, _, returncode = run_command(['hostname', '-I'])
    if returncode != 0 or not stdout:
        return 'unbekannt'
    ips = [item for item in stdout.split() if item]
    return ', '.join(ips) if ips else 'unbekannt'


def get_wifi_status():
    stdout, _, returncode = run_command(['nmcli', '-t', '-f', 'DEVICE,STATE,CONNECTION', 'device'])
    if returncode != 0 or not stdout:
        return 'unbekannt'

    for line in stdout.splitlines():
        parts = line.split(':', 2)
        if len(parts) != 3:
            continue
        device, state, connection = parts
        if device == 'wlan0':
            if state == 'connected':
                return f'verbunden: {connection}'
            return state

    return 'kein wlan0'


def get_systemd_state(service_name):
    stdout, _, returncode = run_command(['systemctl', 'is-active', service_name])
    if returncode == 0 and stdout:
        return stdout
    return stdout or 'unknown'


def get_container_state(container_name):
    stdout, _, returncode = run_command(['docker', 'inspect', '-f', '{{.State.Status}}', container_name])
    if returncode == 0 and stdout:
        return stdout
    return 'unknown'


def collect_status():
    return {
        'status': 'Bereit',
        'system': {
            'wifi': get_wifi_status(),
            'ips': get_ip_addresses(),
            'cpu_temperature': read_cpu_temperature(),
            'uptime': format_uptime(),
            'disk_free': format_disk_free(),
        },
        'services': {
            'camera': get_systemd_state('exomy-camera-stream.service'),
            'admin_api': get_systemd_state('exomy-admin-api.service'),
            'exomy': get_container_state(EXOMY_CONTAINER),
        }
    }


def restart_camera_service():
    subprocess.Popen(['systemctl', 'restart', 'exomy-camera-stream.service'])


def restart_exomy_container():
    subprocess.Popen(['docker', 'restart', EXOMY_CONTAINER])


def reboot_raspberry_pi():
    subprocess.Popen(['systemctl', 'reboot'])


def shutdown_raspberry_pi():
    subprocess.Popen(['systemctl', 'poweroff'])


class Handler(http.server.BaseHTTPRequestHandler):
    def end_headers(self):
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        self.send_header('Access-Control-Allow-Headers', 'Content-Type')
        super().end_headers()

    def write_json(self, status_code, payload):
        body = json.dumps(payload).encode('utf-8')
        self.send_response(status_code)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self):
        self.send_response(204)
        self.end_headers()

    def do_GET(self):
        if self.path == '/api/status':
            self.write_json(200, collect_status())
            return
        self.write_json(404, {'status': 'Unbekannt'})

    def do_POST(self):
        actions = {
            '/api/restart-camera': {
                'handler': restart_camera_service,
                'status': 'Kameradienst wird neu gestartet'
            },
            '/api/restart-container': {
                'handler': restart_exomy_container,
                'status': 'ExoMy-Container wird neu gestartet'
            },
            '/api/reboot': {
                'handler': reboot_raspberry_pi,
                'status': 'Raspberry Pi wird neu gestartet'
            },
            '/api/shutdown': {
                'handler': shutdown_raspberry_pi,
                'status': 'Raspberry Pi wird heruntergefahren'
            }
        }

        action = actions.get(self.path)
        if action is None:
            self.write_json(404, {'status': 'Unbekannt'})
            return

        self.write_json(202, {'status': action['status']})
        self.wfile.flush()
        time.sleep(0.1)
        action['handler']()

    def log_message(self, fmt, *args):
        return


if __name__ == '__main__':
    with ReusableThreadingTCPServer(('0.0.0.0', PORT), Handler) as server:
        server.daemon_threads = True
        server.serve_forever()