Compare commits

...
8 changed files with 710 additions and 0 deletions

No files matched your search

+5
View File
@@ -2,6 +2,11 @@
## Unreleased
- Add `scripts/frame-power-sample.sh`, a bounded read-only SSH sampler for
battery, CPU, named temperatures, fan and available power rails (issue #52).
JSON samples/summaries and CSV exports retain explicit unavailable states;
active smoke-test locks stop measurement. No power settings are changed.
- Show saved Android compatibility reports on store search cards and app details.
Reports refresh in the background, so a stalled database or credential lookup
cannot delay search. Warnings allow installation; build compatibility blockers
+3
View File
@@ -44,6 +44,9 @@ Live video of what the lenses show, or of the app panel in use, flat and still h
**🔋 Battery and status**<br>
Charge, charging watts and time left, storage, memory, temperature, Wi-Fi, and what's running.
For repeatable read-only battery, power-rail, temperature and fan measurements,
use the [power and heat sampler](docs/power-and-heat.md).
</td>
</tr>
<tr>
+149
View File
@@ -0,0 +1,149 @@
{
"date": "2026-10-24",
"osVersion": "0.5.6",
"osBuild": "20261015.6141207",
"commands": [
"scripts/frame-power-sample.sh --count 3 --interval 1",
"scripts/frame-power-sample.sh --count 1 --csv"
],
"jsonSampleCount": 3,
"elapsedSeconds": [
0.078011,
1.088542,
2.107291
],
"stopped": null,
"unavailableGroups": [
[
"proximity"
],
[
"proximity"
],
[
"proximity"
]
],
"selectedSummary": {
"battery.percent": {
"unit": "%",
"sources": [
"/sys/class/power_supply/max1720x_bat_7-36"
],
"available": 3,
"unavailable": 0,
"min": 99,
"max": 99,
"mean": 99.0,
"first": 99,
"last": 99
},
"battery.status": {
"unit": "state",
"sources": [
"/sys/class/power_supply/max1720x_bat_7-36"
],
"available": 3,
"unavailable": 0,
"states": [
"Not charging"
]
},
"battery.tempC": {
"unit": "C",
"sources": [
"/sys/class/power_supply/max1720x_bat_7-36"
],
"available": 3,
"unavailable": 0,
"min": 26.400000000000002,
"max": 26.400000000000002,
"mean": 26.4,
"first": 26.400000000000002,
"last": 26.400000000000002
},
"battery.watts": {
"unit": "W",
"sources": [
"/sys/class/power_supply/max1720x_bat_7-36"
],
"available": 3,
"unavailable": 0,
"min": -0.12,
"max": -0.02,
"mean": -0.07,
"first": -0.02,
"last": -0.07
},
"cpu.percent": {
"unit": "%",
"sources": [
"/proc/stat"
],
"available": 2,
"unavailable": 1,
"min": 8.7,
"max": 11.0,
"mean": 9.85,
"first": 11.0,
"last": 8.7
},
"fan.pwmRaw": {
"unit": "raw",
"sources": [
"/sys/class/hwmon/hwmon50/pwm1"
],
"available": 3,
"unavailable": 0,
"min": 41,
"max": 41,
"mean": 41.0,
"first": 41,
"last": 41
},
"fan.slg4ax46073v.fan1_input": {
"unit": "RPM",
"sources": [
"/sys/class/hwmon/hwmon50/fan1_input"
],
"available": 3,
"unavailable": 0,
"min": 8624,
"max": 8638,
"mean": 8628.666667,
"first": 8638,
"last": 8624
},
"proximity.raw": {
"unit": "raw",
"sources": [
"/sys/bus/iio/devices/iio:device2/in_proximity_raw"
],
"available": 0,
"unavailable": 3,
"states": []
},
"rail.max34417_9_10.vph.power1": {
"unit": "W",
"sources": [
"/sys/class/hwmon/hwmon53/power1_input"
],
"available": 3,
"unavailable": 0,
"min": 4.656299,
"max": 4.804066,
"mean": 4.705555,
"first": 4.656299,
"last": 4.804066
}
},
"csvRows": 92,
"csvSummaryMetrics": 91,
"csvHasExplicitUnavailable": true,
"limits": [
"Read-only short transport/parser check; no controlled baseline or worn/load/battery experiment.",
"Proximity unavailable; numeric thermistors are not proof of sensor accuracy.",
"No settings, services, device files or application state changed; no remote installation.",
"No smoke lock at preflight or sampler checks; advisory checks cannot exclude a concurrent lock appearing during a read."
]
}
+122
View File
@@ -0,0 +1,122 @@
# Read-only power and heat measurements
Issue [#52](https://github.com/saphid/frame-control/issues/52) starts with
measurement. `scripts/frame-power-sample.sh` is a bounded, on-demand sampler
over the existing `ssh frame` alias. It needs Python 3 on both computers and
an already paired, awake, reachable Frame. It uses strict SSH host-key checking
and batch mode; it never pairs, prompts for a password or changes SSH security.
```sh
# Twelve samples, five seconds apart (about 55 seconds), JSON report:
scripts/frame-power-sample.sh > power.json
# Five-minute observation: samples at 0, 5, ... 300 seconds:
scripts/frame-power-sample.sh --count 61 --interval 5 > baseline.json
# Long-form CSV samples; JSON summary goes separately to stderr:
scripts/frame-power-sample.sh --count 12 --interval 5 --csv > power.csv 2> summary.json
```
Count is limited to 1–720; interval to finite values from 1–60 seconds. These
are target start-to-start intervals, not guarantees: read latency can delay a
sample. Wall-clock `time` and monotonic `elapsedSeconds` record actual timing.
The first sample has no CPU utilization: two valid `/proc/stat` observations
are required. CPU utilization then covers the interval between observations,
averaged across the machine, rather than the utilization of one process.
## What it reads
- Battery percentage, driver charging state, signed battery watts and battery
temperature, using the same parser as Frame Control's battery details.
Positive battery watts mean charging on the Frame fuel gauge; negative means
discharge. `Not charging` alone does not prove a charger is connected.
- CPU utilization from `/proc/stat`; GPU clock from the known Frame devfreq path.
- Named thermal zones in °C, with zone names retained. A thermal zone is not
automatically a CPU sensor; there is no inferred “CPU temperature” or safe
temperature threshold.
- hwmon power channels, labels and driver names in watts. `vph` is the measured
system rail, not power at the wall. CPU `apc*` and GPU `gfx` rails are separate
readings; do not sum all channels as if they were independent power consumers.
- Fan RPM and the Frame fan driver's raw `pwm1`. No generic percentage conversion
is assumed. The historical Frame driver reported a percent-like value of 41.
- Named IIO board temperatures, PM8550B input/charge current in amps, and raw
VCNL4000 proximity if the account can read it. Raw proximity has no inferred
worn/unworn threshold.
Each metric includes its unit, source path and availability state. Missing,
unreadable or invalid values are JSON `null`, not zero. A missing sensor family
is also listed in each sample's `unavailableGroups`; CSV adds unavailable group
rows. Numeric zero remains a valid reading. The OS build is recorded per sample
when `/etc/os-release` is readable; hostnames, addresses, machine IDs, process
lists, logs, screenshots and application inventories are not collected.
The JSON report contains raw samples and a summary per metric: available and
unavailable counts, numeric minimum/maximum/arithmetic mean/first/last, or the
observed charging states. Means exclude unavailable readings. They are sample
means, not time-weighted energy or battery-runtime estimates. Different sample
availability and intervals must be considered when comparing sessions.
Numeric validity does not establish sensor accuracy. Historical readings
already showed an apparently absent/broken right-display thermistor near
−16 °C. This firmware also reports one charger-lite die sensor at −60 °C.
Those raw numbers remain visible for diagnosis; neither should be used to
infer cooling performance or safety. This tool does not diagnose heat, explain
a spike, predict battery health, or promise power savings.
## Safety and shared-device use
Only file reads under `/proc`, `/sys` and `/etc/os-release` run on the Frame.
The Python sources are sent on SSH stdin and not installed on the headset.
There is no sudo, OpenVR initialization, service call, settings write, process
control, journal query, wake, suspend or reboot. No daemon, telemetry upload or
automatic measurement is added. Redirects above write only to your computer.
The observer itself uses a little CPU and SSH traffic, so label baseline
conditions and avoid interpreting very small changes as savings.
The sampler checks `/tmp/frame-test.lock` on the computer before SSH, and on
the Frame before any probe and before each sample. If a smoke session is active,
it exits 75. A lock appearing mid-run stops the next sample, retains earlier
samples, sets `stopped`, and exits 75. It never removes or acquires another
session's lock. This is an advisory check, not an atomic lease: a smoke session
can start during a file read. Coordinate sessions and defer measurement when
timing overlaps. It does not block someone else from starting a smoke test.
SSH failures/timeouts return a nonzero exit and an explanation, without
presenting a zero or an old report as a successful measurement. Fully missing
sensors can still produce a successful report with explicit unavailable states.
On a connection failure, partial samples buffered on the headset can be lost.
## Evidence and remaining work
Read-only checks on **2026-10-24**, SteamOS **0.5.6**, build
**20261015.6141207**, confirmed battery, CPU ticks, thermal zones, rail power,
fan, GPU clock, board temperatures and charger current through the final
sampler. Proximity was unavailable. Both JSON and CSV paths were exercised;
see [the recorded results](evidence/power-sampler-2026-10-24.json).
No smoke lock was present at the preflight or sampler checks. No settings,
services, application state or device files were changed. This short run is
a parser/transport check, not a controlled baseline or a worn-headset test.
The original planning document is in commit `fe132e3` (2026-09-29). It recorded
an off-head, charging idle baseline of about 4.71 W, CPU zones around 40–45 °C,
and about 8,350 fan RPM on build 20260925.6191901. Those conditions and firmware
differ from the current check; comparisons do not establish a regression.
This change completes only the sampler portion of “Plan: leave it on, but
cheaply.” Controlled worn/home-space, streaming/game load and unplugged
baselines, and the cause of the 2026-09-28 95 °C spike, remain uninvestigated.
Quiet mode, compositor changes, CPU/fan/charging changes and remote-wake tests
are separate work. No controls are offered here. Follow the shared-device
procedure in [testing.md](testing.md#headset-smoke-test) for future experiments.
Offline validation:
```sh
python3 -m unittest discover -s tests -p test_power_sample.py -v
bash -n scripts/frame-power-sample.sh
python3 -m py_compile scripts/frame_power_sample.py
```
Fixtures cover units, signed and zero values, missing/malformed/permission-denied
readings, named/discovered channels, unknown charging state, CPU warmup,
summary counts, CSV quoting, strict SSH transport and smoke-lock interruption.
+6
View File
@@ -12,6 +12,12 @@ A fourth, a SteamOS VM, may come later ([issue #6](https://github.com/saphid/ste
## Unit tests
The read-only power/heat sampler has fixture and transport tests in
`tests/test_power_sample.py`. Run
`python3 -m unittest discover -s tests -p test_power_sample.py -v`.
See [power-and-heat.md](power-and-heat.md) for sampling, smoke-lock behavior and
real-device verification limits.
```sh
python3 -m unittest discover -s tests
```
+5
View File
@@ -0,0 +1,5 @@
#!/usr/bin/env bash
# Read-only, bounded power/heat sampler. See docs/power-and-heat.md.
set -euo pipefail
root=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)
exec python3 "$root/scripts/frame_power_sample.py" "$@"
+211
View File
@@ -0,0 +1,211 @@
"""Bounded read-only Frame measurements; no device files are written.
The host sends this and frame_status.py over one trusted SSH connection. Only
battery() and CPU tick helpers are reused: no OpenVR, services or analytics.
"""
import argparse
import csv
import glob
import io
import json
import math
from pathlib import Path
import subprocess
import sys
import time
LOCK = Path('/tmp/frame-test.lock')
UNAVAILABLE = 'unavailable: not reported, unreadable or invalid'
def read(path):
try:
return Path(path).read_text().strip()
except (OSError, UnicodeError):
return None
def number(path, scale=1, minimum=None, maximum=None):
try:
value = int(read(path)) * scale
if not math.isfinite(value):
return None
if minimum is not None and value < minimum or maximum is not None and value > maximum:
return None
return round(value, 6)
except (TypeError, ValueError, OverflowError):
return None
def snapshot(status, before):
if LOCK.exists():
raise RuntimeError('Smoke session active (/tmp/frame-test.lock); no sensors read.')
ticks = status.cpu_ticks()
readings = {}
def add(key, value, unit, source):
readings[key] = {'value': value, 'unit': unit, 'source': source,
'state': 'available' if value is not None else UNAVAILABLE}
battery = status.battery() or {}
for field, unit in (('percent', '%'), ('status', 'state'), ('watts', 'W'),
('tempC', 'C')):
value = battery.get(field)
if field == 'status' and value not in ('Charging', 'Discharging', 'Not charging', 'Full'):
value = None
add('battery.' + field, value, unit, battery.get('source'))
add('cpu.percent', status.cpu_percent(before, ticks), '%', '/proc/stat')
if before is None:
readings['cpu.percent']['state'] = 'unavailable: needs two valid CPU tick samples'
add('gpu.clock', number('/sys/class/devfreq/3d00000.gpu/cur_freq', 1e-6, 0),
'MHz', '/sys/class/devfreq/3d00000.gpu/cur_freq')
# Discover by driver name/label, never assume a hwmon index. Keep source
# paths and channel numbers to distinguish duplicate labels.
for node in sorted(glob.glob('/sys/class/hwmon/hwmon*')):
name = read(node + '/name') or Path(node).name
for path in sorted(glob.glob(node + '/power*_input')):
channel = Path(path).name[:-6]
label = read(node + '/' + channel + '_label') or channel
add(f'rail.{name}.{label}.{channel}', number(path, 1e-6, 0), 'W', path)
for path in sorted(glob.glob(node + '/fan*_input')):
add(f'fan.{name}.{Path(path).stem}', number(path, minimum=0), 'RPM', path)
# pwm1 on Frame's driver is percent; leave raw to avoid assuming the
# generic hwmon 0..255 scale on other drivers or later firmware.
if name == 'slg4ax46073v':
add('fan.pwmRaw', number(node + '/pwm1', minimum=0), 'raw', node + '/pwm1')
for node in sorted(glob.glob('/sys/class/thermal/thermal_zone*')):
name = read(node + '/type') or Path(node).name
add(f'thermal.{name}.{Path(node).name}', number(node + '/temp', 0.001, -273.15, 200),
'C', node + '/temp')
for node in sorted(glob.glob('/sys/bus/iio/devices/iio:device*')):
for path in sorted(glob.glob(node + '/in_temp_*_input')):
add(f'board.{Path(node).name}.{Path(path).stem}', number(path, 0.001, -273.15, 200),
'C', path)
for field in ('iin', 'ichg'):
path = node + '/in_current_pm8550b_' + field + '_fb_input'
# Known names, but no fixed IIO index; absent channels are covered
# by group availability below.
if Path(path).exists():
add(f'charger.{Path(node).name}.{field}', number(path, 1e-6), 'A', path)
if read(node + '/name') == 'vcnl4000':
add('proximity.raw', number(node + '/in_proximity_raw', minimum=0),
'raw', node + '/in_proximity_raw')
unavailable = [group for group in ('rail', 'fan', 'thermal', 'board', 'charger', 'proximity')
if not any(k.startswith(group + '.') and v['value'] is not None
for k, v in readings.items())]
os_info = dict(line.split('=', 1) for line in (read('/etc/os-release') or '').splitlines()
if '=' in line)
return {'time': time.time(), 'osBuild': os_info.get('BUILD_ID', '').strip('"') or None,
'readings': readings, 'unavailableGroups': unavailable}, ticks
def summarize(samples):
result = {}
keys = sorted({key for sample in samples for key in sample['readings']})
for key in keys:
rows = [s['readings'][key] for s in samples if key in s['readings']]
values = [r['value'] for r in rows if r['value'] is not None]
item = {'unit': rows[0]['unit'], 'sources': sorted({r['source'] for r in rows if r['source']}),
'available': len(values), 'unavailable': len(samples) - len(values)}
if values and all(isinstance(v, (int, float)) for v in values):
item.update(min=min(values), max=max(values), mean=round(sum(values) / len(values), 6),
first=values[0], last=values[-1])
else:
item['states'] = sorted(set(values))
result[key] = item
return result
def collect(status, count, interval):
samples, before = [], None
started = time.monotonic()
stopped = None
for index in range(count):
if index:
time.sleep(max(0, started + index * interval - time.monotonic()))
try:
sample, before = snapshot(status, before)
except RuntimeError as exc:
stopped = str(exc)
break
sample['elapsedSeconds'] = round(time.monotonic() - started, 6)
samples.append(sample)
return {'schemaVersion': 1, 'requestedSamples': count, 'intervalSeconds': interval,
'stopped': stopped, 'samples': samples, 'summary': summarize(samples)}
def csv_samples(report):
output = io.StringIO(newline='')
writer = csv.writer(output)
writer.writerow(('time', 'elapsedSeconds', 'metric', 'value', 'unit', 'source', 'state'))
for sample in report['samples']:
for key, row in sample['readings'].items():
# csv quoting does not prevent spreadsheet formulas in labels.
def cell(value):
return "'" + value if isinstance(value, str) and value.startswith(('=', '+', '-', '@')) else value
writer.writerow([cell(v) for v in (sample['time'], sample['elapsedSeconds'], key,
row['value'], row['unit'], row['source'], row['state'])])
for group in sample['unavailableGroups']:
writer.writerow((sample['time'], sample['elapsedSeconds'], group, '', '', '', UNAVAILABLE))
return output.getvalue()
def parser():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument('--count', type=int, default=12, help='samples, 1..720 (default 12)')
ap.add_argument('--interval', type=float, default=5, help='seconds, 1..60 (default 5)')
ap.add_argument('--csv', action='store_true', help='CSV samples on stdout; JSON summary on stderr')
ap.add_argument('--probe', action='store_true', help=argparse.SUPPRESS)
return ap
def remote_payload():
root = Path(__file__).resolve().parents[1]
source = (root / 'ui/frame_status.py').read_text()
# Executing as a module prevents frame_status.main() from probing unrelated
# status. Nothing is installed, cached or logged on the headset.
return ('import types, sys\n'
'status = types.ModuleType("frame_status")\n'
f'exec(compile({source!r}, "frame_status.py", "exec"), status.__dict__)\n'
'sys.modules["frame_status"] = status\n' + Path(__file__).read_text())
def main(argv=None):
ap = parser()
args = ap.parse_args(argv)
if not 1 <= args.count <= 720 or not math.isfinite(args.interval) or not 1 <= args.interval <= 60:
ap.error('count must be 1..720 and interval must be finite, 1..60 seconds')
# Also honor a coordinator lock on the computer before opening SSH.
if LOCK.exists():
print('Smoke session active (/tmp/frame-test.lock); no SSH or sensors accessed.', file=sys.stderr)
return 75
if args.probe:
import frame_status
report = collect(frame_status, args.count, args.interval)
else:
command = ['ssh', '-T', '-o', 'BatchMode=yes', '-o', 'StrictHostKeyChecking=yes',
'-o', 'ConnectTimeout=5', '-o', 'ServerAliveInterval=5',
'-o', 'ServerAliveCountMax=2', 'frame',
'python3', '-', '--probe', '--count', str(args.count), '--interval', str(args.interval)]
try:
run = subprocess.run(command, input=remote_payload(), text=True, capture_output=True,
timeout=args.count * args.interval + 30)
if run.returncode:
print('Read-only sample failed: ' + (run.stderr.strip() or f'SSH exit {run.returncode}'),
file=sys.stderr)
return run.returncode
report = json.loads(run.stdout)
except (OSError, subprocess.TimeoutExpired, ValueError) as exc:
print(f'Read-only sample unavailable: {exc}', file=sys.stderr)
return 1
if args.csv:
sys.stdout.write(csv_samples(report))
print(json.dumps({k: v for k, v in report.items() if k != 'samples'}, allow_nan=False), file=sys.stderr)
else:
print(json.dumps(report, allow_nan=False))
return 75 if report['stopped'] else 0
if __name__ == '__main__':
sys.exit(main())
+209
View File
@@ -0,0 +1,209 @@
"""Power sampler fixtures and SSH transport contract; no real device contact."""
import sandbox # noqa: F401
import contextlib
import csv
import importlib.util
import io
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('power_sample', ROOT / 'scripts/frame_power_sample.py')
power = importlib.util.module_from_spec(spec)
spec.loader.exec_module(power)
sys.path.insert(0, str(ROOT / 'ui'))
import frame_status as status
class Sampler(unittest.TestCase):
def patch(self, obj, name, **kwargs):
p = mock.patch.object(obj, name, **kwargs)
result = p.start()
self.addCleanup(p.stop)
return result
def fixture(self, files):
# Windows cannot create a directory named iio:device8. Keep the
# production sysfs spelling in returned paths while using a portable
# name inside this temporary fixture tree.
def local_path(path):
return self.root / str(path).replace('\\', '/').lstrip('/').replace(':', '__COLON__')
for path, value in files.items():
node = local_path(path)
node.parent.mkdir(parents=True, exist_ok=True)
node.write_text(str(value))
actual_glob = power.glob.glob
def paths(pattern):
return ['/' + Path(p).relative_to(self.root).as_posix().replace('__COLON__', ':')
for p in actual_glob(str(local_path(pattern)))]
actual_read = power.read
self.patch(power.glob, 'glob', side_effect=paths)
self.patch(power, 'read', side_effect=lambda p: actual_read(str(local_path(str(p)))))
self.patch(status, 'read', side_effect=lambda p: actual_read(str(local_path(str(p)))))
return local_path
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name)
p = mock.patch.object(power, 'LOCK', self.root / 'lock')
p.start()
self.addCleanup(p.stop)
def test_real_file_units_zero_signed_battery_and_named_sensors(self):
h = '/sys/class/hwmon/hwmon88'
b = '/sys/class/power_supply/max1720x_bat_7-36'
self.fixture({h + '/name': 'max34417_9_10', h + '/power1_label': 'vph',
h + '/power1_input': '4710000',
'/sys/class/hwmon/hwmon3/name': 'slg4ax46073v',
'/sys/class/hwmon/hwmon3/fan1_input': '0',
'/sys/class/hwmon/hwmon3/pwm1': '41',
'/sys/class/thermal/thermal_zone9/type': 'cpu3_top_thermal',
'/sys/class/thermal/thermal_zone9/temp': '45321',
b + '/type': 'Battery', b + '/capacity': '0', b + '/status': 'Discharging',
b + '/voltage_now': '3800000', b + '/current_now': '-1000000',
'/proc/stat': 'cpu 10 0 0 90 0 0 0 0'})
sample, ticks = power.snapshot(status, (80, 80))
r = sample['readings']
self.assertEqual(r['rail.max34417_9_10.vph.power1']['value'], 4.71)
self.assertEqual(r['battery.percent']['value'], 0)
self.assertEqual(r['battery.watts']['value'], -3.8)
self.assertEqual(r['battery.status']['value'], 'Discharging')
self.assertEqual(r['fan.slg4ax46073v.fan1_input']['value'], 0)
self.assertEqual(r['fan.pwmRaw']['unit'], 'raw')
self.assertEqual(r['thermal.cpu3_top_thermal.thermal_zone9']['value'], 45.321)
self.assertEqual(r['cpu.percent']['value'], 50)
self.assertEqual(ticks, (100, 90))
self.assertEqual(sample['unavailableGroups'], ['board', 'charger', 'proximity'])
def test_missing_and_invalid_values_remain_unavailable(self):
self.fixture({'/sys/class/thermal/thermal_zone0/temp': '999999999',
'/sys/class/hwmon/hwmon2/name': 'fan',
'/sys/class/hwmon/hwmon2/fan1_input': '-1'})
sample, ticks = power.snapshot(status, None)
self.assertIsNone(ticks)
self.assertTrue(all(r['value'] is None for r in sample['readings'].values()))
self.assertEqual(len(sample['unavailableGroups']), 6)
self.assertIn('two valid CPU', sample['readings']['cpu.percent']['state'])
def test_permission_error_is_unavailable(self):
with mock.patch.object(Path, 'read_text', side_effect=PermissionError):
self.assertIsNone(power.number('/unreadable'))
for value in ('NaN', 'Infinity', '1.5', '', '9' * 5000):
with mock.patch.object(power, 'read', return_value=value):
self.assertIsNone(power.number('/invalid'))
def test_iio_units_and_permission_denied_proximity(self):
node = '/sys/bus/iio/devices/iio:device8'
local_path = self.fixture({node + '/name': 'vcnl4000',
node + '/in_temp_pcba_at_wifi_input': '37500',
node + '/in_current_pm8550b_iin_fb_input': '241671',
'/etc/os-release': 'BUILD_ID=20261015.6141207\n'})
# exists() is used only for optional known charger channels and lock.
exists = Path.exists
with mock.patch.object(Path, 'exists', lambda p: exists(p) if p == power.LOCK
else exists(local_path(str(p)))):
sample, _ = power.snapshot(status, None)
self.assertEqual(sample['osBuild'], '20261015.6141207')
self.assertEqual(sample['readings']['board.iio:device8.in_temp_pcba_at_wifi_input']['value'], 37.5)
self.assertEqual(sample['readings']['charger.iio:device8.iin']['value'], 0.241671)
self.assertIsNone(sample['readings']['proximity.raw']['value'])
self.assertIn('proximity', sample['unavailableGroups'])
def test_unknown_charging_state_is_not_claimed(self):
self.fixture({})
with mock.patch.object(status, 'battery', return_value={'status': 'Unknown', 'percent': 50}):
r = power.snapshot(status, None)[0]['readings']
self.assertIsNone(r['battery.status']['value'])
self.assertEqual(r['battery.percent']['value'], 50)
def test_smoke_lock_prevents_any_sensor_reads(self):
power.LOCK.mkdir()
with mock.patch.object(status, 'cpu_ticks') as ticks:
with self.assertRaisesRegex(RuntimeError, 'Smoke session'):
power.snapshot(status, None)
ticks.assert_not_called()
def test_smoke_lock_appearing_between_samples_stops_and_preserves_partial_report(self):
self.fixture({'/proc/stat': 'cpu 10 0 0 90 0 0 0 0'})
with mock.patch.object(power.time, 'sleep', side_effect=lambda _: power.LOCK.mkdir()):
report = power.collect(status, 3, 1)
self.assertEqual(len(report['samples']), 1)
self.assertIn('Smoke session', report['stopped'])
self.assertEqual(report['requestedSamples'], 3)
def test_summaries_ignore_missing_values_and_keep_state_changes(self):
def row(value, unit='W'):
return {'value': value, 'unit': unit, 'source': '/sensor', 'state': 'available'}
samples = [{'readings': {'rail': row(0), 'state': row('Charging', 'state')}},
{'readings': {'rail': row(None), 'state': row('Full', 'state')}},
{'readings': {'rail': row(6)}}]
summary = power.summarize(samples)
self.assertEqual(summary['rail'], dict(unit='W', sources=['/sensor'], available=2,
unavailable=1, min=0, max=6, mean=3, first=0, last=6))
self.assertEqual(summary['state']['states'], ['Charging', 'Full'])
self.assertEqual(summary['state']['unavailable'], 1)
def test_csv_quoting_and_unavailable_groups(self):
report = {'samples': [{'time': 1, 'elapsedSeconds': 0,
'readings': {'fan,test': {'value': -3.8, 'unit': 'W',
'source': '=formula', 'state': 'available'}},
'unavailableGroups': ['thermal']}]}
rows = list(csv.reader(io.StringIO(power.csv_samples(report))))
self.assertEqual(rows[1][2:6], ['fan,test', '-3.8', 'W', "'=formula"])
self.assertEqual(rows[2][-1], power.UNAVAILABLE)
def test_bounded_arguments_rejected_before_ssh(self):
for args in (['--count', '0'], ['--count', '721'], ['--interval', 'nan'],
['--interval', 'inf'], ['--interval', '0.5'], ['--interval', '61']):
with mock.patch.object(power.subprocess, 'run') as run, contextlib.redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit):
power.main(args)
run.assert_not_called()
def test_local_lock_prevents_ssh(self):
power.LOCK.mkdir()
with mock.patch.object(power.subprocess, 'run') as run, contextlib.redirect_stderr(io.StringIO()):
self.assertEqual(power.main([]), 75)
run.assert_not_called()
def test_ssh_payload_runs_as_module_and_outputs_report(self):
report = dict(samples=[], summary={}, stopped=None)
def ssh(command, **kwargs):
self.assertIn('BatchMode=yes', command)
self.assertIn('StrictHostKeyChecking=yes', command)
self.assertIn('frame', command)
self.assertEqual(command[-4:], ['--count', '2', '--interval', '1.0'])
# Execute the exact transmitted source, with main suppressed. Its
# frame_status module must be importable without running status().
namespace = {'__name__': 'payload_test'}
with mock.patch.dict(sys.modules):
exec(compile(kwargs['input'], '<ssh-stdin>', 'exec'), namespace)
remote = sys.modules['frame_status']
self.assertTrue(callable(remote.battery))
with mock.patch.object(remote, 'status', side_effect=AssertionError('full status forbidden')), \
mock.patch.object(remote, 'battery', return_value=None), \
mock.patch.object(remote, 'cpu_ticks', return_value=(100, 90)), \
mock.patch.object(namespace['glob'], 'glob', return_value=[]), \
mock.patch.object(namespace['Path'], 'exists', return_value=False):
report.update(namespace['collect'](remote, 1, 1))
return subprocess.CompletedProcess(command, 0, json.dumps(report), '')
output = io.StringIO()
with mock.patch.object(power.subprocess, 'run', side_effect=ssh), contextlib.redirect_stdout(output):
self.assertEqual(power.main(['--count', '2', '--interval', '1']), 0)
self.assertEqual(len(json.loads(output.getvalue())['samples']), 1)
def test_ssh_failures_are_not_measurements(self):
for result in (subprocess.CompletedProcess([], 255, '', 'offline'),
subprocess.CompletedProcess([], 0, 'bad json', '')):
output = io.StringIO()
with mock.patch.object(power.subprocess, 'run', return_value=result), \
contextlib.redirect_stdout(output), contextlib.redirect_stderr(io.StringIO()):
self.assertNotEqual(power.main([]), 0)
self.assertEqual(output.getvalue(), '')