220 lines
8.6 KiB
Python
220 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only Philips TPM191E EPG health check.
|
|
|
|
ADB mode runs only `dumpsys dbinfo` and `df`. Snapshot mode opens an existing
|
|
SQLite file with mode=ro&immutable=1. It contains no repair code.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
MIB = 1024 * 1024
|
|
OK_MAX_BYTES = 64 * MIB
|
|
WARN_MAX_BYTES = 256 * MIB
|
|
WARN_ROWS = 100_000
|
|
MAINTENANCE_ROWS = 250_000
|
|
WARN_OLDEST_DAYS = 14
|
|
MAINTENANCE_OLDEST_DAYS = 30
|
|
|
|
|
|
def run_readonly_adb(adb: Path, serial: str, args: list[str]) -> str:
|
|
allowed = {
|
|
("shell", "dumpsys", "dbinfo", "org.droidtv.channels", "-v"),
|
|
("shell", "df", "-k", "/data"),
|
|
}
|
|
if tuple(args) not in allowed:
|
|
raise RuntimeError(f"internal safety check rejected ADB command: {args!r}")
|
|
completed = subprocess.run(
|
|
[str(adb), "-s", serial, *args],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
output = completed.stdout + completed.stderr
|
|
if completed.returncode != 0 or "device offline" in output.lower():
|
|
raise RuntimeError(output.strip() or f"adb exited {completed.returncode}")
|
|
return output
|
|
|
|
|
|
def parse_dbinfo(text: str) -> dict[str, Any]:
|
|
match = re.search(r"^\s*EpgDatabase\.db\s+(\d+)b\b", text, re.MULTILINE)
|
|
if not match:
|
|
raise RuntimeError("EpgDatabase.db size not found in dumpsys dbinfo")
|
|
return {"db_bytes": int(match.group(1)), "source": "adb-dumpsys-dbinfo"}
|
|
|
|
|
|
def parse_df(text: str) -> dict[str, Any]:
|
|
for line in reversed(text.splitlines()):
|
|
fields = line.split()
|
|
# Android's `df /data` may report the process-visible bind mount as
|
|
# `/data/user/0`; it is backed by the same userdata filesystem.
|
|
if len(fields) >= 6 and fields[-1].startswith("/data"):
|
|
try:
|
|
total_kib, used_kib, free_kib = map(int, fields[1:4])
|
|
return {
|
|
"data_total_kib": total_kib,
|
|
"data_used_kib": used_kib,
|
|
"data_free_kib": free_kib,
|
|
"data_use_percent": fields[4],
|
|
}
|
|
except ValueError:
|
|
pass
|
|
return {"df_note": "Could not parse /data line"}
|
|
|
|
|
|
def snapshot_metrics(path: Path, now_epoch: int) -> dict[str, Any]:
|
|
resolved = path.resolve(strict=True)
|
|
uri = resolved.as_uri() + "?mode=ro&immutable=1"
|
|
connection = sqlite3.connect(uri, uri=True, timeout=5)
|
|
try:
|
|
connection.execute("PRAGMA query_only=ON")
|
|
table = connection.execute(
|
|
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='scheduletable'"
|
|
).fetchone()
|
|
if table is None:
|
|
raise RuntimeError("snapshot has no scheduletable")
|
|
# Philips data has historically used epoch seconds; the CASE also
|
|
# handles accidental millisecond values without changing the database.
|
|
epoch = "CASE WHEN ABS(endtime) >= 100000000000 THEN endtime / 1000 ELSE endtime END"
|
|
count, oldest, newest, older_30 = connection.execute(
|
|
f"SELECT COUNT(*), MIN({epoch}), MAX({epoch}), "
|
|
f"SUM(CASE WHEN {epoch} < ? THEN 1 ELSE 0 END) FROM scheduletable",
|
|
(now_epoch - MAINTENANCE_OLDEST_DAYS * 86400,),
|
|
).fetchone()
|
|
page_size = int(connection.execute("PRAGMA page_size").fetchone()[0])
|
|
page_count = int(connection.execute("PRAGMA page_count").fetchone()[0])
|
|
freelist = int(connection.execute("PRAGMA freelist_count").fetchone()[0])
|
|
integrity = connection.execute("PRAGMA quick_check(1)").fetchone()[0]
|
|
finally:
|
|
connection.close()
|
|
oldest_days = None if oldest is None else max(0.0, (now_epoch - int(oldest)) / 86400)
|
|
return {
|
|
"source": "offline-sqlite-snapshot",
|
|
"snapshot": str(resolved),
|
|
"db_bytes": resolved.stat().st_size,
|
|
"schedule_rows": int(count),
|
|
"oldest_end_epoch": None if oldest is None else int(oldest),
|
|
"newest_end_epoch": None if newest is None else int(newest),
|
|
"oldest_age_days": oldest_days,
|
|
"rows_older_than_30_days": int(older_30 or 0),
|
|
"page_size": page_size,
|
|
"page_count": page_count,
|
|
"freelist_count": freelist,
|
|
"quick_check": integrity,
|
|
}
|
|
|
|
|
|
def assess(metrics: dict[str, Any]) -> tuple[str, list[str]]:
|
|
size = int(metrics["db_bytes"])
|
|
reasons: list[str] = []
|
|
level = "OK"
|
|
|
|
if size >= WARN_MAX_BYTES:
|
|
level = "MAINTENANCE_RECOMMENDED"
|
|
reasons.append(f"DB is at least {WARN_MAX_BYTES // MIB} MiB")
|
|
elif size >= OK_MAX_BYTES:
|
|
level = "WARN"
|
|
reasons.append(f"DB is at least {OK_MAX_BYTES // MIB} MiB")
|
|
|
|
if "schedule_rows" in metrics:
|
|
rows = int(metrics["schedule_rows"])
|
|
age = metrics.get("oldest_age_days")
|
|
if rows >= MAINTENANCE_ROWS:
|
|
level = "MAINTENANCE_RECOMMENDED"
|
|
reasons.append(f"schedule rows >= {MAINTENANCE_ROWS:,}")
|
|
elif rows >= WARN_ROWS and level == "OK":
|
|
level = "WARN"
|
|
reasons.append(f"schedule rows >= {WARN_ROWS:,}")
|
|
if age is not None and age >= MAINTENANCE_OLDEST_DAYS and size >= OK_MAX_BYTES:
|
|
level = "MAINTENANCE_RECOMMENDED"
|
|
reasons.append(
|
|
f"oldest entry >= {MAINTENANCE_OLDEST_DAYS} days and DB >= 64 MiB"
|
|
)
|
|
elif age is not None and age >= WARN_OLDEST_DAYS and level == "OK":
|
|
level = "WARN"
|
|
reasons.append(f"oldest entry >= {WARN_OLDEST_DAYS} days")
|
|
if metrics.get("quick_check") != "ok":
|
|
level = "WARN"
|
|
reasons.append(f"SQLite quick_check={metrics.get('quick_check')!r}")
|
|
|
|
if not reasons:
|
|
reasons.append("all available metrics are below empirical thresholds")
|
|
if metrics["source"] == "adb-dumpsys-dbinfo" and level != "OK":
|
|
reasons.append("use an offline snapshot to confirm row count and age")
|
|
return level, reasons
|
|
|
|
|
|
def print_human(metrics: dict[str, Any], level: str, reasons: list[str]) -> None:
|
|
print("Philips TPM191E EPG health check (READ-ONLY)")
|
|
print(f"Source: {metrics['source']}")
|
|
print(f"EPG DB: {metrics['db_bytes']:,} bytes ({metrics['db_bytes'] / MIB:.2f} MiB)")
|
|
if "schedule_rows" in metrics:
|
|
age = metrics.get("oldest_age_days")
|
|
age_text = "n/a" if age is None else f"{age:.1f} days"
|
|
print(f"Schedule rows: {metrics['schedule_rows']:,}")
|
|
print(f"Oldest schedule age: {age_text}")
|
|
print(f"Rows older than 30 days: {metrics['rows_older_than_30_days']:,}")
|
|
print(f"SQLite quick_check: {metrics['quick_check']}")
|
|
if "data_free_kib" in metrics:
|
|
print(
|
|
f"/data free: {metrics['data_free_kib'] / 1024:.1f} MiB "
|
|
f"({metrics['data_use_percent']} used)"
|
|
)
|
|
print(f"Decision: {level}")
|
|
for reason in reasons:
|
|
print(f"- {reason}")
|
|
print("No repair was attempted.")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--snapshot", type=Path, help="consistent offline EpgDatabase.db copy")
|
|
parser.add_argument("--adb", type=Path, default=Path("platform-tools/adb.exe"))
|
|
parser.add_argument("--serial", default="<TV_IP>:5555", help="ADB target, for example 192.0.2.10:5555")
|
|
parser.add_argument("--json", action="store_true")
|
|
parser.add_argument("--now", type=int, help="UTC epoch seconds for reproducible snapshot tests")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
now_epoch = args.now or int(time.time())
|
|
if args.snapshot:
|
|
metrics = snapshot_metrics(args.snapshot, now_epoch)
|
|
else:
|
|
if args.serial == "<TV_IP>:5555":
|
|
raise RuntimeError("pass the TV address with --serial <TV_IP>:5555")
|
|
adb = args.adb.resolve(strict=True)
|
|
metrics = parse_dbinfo(
|
|
run_readonly_adb(
|
|
adb,
|
|
args.serial,
|
|
["shell", "dumpsys", "dbinfo", "org.droidtv.channels", "-v"],
|
|
)
|
|
)
|
|
metrics.update(
|
|
parse_df(run_readonly_adb(adb, args.serial, ["shell", "df", "-k", "/data"]))
|
|
)
|
|
level, reasons = assess(metrics)
|
|
except (OSError, RuntimeError, sqlite3.Error, subprocess.TimeoutExpired) as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 3
|
|
|
|
if args.json:
|
|
print(json.dumps({"decision": level, "reasons": reasons, **metrics}, indent=2))
|
|
else:
|
|
print_human(metrics, level, reasons)
|
|
return {"OK": 0, "WARN": 1, "MAINTENANCE_RECOMMENDED": 2}[level]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|