#!/usr/bin/env python3
# SPDX-License-Identifier: MPL-2.0
"""Read readiness, never dial, send messages, enter PINs or open camera nodes."""
import json
import os
from pathlib import Path
import subprocess
import time


def command(args):
    try:
        result = subprocess.run(args, capture_output=True, text=True, timeout=12)
        if result.returncode:
            return None
        return result.stdout
    except (OSError, subprocess.TimeoutExpired):
        return None


def bus(*args):
    text = command(['busctl', '--system', '--json=short', *args])
    try:
        return json.loads(text)['data'] if text is not None else None
    except (ValueError, KeyError, TypeError):
        return None


def properties(interface):
    data = bus('call', 'org.ofono', '/ril_0', interface, 'GetProperties')
    if isinstance(data, list) and len(data) == 1:
        data = data[0]
    return data if isinstance(data, dict) else {}


def selected(values, names):
    # GetProperties returns identifiers too: none are copied into the report.
    return {key: values[key].get('data') for key in names
            if key in values and isinstance(values[key], dict)}


def process_cpu_snapshot():
    """Return process names and CPU ticks without command lines or arguments."""
    result = {}
    for entry in Path('/proc').iterdir():
        if not entry.name.isdigit():
            continue
        try:
            raw = (entry / 'stat').read_text()
            # comm can contain spaces and parentheses; fields after the last )
            # start at field 3 (state), so utime/stime are offsets 11 and 12.
            fields = raw[raw.rfind(')') + 2:].split()
            name = raw[raw.find('(') + 1:raw.rfind(')')]
            result[int(entry.name)] = (name,
                                       int(fields[11]) + int(fields[12]))
        except (OSError, ValueError, IndexError):
            continue
    return result


def thread_cpu_snapshot():
    """Return thread names and CPU ticks; never collect command-line data."""
    result = {}
    for task in Path('/proc').glob('[0-9]*/task/[0-9]*'):
        try:
            raw = (task/'stat').read_text()
            fields = raw[raw.rfind(')') + 2:].split()
            name = raw[raw.find('(') + 1:raw.rfind(')')]
            result[(int(task.parent.parent.name), int(task.name))] = (
                name, int(fields[11]) + int(fields[12]))
        except (OSError, ValueError, IndexError):
            continue
    return result


def cpu_idle_snapshot():
    """Read per-CPU idle-state residency counters where the kernel exports them."""
    result = {}
    for cpu in sorted(Path('/sys/devices/system/cpu').glob('cpu[0-9]*')):
        if not cpu.name[3:].isdigit():
            continue
        states = []
        for state in sorted((cpu/'cpuidle').glob('state[0-9]*')):
            try:
                states.append({key:(state/key).read_text().strip() for key in
                               ('name','desc','usage','time','disable')
                               if (state/key).is_file()})
            except OSError:
                continue
        if states:
            result[cpu.name] = states
    return result


def cpu_frequency_snapshot():
    result = {}
    for cpu in sorted(Path('/sys/devices/system/cpu').glob('cpu[0-9]*')):
        if not cpu.name[3:].isdigit():
            continue
        path = cpu/'cpufreq'
        values = {}
        for key in ('scaling_governor','scaling_cur_freq','cpuinfo_cur_freq',
                    'scaling_min_freq','scaling_max_freq'):
            try:
                values[key] = (path/key).read_text().strip()
            except OSError:
                pass
        if values:
            result[cpu.name] = values
    return result


def thermal_readings():
    readings = {}
    for zone in Path('/sys/class/thermal').glob('thermal_zone*'):
        try:
            label = (zone / 'type').read_text().strip()
            value = int((zone / 'temp').read_text().strip())
            # Linux thermal sysfs normally reports millidegrees Celsius.
            readings[label or zone.name] = round(value / 1000, 1)
        except (OSError, ValueError):
            continue
    return readings


def power_supplies():
    result = {}
    for supply in Path('/sys/class/power_supply').glob('*'):
        data = {}
        for name in ('type', 'status', 'capacity', 'voltage_now',
                     'current_now', 'power_now', 'temp', 'online'):
            try:
                data[name] = (supply / name).read_text().strip()
            except OSError:
                pass
        if data:
            result[supply.name] = data
    return result


def wakeup_snapshot():
    """Read kernel wakeup counters; source labels are kernel driver names."""
    for path in (Path('/sys/kernel/debug/wakeup_sources'),
                 Path('/sys/kernel/tracing/wakeup_sources')):
        try:
            lines = path.read_text().splitlines()
        except OSError:
            continue
        if len(lines) < 2:
            continue
        headers = lines[0].split()
        active_index = next((i for i, value in enumerate(headers)
                             if value.lower() == 'active_count'), None)
        event_index = next((i for i, value in enumerate(headers)
                            if value.lower() == 'event_count'), None)
        if active_index is None or event_index is None:
            return None
        result = {}
        for line in lines[1:]:
            fields = line.split()
            try:
                if len(fields) <= max(active_index, event_index):
                    continue
                result[fields[0]] = {
                    'activeCount':int(fields[active_index]),
                    'eventCount':int(fields[event_index]),
                }
            except (ValueError, IndexError):
                continue
        return result
    return None


def counter_delta(before, after, keys):
    """Return positive counter deltas, tolerating counters reset at reboot."""
    result = {}
    for name, values in (after or {}).items():
        prior = (before or {}).get(name, {})
        delta = {key:max(0, int(values[key])-int(prior.get(key, values[key])))
                 for key in keys if key in values}
        if any(delta.values()):
            result[name] = delta
    return result


def inspect():
    native = 'tundra.rescue=1' in Path('/proc/cmdline').read_text()
    report = {'recordedAtUnix': time.time(), 'environment': 'tundra' if native else 'reference',
              'scope': 'Read-only readiness, not functional call/SMS/audio/camera qualification',
              'services': {}}
    for service in ('tundra-hal', 'tundra-wifi', 'NetworkManager', 'ofono', 'ModemManager'):
        report['services'][service] = (command(['systemctl', 'is-active', service]) or 'inactive').strip()
    modem = selected(properties('org.ofono.Modem'), ('Powered', 'Online', 'Interfaces'))
    sim = selected(properties('org.ofono.SimManager'), ('Present', 'PinRequired', 'LockedPins'))
    registration = selected(properties('org.ofono.NetworkRegistration'), ('Status', 'Technology', 'Strength'))
    interfaces = modem.get('Interfaces') or []
    report['telephony'] = {'modem': modem, 'sim': sim, 'registration': registration,
                           'voiceInterface': 'org.ofono.VoiceCallManager' in interfaces,
                           'smsInterface': 'org.ofono.MessageManager' in interfaces,
                           'outgoingOperationsPerformed': False}
    report['audio'] = {}
    # Run as the session user for access to their PulseAudio server.
    for kind in ('sinks', 'sources'):
        raw = command(['pactl', '--format=json', 'list', kind])
        try:
            items = json.loads(raw) if raw else None
            if items is None:
                short = command(['pactl', 'list', 'short', kind])
                items = [] if short is not None else None
                for line in (short or '').splitlines():
                    fields = line.split('\t')
                    if len(fields) >= 5:
                        items.append({'name': fields[1], 'state': fields[-1],
                                      'driver': fields[2]})
            if items is None:
                report['audio'][kind] = None
                continue
            report['audio'][kind] = [{key: item.get(key) for key in
                                     ('name', 'state', 'mute', 'active_port', 'driver')}
                                    for item in items]
        except (TypeError, ValueError):
            report['audio'][kind] = []
    report['camera'] = {
        'halProbeOnly': True,
        'applicationInstalled': Path('/usr/bin/droidian-camera').exists(),
        'captureTested': False,
    }
    report['battery'] = power_supplies()
    report['thermalCelsius'] = thermal_readings()
    backlights = {}
    for path in Path('/sys/class/backlight').glob('*/brightness'):
        try:
            backlights[path.parent.name] = (path.read_text().strip(),
                (path.parent / 'max_brightness').read_text().strip())
        except OSError:
            continue
    report['backlight'] = backlights
    before = process_cpu_snapshot()
    threads_before = thread_cpu_snapshot()
    idle_before = cpu_idle_snapshot()
    wakeups_before = wakeup_snapshot()
    battery_before = power_supplies()
    started = time.monotonic()
    time.sleep(5)
    elapsed = time.monotonic() - started
    after = process_cpu_snapshot()
    threads_after = thread_cpu_snapshot()
    ticks_per_second = os.sysconf('SC_CLK_TCK')
    cpu = []
    for pid, (name, ticks) in after.items():
        if pid not in before or before[pid][0] != name:
            continue
        delta = max(0, ticks - before[pid][1])
        percent = 100 * delta / ticks_per_second / elapsed
        if percent >= 1:
            cpu.append({'process': name, 'percentOfOneCore': round(percent, 1)})
    report['cpuSample'] = {'durationSeconds': round(elapsed, 2),
                           'busyProcesses': sorted(cpu,
                               key=lambda item: item['percentOfOneCore'],
                               reverse=True)[:12]}
    busy_threads = []
    for key, (name, ticks) in threads_after.items():
        prior = threads_before.get(key)
        if prior is None or prior[0] != name:
            continue
        percent = 100 * max(0, ticks-prior[1]) / ticks_per_second / elapsed
        if percent >= 1:
            busy_threads.append({'processId':key[0], 'threadId':key[1],
                                 'thread':name,
                                 'percentOfOneCore':round(percent,1)})
    report['cpuSample']['busyThreads'] = sorted(
        busy_threads, key=lambda item:item['percentOfOneCore'], reverse=True)[:20]
    idle_after = cpu_idle_snapshot()
    residency = {}
    for cpu_name, states in idle_after.items():
        previous = idle_before.get(cpu_name, [])
        before_by_name = {state.get('name'):state for state in previous}
        deltas = []
        for state in states:
            prior = before_by_name.get(state.get('name'), {})
            item = {'name':state.get('name')}
            for key in ('usage','time'):
                if key in state:
                    try:
                        item[key+'Delta'] = max(0,int(state[key])-int(prior.get(key,state[key])))
                    except ValueError:
                        pass
            if 'disable' in state:
                item['disabled'] = state['disable']
            deltas.append(item)
        if deltas:
            residency[cpu_name] = deltas
    report['cpuIdle'] = {'available':bool(residency),
                         'statesDuringSample':residency,
                         'frequencyAfterSample':cpu_frequency_snapshot(),
                         'kernelCommandLineFlags':[
                             token for token in Path('/proc/cmdline').read_text().split()
                             if token.startswith(('lpm_levels.','cpuidle.','nohz='))],
                         'lowPowerParameter':None,
                         'systemSuspend':{}}
    for key, path in {
        'sleepDisabled':'/sys/module/lpm_levels/parameters/sleep_disabled',
        'cpuIdleEnabled':'/sys/module/lpm_levels/parameters/enable_lpm',
        'powerState':'/sys/power/state',
        'memorySleep':'/sys/power/mem_sleep',
    }.items():
        try:
            value = Path(path).read_text().strip()
            if key in ('powerState','memorySleep'):
                report['cpuIdle']['systemSuspend'][key] = value
            else:
                report['cpuIdle']['lowPowerParameter'] = {
                    **(report['cpuIdle']['lowPowerParameter'] or {}), key:value}
        except OSError:
            pass
    suspend_stats = {}
    for path in Path('/sys/power/suspend_stats').glob('*'):
        if path.is_file():
            try:
                suspend_stats[path.name] = path.read_text().strip()
            except OSError:
                pass
    report['cpuIdle']['systemSuspend']['statistics'] = suspend_stats or None
    wakeups_after = wakeup_snapshot()
    report['wakeupSources'] = {
        'available':wakeups_after is not None,
        'eventDeltas':counter_delta(wakeups_before,wakeups_after,('eventCount',)),
    }
    report['batteryDelta'] = {}
    for supply, values in power_supplies().items():
        prior = battery_before.get(supply,{})
        for key in ('current_now','voltage_now','power_now','capacity','status'):
            if key in values:
                report['batteryDelta'].setdefault(supply,{})[key] = {
                    'before':prior.get(key), 'after':values[key]}
    return report


if __name__ == '__main__':
    print(json.dumps(inspect(), indent=2))