340 lines
14 KiB
Python
340 lines
14 KiB
Python
#!/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))
|