Merge Home & Phone into a three-tab Home that knows your house
Home is now Overview | My Home | Phone. Overview leads with quick-action tiles (focus, Do Not Disturb, health, snapshots, storage), keeps the findings card — updates fold in, the reclaim-space prompt is gone on purpose — and adds glance cards, the next calendar event, and weather. My Home groups every light by Home Assistant area: the helper gained an `areas` command (one REST template render, no websocket), and the rooms degrade to a flat list on setups without areas. The favorites editor and connection card moved intact. Phone gains a vitals strip — battery and cell signal read from KDE Connect's plugin D-Bus objects, where absence is data, not an error — beside ring, clipboard, send-a-file, and the BlueBubbles handoff. The retired home-phone id resolves to my-home forever via a new alias map in SettingsRoutes (with a hasOwnProperty guard so prototype names cannot leak into settingsPage). Storage no longer claims 0 B free — the old page read a field the disks helper never emitted. Contracts updated alongside; per the new workflow, the full suite runs once at the end of the redesign (see the test backlog note). Claude-Session: https://claude.ai/code/session_01Ms2FbjQy31TVf3CEvQhGM8
This commit is contained in:
@@ -27,6 +27,18 @@ DEVICE_OBJECT_PREFIX = "/modules/kdeconnect/devices"
|
||||
DEVICE_OBJECT_LINE = re.compile(
|
||||
rf"(?P<path>{re.escape(DEVICE_OBJECT_PREFIX)}/(?P<id>[A-Fa-f0-9]{{32,64}}))$"
|
||||
)
|
||||
BATTERY_INTERFACE = "org.kde.kdeconnect.device.battery"
|
||||
CONNECTIVITY_INTERFACE = "org.kde.kdeconnect.device.connectivity_report"
|
||||
# busctl spells every integer width with its own type code; a property that is
|
||||
# an int32 today may be read back as a uint32 by another kdeconnectd build.
|
||||
INT_PROPERTY_TYPES = frozenset({"y", "n", "q", "i", "u", "x", "t"})
|
||||
# Vitals are enrichment, not the answer: a phone that has just gone out of range
|
||||
# can leave a plugin object that blocks, and the status read must not stall
|
||||
# behind it. Shorter than the eight seconds the identity reads are allowed.
|
||||
VITALS_TIMEOUT = 3
|
||||
# Every device carries both keys so the shell never has to tell "not read" from
|
||||
# "no such field"; absent vitals are null, not missing.
|
||||
EMPTY_VITALS: dict[str, object] = {"battery": None, "signal": None}
|
||||
Runner = Callable[..., subprocess.CompletedProcess[str]]
|
||||
|
||||
|
||||
@@ -98,14 +110,21 @@ def normalize_device_line(
|
||||
#
|
||||
# The JSON form returns real UTF-8 and needs no unescaping, which is why these
|
||||
# parse a document rather than splitting words.
|
||||
def parse_property(output: str, expected: str) -> object | None:
|
||||
"""The value of a busctl --json=short property read, or None if it is not
|
||||
the type asked for."""
|
||||
def property_payload(output: str | None) -> dict[str, object] | None:
|
||||
"""The decoded document of a busctl --json=short read, or None when the
|
||||
output is missing or is not a property document at all."""
|
||||
try:
|
||||
payload = json.loads(output)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
return None
|
||||
if not isinstance(payload, dict) or payload.get("type") != expected:
|
||||
return payload if isinstance(payload, dict) else None
|
||||
|
||||
|
||||
def parse_property(output: str | None, expected: str) -> object | None:
|
||||
"""The value of a busctl --json=short property read, or None if it is not
|
||||
the type asked for."""
|
||||
payload = property_payload(output)
|
||||
if payload is None or payload.get("type") != expected:
|
||||
return None
|
||||
return payload.get("data")
|
||||
|
||||
@@ -117,16 +136,31 @@ def parse_loaded_plugins(output: str) -> list[str]:
|
||||
return [str(item) for item in data]
|
||||
|
||||
|
||||
def parse_string_property(output: str) -> str:
|
||||
def parse_string_property(output: str | None) -> str:
|
||||
data = parse_property(output, "s")
|
||||
return data if isinstance(data, str) else ""
|
||||
|
||||
|
||||
def parse_bool_property(output: str) -> bool | None:
|
||||
def parse_bool_property(output: str | None) -> bool | None:
|
||||
data = parse_property(output, "b")
|
||||
return data if isinstance(data, bool) else None
|
||||
|
||||
|
||||
def parse_int_property(output: str | None) -> int | None:
|
||||
"""The integer value of a busctl --json=short property read, or None.
|
||||
|
||||
JSON has one number type, so booleans are rejected explicitly: `true` is an
|
||||
int to isinstance and would otherwise read back as a charge of 1.
|
||||
"""
|
||||
payload = property_payload(output)
|
||||
if payload is None or payload.get("type") not in INT_PROPERTY_TYPES:
|
||||
return None
|
||||
data = payload.get("data")
|
||||
if isinstance(data, bool) or not isinstance(data, int):
|
||||
return None
|
||||
return data
|
||||
|
||||
|
||||
def run_command(
|
||||
command: list[str],
|
||||
*,
|
||||
@@ -222,6 +256,115 @@ def device_property(
|
||||
)
|
||||
|
||||
|
||||
# Each KDE Connect plugin hangs its own object off the device path, and those
|
||||
# objects exist only while the device is paired, reachable AND the plugin is
|
||||
# loaded. A phone with the battery plugin turned off, or one that just walked
|
||||
# out of range, makes busctl exit non-zero -- that is an absence of data, never
|
||||
# a fault to report, so every failure mode here collapses to None.
|
||||
def plugin_property(
|
||||
device_id: str,
|
||||
plugin: str,
|
||||
interface: str,
|
||||
member: str,
|
||||
runner: Runner = subprocess.run,
|
||||
) -> str | None:
|
||||
"""A plugin object's property read as raw busctl output, or None when the
|
||||
object, the plugin, or the daemon is not there."""
|
||||
try:
|
||||
result = run_command(
|
||||
[
|
||||
"busctl",
|
||||
"--user",
|
||||
"--json=short",
|
||||
"get-property",
|
||||
"org.kde.kdeconnect",
|
||||
f"{device_object(device_id)}/{plugin}",
|
||||
interface,
|
||||
member,
|
||||
],
|
||||
runner=runner,
|
||||
timeout=VITALS_TIMEOUT,
|
||||
)
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||
return None
|
||||
return result.stdout if result.returncode == 0 else None
|
||||
|
||||
|
||||
def device_battery(
|
||||
device_id: str,
|
||||
runner: Runner = subprocess.run,
|
||||
) -> dict[str, object] | None:
|
||||
"""Charge and charging state, or None when the phone has no battery to
|
||||
report.
|
||||
|
||||
Charge is read first because it is the cheapest proof the plugin object
|
||||
exists at all: when it is missing the other two reads are skipped, which is
|
||||
the common case for a device that is merely paired. kdeconnectd reports -1
|
||||
for "not known yet", and anything outside a percentage is not a charge.
|
||||
"""
|
||||
charge = parse_int_property(
|
||||
plugin_property(device_id, "battery", BATTERY_INTERFACE, "charge", runner)
|
||||
)
|
||||
if charge is None or not 0 <= charge <= 100:
|
||||
return None
|
||||
has_battery = parse_bool_property(
|
||||
plugin_property(device_id, "battery", BATTERY_INTERFACE, "hasBattery", runner)
|
||||
)
|
||||
if has_battery is False:
|
||||
return None
|
||||
charging = parse_bool_property(
|
||||
plugin_property(device_id, "battery", BATTERY_INTERFACE, "isCharging", runner)
|
||||
)
|
||||
return {"charge": charge, "charging": charging is True}
|
||||
|
||||
|
||||
def device_signal(
|
||||
device_id: str,
|
||||
runner: Runner = subprocess.run,
|
||||
) -> dict[str, object] | None:
|
||||
"""Cell network type and bar count, or None when there is no cellular
|
||||
report -- strength is -1 on a phone with no modem or no service."""
|
||||
strength = parse_int_property(
|
||||
plugin_property(
|
||||
device_id,
|
||||
"connectivity_report",
|
||||
CONNECTIVITY_INTERFACE,
|
||||
"cellularNetworkStrength",
|
||||
runner,
|
||||
)
|
||||
)
|
||||
if strength is None or strength < 0:
|
||||
return None
|
||||
network_type = parse_string_property(
|
||||
plugin_property(
|
||||
device_id,
|
||||
"connectivity_report",
|
||||
CONNECTIVITY_INTERFACE,
|
||||
"cellularNetworkType",
|
||||
runner,
|
||||
)
|
||||
)
|
||||
return {"networkType": network_type, "strength": strength}
|
||||
|
||||
|
||||
def device_vitals(
|
||||
device: dict[str, object],
|
||||
runner: Runner = subprocess.run,
|
||||
) -> dict[str, object]:
|
||||
"""The battery and signal fields for a device entry.
|
||||
|
||||
Only a paired, reachable device is queried; for anything else the plugin
|
||||
objects cannot exist, so asking would spend busctl calls to learn nothing.
|
||||
"""
|
||||
if not (device.get("paired") and device.get("reachable")):
|
||||
return dict(EMPTY_VITALS)
|
||||
device_id = str(device["id"])
|
||||
return {
|
||||
"battery": device_battery(device_id, runner),
|
||||
"signal": device_signal(device_id, runner),
|
||||
}
|
||||
|
||||
|
||||
def dbus_device_ids(runner: Runner = subprocess.run) -> list[str]:
|
||||
try:
|
||||
result = run_command(
|
||||
@@ -239,7 +382,11 @@ def dbus_device_ids(runner: Runner = subprocess.run) -> list[str]:
|
||||
]
|
||||
|
||||
|
||||
def dbus_devices(runner: Runner = subprocess.run) -> list[dict[str, object]]:
|
||||
def dbus_devices(
|
||||
runner: Runner = subprocess.run,
|
||||
*,
|
||||
vitals: bool = True,
|
||||
) -> list[dict[str, object]]:
|
||||
devices: list[dict[str, object]] = []
|
||||
for device_id in dbus_device_ids(runner):
|
||||
try:
|
||||
@@ -266,20 +413,27 @@ def dbus_devices(runner: Runner = subprocess.run) -> list[dict[str, object]]:
|
||||
if plugin in plugins
|
||||
}
|
||||
)
|
||||
devices.append(
|
||||
{
|
||||
"id": device_id,
|
||||
"name": name,
|
||||
"type": device_type or inferred_type(name),
|
||||
"paired": paired,
|
||||
"reachable": reachable,
|
||||
"actions": actions,
|
||||
}
|
||||
)
|
||||
device = {
|
||||
"id": device_id,
|
||||
"name": name,
|
||||
"type": device_type or inferred_type(name),
|
||||
"paired": paired,
|
||||
"reachable": reachable,
|
||||
"actions": actions,
|
||||
}
|
||||
device.update(device_vitals(device, runner) if vitals else EMPTY_VITALS)
|
||||
devices.append(device)
|
||||
return devices
|
||||
|
||||
|
||||
def collect_status(runner: Runner = subprocess.run) -> dict[str, object]:
|
||||
# `vitals` is off for the read that guards an action: ringing a phone needs its
|
||||
# identity and its action list, not its charge, and the extra busctl round trips
|
||||
# would sit between the tap and the ring.
|
||||
def collect_status(
|
||||
runner: Runner = subprocess.run,
|
||||
*,
|
||||
vitals: bool = True,
|
||||
) -> dict[str, object]:
|
||||
try:
|
||||
listing = run_command(
|
||||
["kdeconnect-cli", "--list-devices"],
|
||||
@@ -309,12 +463,13 @@ def collect_status(runner: Runner = subprocess.run) -> dict[str, object]:
|
||||
device = normalize_device_line(line, plugins, device_type)
|
||||
except ValueError:
|
||||
continue
|
||||
device.update(device_vitals(device, runner) if vitals else EMPTY_VITALS)
|
||||
devices.append(device)
|
||||
|
||||
known_ids = {str(device["id"]) for device in devices}
|
||||
devices.extend(
|
||||
device
|
||||
for device in dbus_devices(runner)
|
||||
for device in dbus_devices(runner, vitals=vitals)
|
||||
if str(device["id"]) not in known_ids
|
||||
)
|
||||
|
||||
@@ -358,7 +513,7 @@ def invoke_action(
|
||||
runner: Runner = subprocess.run,
|
||||
) -> dict[str, object]:
|
||||
device_id = validate_device_id(device_id)
|
||||
status = collect_status(runner)
|
||||
status = collect_status(runner, vitals=False)
|
||||
device = next(
|
||||
(
|
||||
item
|
||||
|
||||
Reference in New Issue
Block a user