1 Commits
Author SHA1 Message Date
zach 4f544981c1 delete unknown json objects + write missing objects to config file 2026-08-16 12:17:50 +02:00
24 changed files with 317 additions and 1082 deletions
+89 -24
View File
@@ -1,15 +1,12 @@
name: Python
env:
APT_DEPS: git python3 python3-pip python3-venv python3-gi python3-cairo python3-dbus python3-pyudev python3-psutil python3-evdev python3-yaml python3-xlib python3-pillow gir1.2-gtk-3.0
on:
pull_request:
jobs:
static:
runs-on: debian
container: node:26-trixie-slim
fmt:
runs-on: alpine
container: node:26-alpine
steps:
- name: Checkout
@@ -17,8 +14,10 @@ jobs:
- name: Install tools
run: |
apt-get update
apt-get install -y --no-install-recommends git python3 python3-pip python3-venv
apk add --no-cache \
git \
python3 \
py3-pip
python3 -m venv .venv
. .venv/bin/activate
pip install --no-cache-dir ruff
@@ -28,15 +27,9 @@ jobs:
. .venv/bin/activate
ruff format --check .
- name: Lint
run: |
. .venv/bin/activate
ruff check .
verify:
if: always()
runs-on: debian
container: node:26-trixie-slim
lint:
runs-on: alpine
container: node:26-alpine
steps:
- name: Checkout
@@ -44,16 +37,43 @@ jobs:
- name: Install tools
run: |
apt-get update
apt-get install -y --no-install-recommends $APT_DEPS build-essential python3-dev
python3 -m venv --system-site-packages .venv
apk add --no-cache \
git \
python3 \
py3-pip
python3 -m venv .venv
. .venv/bin/activate
pip install --no-cache-dir basedpyright nuitka .
pip install --no-cache-dir ruff
- name: Type check
- name: Lint
run: |
. .venv/bin/activate
basedpyright
ruff check .
test:
runs-on: alpine
container: node:26-alpine
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Install tools
run: |
apk add --no-cache \
git \
python3 \
py3-pip \
py3-pillow \
build-base
python3 -m venv .venv
. .venv/bin/activate
pip install --no-cache-dir \
typer \
pillow \
materialyoucolor \
jinja2 \
pytest
- name: Test
run: |
@@ -61,7 +81,52 @@ jobs:
cd cli
python -m pytest tests/ -v
typecheck:
runs-on: alpine
container: node:26-alpine
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Install tools
run: |
apk add --no-cache \
git \
python3 \
py3-pip
python3 -m venv .venv
. .venv/bin/activate
pip install --no-cache-dir basedpyright typer pillow materialyoucolor jinja2
- name: Type check
run: |
. .venv/bin/activate
basedpyright
buildcheck:
runs-on: alpine
container: node:26-alpine
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Install tools
run: |
apk add --no-cache \
git \
python3 \
py3-pip \
build-base \
python3-dev \
gcc \
g++
python3 -m venv .venv
. .venv/bin/activate
pip install --no-cache-dir nuitka
- name: Nuitka module check
run: |
. .venv/bin/activate
nuitka --module --include-package=zshell cli/src/zshell/
nuitka --module --include-package=zshell cli/src/zshell/
-2
View File
@@ -16,5 +16,3 @@ dist/
**/test-plugins/
**/Charts/
network-dev/
**/zshell.build/
**/zshell.dist/
-2
View File
@@ -93,8 +93,6 @@ if("shell" IN_LIST ENABLE_MODULES)
${NUITKA_EXECUTABLE}
--standalone
--include-data-dir=${CMAKE_SOURCE_DIR}/cli/src/zshell/assets=zshell/assets
--include-package=gi.overrides
--include-package-data=solaar
--output-dir=${ZSHELL_CLI_BUILD_DIR}
--output-filename=zshell-cli
${CMAKE_SOURCE_DIR}/cli/src/zshell/
+4 -4
View File
@@ -126,7 +126,7 @@ MouseArea {
CustomRect {
id: item
readonly property bool active: modelData === root.active
readonly property bool active: modelData === root?.active
required property int index
required property MenuItem modelData
@@ -163,19 +163,19 @@ MouseArea {
MaterialIcon {
Layout.alignment: Qt.AlignVCenter
color: item.active ? Colors.palette.m3onTertiaryContainer : Colors.palette.m3onSurfaceVariant
text: item.modelData.icon
text: item.modelData?.icon ?? ""
}
CustomText {
Layout.alignment: Qt.AlignVCenter
Layout.fillWidth: true
color: item.active ? Colors.palette.m3onTertiaryContainer : Colors.palette.m3onSurface
text: item.modelData.text
text: item.modelData?.text ?? ""
}
Loader {
Layout.alignment: Qt.AlignVCenter
active: item.modelData.trailingIcon.length > 0
active: item.modelData?.trailingIcon.length > 0
asynchronous: true
visible: active
-2
View File
@@ -302,8 +302,6 @@ Item {
root.visibilities.sidebar = false;
root.panels.popouts.hasCurrent = false;
root.visibilities.launcher = false;
} else {
Config.save();
}
}
+12 -40
View File
@@ -1,63 +1,35 @@
pragma Singleton
import Quickshell
import Quickshell.Io
import Quickshell.Services.UPower
import ZShell.Config
import qs.Services
import qs.Paths
Singleton {
id: root
readonly property list<var> allPeripherals: [...logiDevices, ...upowerDevices.filter(d => !Battery.logiDevices.some(l => l.nativePath === d.nativePath))]
readonly property real currentPerc: UPower.displayDevice.percentage
readonly property var deviceState: UPower.displayDevice.state
readonly property string deviceStateString: UPowerDeviceState.toString(deviceState)
readonly property bool isLaptop: UPower.displayDevice.isLaptopBattery
readonly property alias logiDevices: adapter.devices
readonly property var lowestPeripheral: allPeripherals.reduce((lowest, current) => current.percentage < lowest.percentage ? current : lowest)
readonly property bool onBattery: UPower.onBattery
readonly property bool ready: UPower.displayDevice.ready
readonly property real timeToEmpty: UPower.displayDevice.timeToEmpty
readonly property real timeToFull: UPower.displayDevice.timeToFull
readonly property list<UPowerDevice> upowerDevices: UPower.devices.values.filter(d => d.state !== UPowerDeviceState.Unknown)
function getColors(percentage: real, state: string): var {
if (state === "charging" || state === "full" || state === "recharging")
readonly property var colors: {
if (deviceState === UPowerDeviceState.Charging || deviceState === UPowerDeviceState.FullyCharged)
return {
fg: Colors.swapRG(Colors.palette.m3error),
bg: Colors.swapRG(Colors.palette.m3onError)
};
else if (percentage <= 0.2)
else if (currentPerc <= 0.2)
return {
fg: Colors.palette.m3error,
bg: Colors.palette.m3onError
};
else
return {
fg: Colors.palette.m3tertiary,
bg: Colors.palette.m3onTertiary
fg: Colors.palette.m3onSurface,
bg: Colors.palette.m3surface
};
}
function startDaemon(): void {
Quickshell.execDetached(["zshell-cli", "battery", "daemon"]);
}
FileView {
id: fileView
path: `${Paths.cache}/battery.json`
watchChanges: true
onFileChanged: reload()
JsonAdapter {
id: adapter
property list<var> devices: []
property real updated: 0.0
}
}
readonly property real currentPerc: UPower.displayDevice.percentage
readonly property var deviceState: UPower.displayDevice.state
readonly property bool isLaptop: UPower.displayDevice.isLaptopBattery
readonly property bool onBattery: UPower.onBattery
readonly property bool ready: UPower.displayDevice.ready
readonly property real timeToEmpty: UPower.displayDevice.timeToEmpty
readonly property real timeToFull: UPower.displayDevice.timeToFull
}
-1
View File
@@ -73,7 +73,6 @@ Singleton {
Config.dock.pinnedApps = pinnedApps;
root.unpinnedOrder = visibleUnpinned.concat(root.unpinnedOrder.map(normalizeId).filter(id => !pinnedApps.includes(id) && !visibleUnpinned.includes(id)));
Config.saveNoToast();
}
function isPinned(appId) {
@@ -1,85 +0,0 @@
import QtQuick
import ZShell.Config
import qs.Services
import qs.Components
import qs.Helpers
Row {
id: root
property real batHeight: Tokens.padding.smaller * 2
property real batWidth: Tokens.padding.larger * 2
required property string devState
property real nubHeight: Tokens.padding.small
property real nubWidth: 2
required property real percentage
property real radius: Tokens.rounding.smallest / 2
spacing: 1
CustomRect {
id: track
anchors.verticalCenter: parent.verticalCenter
color: Battery.getColors(root.percentage, root.devState).bg
height: root.batHeight
radius: root.radius
width: root.batWidth
CustomText {
color: Battery.getColors(root.percentage, root.devState).fg
font.pointSize: Tokens.font.size.larger / 1.5
font.weight: 800
height: track.height
horizontalAlignment: Text.AlignHCenter
text: Math.round(root.percentage * 100)
verticalAlignment: Text.AlignVCenter
width: track.width
}
Item {
clip: true
width: parent.width * root.percentage
anchors {
bottom: parent.bottom
left: parent.left
top: parent.top
}
CustomRect {
id: fill
color: Battery.getColors(root.percentage, root.devState).fg
height: track.height
radius: track.radius
width: track.width
CustomText {
id: batteryLabel
clip: true
color: Battery.getColors(root.percentage, root.devState).bg
font.pointSize: Tokens.font.size.larger / 1.5
font.weight: 800
height: track.height
horizontalAlignment: Text.AlignHCenter
text: Math.round(root.percentage * 100)
verticalAlignment: Text.AlignVCenter
width: track.width
}
}
}
}
CustomRect {
id: nub
anchors.verticalCenter: parent.verticalCenter
bottomRightRadius: 20
color: root.percentage < 0.99 ? track.color : fill.color
height: root.nubHeight
topRightRadius: 20
width: root.nubWidth
}
}
+1
View File
@@ -112,6 +112,7 @@ WidgetBase {
name: "upower"
UPowerWidget {
horizontal: root.horizontal
}
}
}
@@ -2,17 +2,16 @@ import Quickshell.Services.UPower
import QtQuick
import QtQuick.Layouts
import ZShell.Config
import qs.Modules.Bar.Components.Common
import qs.Components
import qs.Helpers
Item {
id: root
readonly property Item currentItem: batteryIconLoader?.item ?? peripheralIconLoader?.item ?? upowerIconLoader.item
required property bool horizontal
implicitHeight: currentItem?.implicitHeight ?? 0
implicitWidth: currentItem?.implicitWidth ?? 0
implicitHeight: Battery.isLaptop ? batteryIconLoader.item.implicitHeight : upowerIconLoader.item.implicitHeight
implicitWidth: Battery.isLaptop ? batteryIconLoader.item.implicitWidth : upowerIconLoader.item.implicitWidth
Behavior on Layout.preferredHeight {
Anim {
@@ -31,36 +30,95 @@ Item {
}
}
Component.onCompleted: Battery.startDaemon()
Loader {
id: batteryIconLoader
active: Battery.isLaptop
anchors.centerIn: parent
sourceComponent: BatteryIcon {
devState: Battery.deviceStateString
percentage: Battery.currentPerc
}
}
sourceComponent: Row {
id: batteryIcon
Loader {
id: peripheralIconLoader
property real batHeight: 16
property real batWidth: 30
property real nubHeight: 6
property real nubWidth: 2
property real radius: Tokens.rounding.smallest / 2
active: (Battery.lowestPeripheral?.isPresent ?? false) && !batteryIconLoader.active
anchors.centerIn: parent
spacing: 1
sourceComponent: BatteryIcon {
devState: Battery.lowestPeripheral?.state ?? ""
percentage: Battery.lowestPeripheral?.percentage ?? 0
CustomRect {
id: track
anchors.verticalCenter: parent.verticalCenter
color: Battery.colors.bg
height: batteryIcon.batHeight
radius: batteryIcon.radius
width: batteryIcon.batWidth
CustomText {
color: Battery.colors.fg
font.pointSize: Tokens.font.size.larger / 1.5
font.weight: 800
height: track.height
horizontalAlignment: Text.AlignHCenter
text: Math.round(Battery.currentPerc * 100)
verticalAlignment: Text.AlignVCenter
width: track.width
}
Item {
clip: true
width: parent.width * Battery.currentPerc
anchors {
bottom: parent.bottom
left: parent.left
top: parent.top
}
CustomRect {
id: fill
color: Battery.colors.fg
height: track.height
radius: track.radius
width: track.width
CustomText {
id: batteryLabel
clip: true
color: Battery.colors.bg
font.pointSize: 7.5
font.weight: 800
height: track.height
horizontalAlignment: Text.AlignHCenter
text: Math.round(Battery.currentPerc * 100)
verticalAlignment: Text.AlignVCenter
width: track.width
}
}
}
}
CustomRect {
id: nub
anchors.verticalCenter: parent.verticalCenter
bottomRightRadius: 20
color: Battery.currentPerc < 0.99 ? track.color : fill.color
height: batteryIcon.nubHeight
topRightRadius: 20
width: batteryIcon.nubWidth
}
}
}
Loader {
id: upowerIconLoader
active: !batteryIconLoader.active && !peripheralIconLoader.active
active: !Battery.isLaptop
anchors.centerIn: parent
sourceComponent: MaterialIcon {
+1 -28
View File
@@ -1,10 +1,8 @@
pragma ComponentBehavior: Bound
import QtQuick
import QtQuick.Layouts
import Quickshell.Services.UPower
import ZShell.Config
import qs.Modules.Bar.Components.Common
import qs.Helpers
import qs.Components
import qs.Services
@@ -24,31 +22,6 @@ Column {
}
}
Repeater {
model: Battery.allPeripherals
RowLayout {
id: layout
required property var modelData
anchors.left: parent.left
anchors.leftMargin: Tokens.padding.normal
anchors.right: parent.right
anchors.rightMargin: Tokens.padding.normal
CustomText {
Layout.fillWidth: true
text: layout.modelData.model
}
BatteryIcon {
devState: layout.modelData.state
percentage: layout.modelData.percentage
}
}
}
CustomText {
function formatSeconds(s: int, fallback: string): string {
const day = Math.floor(s / 86400);
@@ -102,7 +75,7 @@ Column {
CustomText {
anchors.verticalCenter: parent.verticalCenter
color: Colors.palette.m3onError
font.family: Config.appearance.font.family.mono
font.family: Appearance.font.family.mono
text: qsTr("Performance Degraded")
}
@@ -82,7 +82,6 @@ Searcher {
list.visibilities.launcher = false;
Quickshell.execDetached(["zshell-cli", "scheme", "generate", "--scheme", variant]);
Config.colors.schemeType = variant;
Config.save();
}
}
}
+1 -4
View File
@@ -70,10 +70,7 @@ PageBase {
}
]
onSelected: item => {
Config.screenshot.mode = item.value;
Config.save();
}
onSelected: item => Config.screenshot.mode = item.value
}
SectionHeader {
-2
View File
@@ -104,7 +104,6 @@ PageBase {
onApplySettings: (start, end) => {
Config.general.color.scheduleDarkStart = start;
Config.general.color.scheduleDarkEnd = end;
Config.save();
ModeScheduler.checkStartup();
PopupManager.requestClose();
}
@@ -146,7 +145,6 @@ PageBase {
onApplySettings: (start, end) => {
Config.general.color.scheduleHyprsunsetStart = start;
Config.general.color.scheduleHyprsunsetEnd = end;
Config.save();
Hyprsunset.checkStartup();
PopupManager.requestClose();
}
+29 -19
View File
@@ -95,15 +95,31 @@ QString Config::filePath() const {
void Config::loadSync() {
m_loading = true;
QFile f(filePath());
QJsonObject before;
bool existed = false;
if (f.open(QIODevice::ReadOnly)) {
const auto doc = QJsonDocument::fromJson(f.readAll());
if (doc.isObject()) loadFromJson(QJsonValue(doc.object()));
if (doc.isObject()) {
before = doc.object();
existed = true;
} else {
qInfo() << "Config: existing config at" << filePath()
<< "is empty or not a valid JSON object - using defaults";
}
} else {
qInfo() << "Config: no existing config at" << filePath()
<< "- using defaults";
}
loadFromJson(QJsonValue(before));
m_loading = false;
const auto after = toJson().toObject();
if (!existed || after != before) saveNow();
}
void Config::updateWatch() {
@@ -148,27 +164,21 @@ void Config::loadAsync() {
if (f.open(QIODevice::ReadOnly)) {
const auto doc = QJsonDocument::fromJson(f.readAll());
const bool valid = doc.isObject();
const QJsonObject before = valid ? doc.object() : QJsonObject();
if (doc.isObject()) {
QMetaObject::invokeMethod(
this,
[this, doc]() {
loadFromJson(QJsonValue(doc.object()));
m_loading = false;
QMetaObject::invokeMethod(
this,
[this, valid, before]() {
loadFromJson(QJsonValue(before));
m_loading = false;
if (m_reloadPending) m_reloadTimer.start();
},
Qt::QueuedConnection);
} else {
QMetaObject::invokeMethod(
this,
[this]() {
m_loading = false;
const auto after = toJson().toObject();
if (!valid || after != before) saveNow();
if (m_reloadPending) m_reloadTimer.start();
},
Qt::QueuedConnection);
}
if (m_reloadPending) m_reloadTimer.start();
},
Qt::QueuedConnection);
} else {
qInfo() << "Config: failed to reload from" << filePath()
<< "- using in-memory values";
+14 -5
View File
@@ -79,13 +79,14 @@ void ConfigList::loadFromJson(const QJsonValue& json) {
if (!json.isArray()) {
qCWarning(
lcConfig,
"Option '%s' must be a list, ignoring",
"Option '%s' must be a list, resetting to default",
qUtf8Printable(propertyPath()));
m_rejectedJson = json;
resetToDefaults();
materializeDefaults();
return;
}
m_rejectedJson = QJsonValue::Undefined;
populate(json.toArray());
m_loaded = true;
}
@@ -93,12 +94,11 @@ void ConfigList::loadFromJson(const QJsonValue& json) {
QJsonValue ConfigList::toJson() const {
if (m_loaded) return elementsToJson();
return m_rejectedJson;
return QJsonValue::Undefined;
}
void ConfigList::clearLoadedKeys() {
m_loaded = false;
m_rejectedJson = QJsonValue::Undefined;
}
QStringList ConfigList::unknownKeys() const {
@@ -113,6 +113,15 @@ QStringList ConfigList::unknownKeys() const {
return keys;
}
void ConfigList::materializeDefaults() {
for (auto* const item : m_items)
item->materializeDefaults();
if (m_global) return;
m_loaded = true;
}
void ConfigList::resyncFromGlobal() {
syncValuesFromGlobal();
}
+3 -5
View File
@@ -33,6 +33,7 @@ class ConfigList : public ConfigNode {
[[nodiscard]] QJsonValue toJson() const override;
void clearLoadedKeys() override;
[[nodiscard]] QStringList unknownKeys() const override;
void materializeDefaults() override;
void resyncFromGlobal() override;
signals:
@@ -71,7 +72,6 @@ class ConfigList : public ConfigNode {
QJsonArray m_defaults;
QList<ConfigObject*> m_items;
bool m_loaded = false;
QJsonValue m_rejectedJson = QJsonValue::Undefined;
};
} // namespace ZShell::config
@@ -103,7 +103,6 @@ class ConfigList : public ConfigNode {
} \
};
#define CONFIG_LIST(Type, name, ...) \
Q_PROPERTY(ZShell::config::Type* name READ name CONSTANT) \
\
@@ -115,6 +114,8 @@ class ConfigList : public ConfigNode {
private: \
Type* m_##name = new Type(this __VA_OPT__(, __VA_ARGS__));
#define LIST_ENTRY(id, enabled) \
vmap({{"id", QString::fromUtf8(#id)}, {"enabled", enabled}})
namespace ZShell::config {
@@ -136,6 +137,3 @@ class ListEntry : public ConfigObject {
CONFIG_LIST_TYPE(ListEntry, EntryList)
} // namespace ZShell::config
#define LIST_ENTRY(id, enabled) \
vmap({{"id", QString::fromUtf8(#id)}, {"enabled", enabled}})
+2
View File
@@ -26,6 +26,8 @@ class ConfigNode : public QObject {
[[nodiscard]] virtual QStringList unknownKeys() const = 0;
[[nodiscard]] virtual QList<ConfigNode*> childNodes() const;
virtual void materializeDefaults() = 0;
void syncFromGlobal(ConfigNode* global);
virtual void resyncFromGlobal() = 0;
+78 -4
View File
@@ -7,6 +7,51 @@
namespace ZShell::config {
namespace {
bool isStringArray(const QJsonArray& arr) {
for (const auto& v : arr) {
if (!v.isString()) return false;
}
return true;
}
bool jsonValueMatchesType(const QJsonValue& val, QMetaType type) {
switch (val.type()) {
case QJsonValue::Bool:
return type.id() == QMetaType::Bool;
case QJsonValue::Double:
switch (type.id()) {
case QMetaType::Int:
case QMetaType::UInt:
case QMetaType::LongLong:
case QMetaType::ULongLong:
case QMetaType::Double:
case QMetaType::Float:
return true;
default:
return false;
}
case QJsonValue::String:
return type.id() == QMetaType::QString;
case QJsonValue::Array:
if (type.id() == QMetaType::QStringList)
return isStringArray(val.toArray());
return type.id() == QMetaType::QVariantList;
case QJsonValue::Object:
case QJsonValue::Null:
case QJsonValue::Undefined:
default:
return false;
}
}
} // namespace
ConfigObject::ConfigObject(QObject* parent) : ConfigNode(parent) {}
void ConfigObject::loadFromJson(const QJsonValue& json) {
@@ -14,6 +59,7 @@ void ConfigObject::loadFromJson(const QJsonValue& json) {
const auto* meta = metaObject();
QSet<QString> known;
QSet<QString> invalid;
for (int i = basePropertyOffset(); i < meta->propertyCount(); ++i) {
auto prop = meta->property(i);
@@ -34,6 +80,15 @@ void ConfigObject::loadFromJson(const QJsonValue& json) {
if (!prop.isWritable()) continue;
if (!jsonValueMatchesType(jsonVal, prop.metaType())) {
qWarning() << "Config: type mismatch for" << key << "in"
<< meta->className() << "- expected"
<< prop.metaType().name() << "got value" << jsonVal
<< "- resetting to default";
invalid.insert(key);
continue;
}
if (prop.metaType().id() == QMetaType::QStringList) {
QStringList list;
const auto jsonArr = jsonVal.toArray();
@@ -56,10 +111,12 @@ void ConfigObject::loadFromJson(const QJsonValue& json) {
m_extras = {};
for (auto it = obj.begin(); it != obj.end(); ++it) {
if (!known.contains(it.key())) {
if (!known.contains(it.key()) || invalid.contains(it.key())) {
m_extras.insert(it.key(), it.value());
}
}
materializeDefaults();
}
QJsonValue ConfigObject::toJson() const {
@@ -104,9 +161,6 @@ QJsonValue ConfigObject::toJson() const {
obj.insert(key, QJsonValue::fromVariant(value));
}
for (auto it = m_extras.begin(); it != m_extras.end(); ++it)
obj.insert(it.key(), it.value());
if (obj.isEmpty()) return QJsonValue::Undefined;
return obj;
@@ -159,6 +213,26 @@ QList<ConfigNode*> ConfigObject::childNodes() const {
return nodes;
}
void ConfigObject::materializeDefaults() {
const auto* meta = metaObject();
for (int i = basePropertyOffset(); i < meta->propertyCount(); ++i) {
const auto prop = meta->property(i);
const auto key = QString::fromUtf8(prop.name());
if (auto* const node = prop.read(this).value<ConfigNode*>()) {
node->materializeDefaults();
continue;
}
if (!prop.isWritable()) continue;
if (m_global) continue;
m_loadedKeys.insert(key);
}
}
void ConfigObject::syncValuesFromGlobal() {
const auto* meta = metaObject();
+1
View File
@@ -62,6 +62,7 @@ class ConfigObject : public ConfigNode {
void clearLoadedKeys() override;
[[nodiscard]] QStringList unknownKeys() const override;
[[nodiscard]] QList<ConfigNode*> childNodes() const override;
void materializeDefaults() override;
void resyncFromGlobal() override;
[[nodiscard]] virtual QStringList identityKeys() const;
-1
View File
@@ -113,7 +113,6 @@ Singleton {
function setMode(mode: string): void {
Quickshell.execDetached(["zshell-cli", "scheme", "generate", "--mode", mode]);
Config.general.color.mode = mode;
Config.save();
}
function swapRG(c: color): color {
+1 -9
View File
@@ -8,14 +8,7 @@ import typer
from typer._completion_classes import completion_init
from typer._completion_shared import _get_shell_name, install
from zshell.subcommands import (
battery,
record,
scheme,
screenshot,
shell,
wallpaper,
)
from zshell.subcommands import record, scheme, screenshot, shell, wallpaper
app = typer.Typer(name="zshell-cli", add_completion=False)
@@ -24,7 +17,6 @@ app.add_typer(scheme.app, name="scheme")
app.add_typer(screenshot.app, name="screenshot")
app.add_typer(wallpaper.app, name="wallpaper")
app.add_typer(record.app, name="record")
app.add_typer(battery.app, name="battery")
def _completion_installed() -> bool:
-822
View File
@@ -1,822 +0,0 @@
from __future__ import annotations
import contextlib
import fcntl
import json
import logging
import os
import re
import signal
import struct
import tempfile
import threading
import time
from collections.abc import Callable
from dataclasses import asdict, dataclass
from pathlib import Path
import typer
from logitech_receiver import hidpp10, hidpp20
from logitech_receiver.common import Battery, BatteryStatus, Notification
from logitech_receiver.hidpp10_constants import Registers
from logitech_receiver.hidpp20_constants import SupportedFeature
app = typer.Typer(
help="Read live battery status from Logitech HID++ peripherals."
)
logger = logging.getLogger("zshell.battery")
DEFAULT_OUTPUT = Path.home() / ".cache" / "zshell" / "battery.json"
_RUNTIME_DIR = Path(
os.environ.get("XDG_RUNTIME_DIR") or (Path.home() / ".cache" / "zshell")
)
DEFAULT_LOCK = _RUNTIME_DIR / "zshell-battery-daemon.lock"
def _acquire_singleton_lock(path: Path) -> int | None:
path.parent.mkdir(parents=True, exist_ok=True)
fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o644)
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
os.close(fd)
return None
os.ftruncate(fd, 0)
os.write(fd, str(os.getpid()).encode())
return fd
@dataclass
class DeviceBattery:
model: str
nativePath: str | None
serial: str | None
type: str | None
wired: bool
percentage: float | None
state: str | None
charging: bool
isPresent: bool
_SYSFS_POWER_SUPPLY = Path("/sys/class/power_supply")
def _normalize_serial(s: str) -> str:
return re.sub(r"[^0-9A-Za-z]", "", s).upper()
def _find_upower_native_path(serial: str | None) -> str | None:
if not serial or not _SYSFS_POWER_SUPPLY.is_dir():
return None
target = _normalize_serial(serial)
for entry in _SYSFS_POWER_SUPPLY.glob("hidpp_battery_*"):
try:
raw = (entry / "serial_number").read_text().strip()
except OSError:
logger.debug(
"couldn't read serial_number under %s", entry, exc_info=True
)
continue
if _normalize_serial(raw) == target:
return entry.name
return None
def _process_battery_notification(dev, n) -> bool:
if not getattr(dev, "isDevice", False):
return False
if int(n.sub_id) & 0x80:
return False
if n.sub_id == Notification.NO_OPERATION:
return False
if getattr(dev, "protocol", None) is not None and dev.protocol < 2.0:
if n.sub_id in (Registers.BATTERY_STATUS, Registers.BATTERY_CHARGE):
if n.data[-1:] != b"\x00":
return False
data = bytes([n.address]) + n.data
dev.set_battery_info(hidpp10.parse_battery_status(n.sub_id, data))
return True
return False
features = getattr(dev, "features", None)
if not features:
return False
try:
feature = features.get_feature(n.sub_id)
except Exception:
return False
if feature == SupportedFeature.BATTERY_STATUS:
if n.address == 0x00:
dev.set_battery_info(hidpp20.decipher_battery_status(n.data)[1])
return True
return False
if feature == SupportedFeature.BATTERY_VOLTAGE:
if n.address == 0x00:
dev.set_battery_info(hidpp20.decipher_battery_voltage(n.data)[1])
return True
return False
if feature == SupportedFeature.UNIFIED_BATTERY:
if n.address == 0x00:
dev.set_battery_info(hidpp20.decipher_battery_unified(n.data)[1])
return True
return False
if feature == SupportedFeature.ADC_MEASUREMENT:
if n.address == 0x00:
result = hidpp20.decipher_adc_measurement(n.data)
if result:
dev.set_battery_info(result[1])
return True
return False
if feature == SupportedFeature.CENTURION_BATTERY_SOC:
dev.set_battery_info(hidpp20.decipher_battery_centurion(n.data)[1])
return True
if feature == SupportedFeature.SOLAR_DASHBOARD:
if n.data[5:9] == b"GOOD":
charge, lux, _ = struct.unpack("!BHH", n.data[:5])
status = BatteryStatus.DISCHARGING
if n.address == 0x10 and lux > 200:
status = BatteryStatus.RECHARGING
dev.set_battery_info(Battery(charge, None, status, None, lux))
return True
return False
return False
def _import_listener_deps():
try:
from logitech_receiver import base, device, receiver
from logitech_receiver.listener import EventsListener
except ImportError as e:
raise RuntimeError(
"logitech_receiver isn't importable -- install it with `pip install solaar`."
) from e
except ValueError as e:
raise RuntimeError(
"logitech_receiver failed to import because a GTK3 typelib is missing "
f"({e}). Install the same GTK3 + PyGObject packages Solaar's GUI needs "
"(e.g. python3-gi + gir1.2-gtk-3.0 on Debian/Ubuntu, or the gtk3/"
"python3-gobject equivalents on your distro)."
) from e
class _BatteryEventsListener(EventsListener):
def __init__(self, receiver_or_device, on_change):
super().__init__(receiver_or_device, self._handle_notification)
self._on_change = on_change
def has_started(self):
if not self.receiver.isDevice:
try:
self.receiver.notification_flags = (
self.receiver.enable_connection_notifications()
)
self.receiver.notify_devices()
except Exception:
logger.exception(
"failed enabling notifications for %s", self.receiver
)
def _handle_notification(self, n):
if self.receiver.isDevice:
try:
_process_battery_notification(self.receiver, n)
except Exception:
logger.exception("processing %s for %s", n, self.receiver)
self._on_change(self.receiver)
return
if n.devnumber == 0xFF:
try:
_process_battery_notification(self.receiver, n)
except Exception:
logger.exception("processing receiver notification %s", n)
self._on_change(self.receiver)
return
if not (0 < n.devnumber <= 16):
logger.warning(
"unexpected device number %s in %s", n.devnumber, n
)
return
try:
dev = self.receiver[n.devnumber]
except Exception:
logger.exception(
"resolving device number %s on %s",
n.devnumber,
self.receiver,
)
return
if not dev:
logger.warning(
"%s: received %s for invalid device %d",
self.receiver,
n,
n.devnumber,
)
return
try:
_process_battery_notification(dev, n)
except Exception:
logger.exception("processing %s for %s", n, dev)
self._on_change(dev)
return base, device, receiver, _BatteryEventsListener
def _start_hotplug_watcher(on_hotplug: Callable[[], None]):
try:
import pyudev
except ImportError:
logger.warning(
"pyudev isn't importable -- hotplugged devices will only be "
"noticed on the next --rescan-interval tick, not instantly. "
"Install it with `pip install pyudev` (it's a core Solaar "
"dependency, so it's normally already present)."
)
return None
debounce_lock = threading.Lock()
debounce_state = {"timer": None}
DEBOUNCE_SECONDS = 0.5
def _fire():
with debounce_lock:
debounce_state["timer"] = None
on_hotplug()
def _handle_event(_device):
if _device.action not in ("add", "remove"):
return
with debounce_lock:
existing = debounce_state["timer"]
if existing is not None:
existing.cancel()
timer = threading.Timer(DEBOUNCE_SECONDS, _fire)
timer.daemon = True
debounce_state["timer"] = timer
timer.start()
context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by(subsystem="hidraw")
observer = pyudev.MonitorObserver(monitor, callback=_handle_event)
observer.start()
return observer
class _ListenerRegistry:
def __init__(self, on_change):
self._on_change = on_change
self._entries = {}
self._last_change: dict[int, float] = {}
self._lock = threading.Lock()
self._rescan_lock = threading.Lock()
self._min_rescan_interval = 1.0
self._last_rescan_time = 0.0
self._hidpp_io_lock = threading.RLock()
def _touch_and_notify(self, dev_or_receiver, *args, **kwargs):
with self._lock:
self._last_change[id(dev_or_receiver)] = time.time()
self._on_change(dev_or_receiver, *args, **kwargs)
def last_change(self, dev) -> float:
with self._lock:
return self._last_change.get(id(dev), 0.0)
def rescan(self):
now = time.monotonic()
if now - self._last_rescan_time < self._min_rescan_interval:
return
if not self._rescan_lock.acquire(blocking=False):
return
try:
self._last_rescan_time = now
with self._hidpp_io_lock:
base, device, receiver, ListenerClass = _import_listener_deps()
with self._lock:
dead = [
p
for p, (_obj, listener) in self._entries.items()
if not listener._active
]
for p in dead:
dying_obj, _dying_listener = self._entries[p]
if dying_obj.isDevice:
ident = getattr(
dying_obj, "unitId", None
) or getattr(dying_obj, "serial", None)
info = getattr(dying_obj, "battery_info", None)
if ident and info is not None:
for (
other_path,
(other_obj, _other_listener),
) in self._entries.items():
if other_path == p or other_obj.isDevice:
continue
for child in other_obj:
child_ident = getattr(
child, "unitId", None
) or getattr(child, "serial", None)
if child_ident == ident:
child.set_battery_info(info)
break
del self._entries[p]
known_paths = set(self._entries.keys())
with self._lock:
id_to_entry = {}
for path, (obj, listener) in self._entries.items():
if obj.isDevice:
ident = obj.unitId or obj.serial
if ident:
id_to_entry[ident] = (path, obj, listener)
for dev_info in base.receivers_and_devices():
if dev_info.path in known_paths:
continue
try:
if dev_info.isDevice:
obj = device.create_device(base, dev_info)
else:
obj = receiver.create_receiver(base, dev_info)
except OSError as e:
if e.errno == 13:
logger.error(
"permission denied opening %s -- check the Solaar udev rule "
"(rules.d/42-logitech-unify-permissions.rules) is installed",
dev_info.path,
)
else:
logger.exception("failed opening %s", dev_info)
continue
except Exception:
logger.exception("failed opening %s", dev_info)
continue
if obj is None:
continue
listener = ListenerClass(obj, self._touch_and_notify)
if obj.isDevice:
ident = None
try:
if obj.protocol >= 2.0:
obj.get_ids()
ident = obj.unitId or obj.serial
except Exception:
pass
if ident:
with self._lock:
existing = id_to_entry.get(ident)
if existing:
old_path, old_obj, old_listener = existing
if old_path != dev_info.path:
logger.info(
"replacing device %s (old path %s) with new path %s",
ident,
old_path,
dev_info.path,
)
if (
getattr(obj, "battery_info", None)
is None
and getattr(
old_obj, "battery_info", None
)
is not None
):
obj.set_battery_info(
old_obj.battery_info
)
old_listener.stop()
del self._entries[old_path]
self._entries[dev_info.path] = (
obj,
listener,
)
self._last_change[id(obj)] = time.time()
to_join = old_listener
break
else:
pass
else:
if (
getattr(obj, "battery_info", None)
is None
):
for (
_,
(other_obj, _other_listener),
) in self._entries.items():
if other_obj.isDevice:
continue
for child in other_obj:
child_ident = getattr(
child, "unitId", None
) or getattr(
child, "serial", None
)
if child_ident == ident:
obj.set_battery_info(
child.battery_info
)
break
if (
getattr(
obj,
"battery_info",
None,
)
is not None
):
break
with self._lock:
if dev_info.path not in self._entries:
self._entries[dev_info.path] = (obj, listener)
self._last_change[id(obj)] = time.time()
listener.start()
logger.info(
"listening on %s (%s)",
dev_info.path,
"device" if dev_info.isDevice else "receiver",
)
if "to_join" in locals():
try:
to_join.join(timeout=1.0)
except Exception:
logger.exception("error joining replaced listener")
del to_join
finally:
self._rescan_lock.release()
def known_devices(self):
with self._lock:
entries = list(self._entries.values())
result = []
seen_idents = set()
for obj, _listener in entries:
if obj.isDevice:
ident = obj.unitId or obj.serial
if ident and ident in seen_idents:
continue
if ident:
seen_idents.add(ident)
result.append(obj)
else:
for child in obj:
ident = child.unitId or child.serial
if ident and ident in seen_idents:
continue
if ident:
seen_idents.add(ident)
result.append(child)
return result
def stop(self):
with self._lock:
entries = list(self._entries.values())
for _obj, listener in entries:
try:
listener.stop()
except Exception:
logger.exception("error stopping listener")
for _obj, listener in entries:
listener.join(timeout=2.0)
def _snapshot(registry: _ListenerRegistry) -> list[DeviceBattery]:
results: list[DeviceBattery] = []
with registry._hidpp_io_lock:
for dev in registry.known_devices():
info = dev.battery_info
if info is None:
try:
if dev.ping():
info = dev.battery()
except Exception:
logger.debug(
"initial battery() failed for %s", dev, exc_info=True
)
info = None
entry = _to_device_battery(dev, info)
if entry is not None:
results.append(entry)
return results
def _clean_json(path: Path) -> None:
_write_json_atomic(path, [])
def _iter_open_devices():
try:
from logitech_receiver import base, device, receiver
except ImportError as e:
raise RuntimeError(
"logitech_receiver isn't importable -- install it with `pip install solaar`."
) from e
except ValueError as e:
raise RuntimeError(
"logitech_receiver failed to import because a GTK3 typelib is missing "
f"({e}). Install the same GTK3 + PyGObject packages Solaar's GUI needs "
"(e.g. python3-gi + gir1.2-gtk-3.0 on Debian/Ubuntu, or the gtk3/"
"python3-gobject equivalents on your distro)."
) from e
for dev_info in base.receivers_and_devices():
try:
if dev_info.isDevice:
d = device.create_device(base, dev_info)
else:
d = receiver.create_receiver(base, dev_info)
except OSError as e:
if e.errno == 13:
logger.error(
"permission denied opening %s -- check the Solaar udev rule "
"(rules.d/42-logitech-unify-permissions.rules) is installed "
"and you're in the right group",
dev_info.path,
)
else:
logger.exception("failed opening %s", dev_info)
continue
except Exception:
logger.exception("failed opening %s", dev_info)
continue
if d is None:
continue
if d.isDevice:
yield d
else:
yield from d
def _to_device_battery(dev, battery) -> DeviceBattery | None:
if battery is None:
return None
percentage = battery.level / 100 if isinstance(battery.level, int) else None
state = battery.status.name.lower() if battery.status is not None else None
serial = (
getattr(dev, "serial", None) or getattr(dev, "unitId", None) or None
)
return DeviceBattery(
model=dev.name or dev.codename or "Unknown device",
nativePath=_find_upower_native_path(serial),
serial=serial,
type=str(dev.kind) if dev.kind is not None else None,
wired=dev.receiver is None,
percentage=percentage,
state=state,
charging=battery.charging(),
isPresent=battery.ok(),
)
def _read_battery(dev) -> DeviceBattery | None:
try:
if not dev.ping():
return None
except Exception:
logger.debug("ping failed for %s", dev, exc_info=True)
return None
try:
battery = dev.battery()
except Exception:
logger.debug("battery() failed for %s", dev, exc_info=True)
return None
return _to_device_battery(dev, battery)
def poll_once() -> list[DeviceBattery]:
results: list[DeviceBattery] = []
seen_idents = set()
for dev in _iter_open_devices():
try:
info = _read_battery(dev)
if info is not None:
entry = _to_device_battery(dev, info)
if entry:
ident = entry.serial or (getattr(dev, "unitId", None))
if ident and ident in seen_idents:
continue
if ident:
seen_idents.add(ident)
results.append(entry)
finally:
with contextlib.suppress(Exception):
dev.close()
return results
def _write_json_atomic(path: Path, results: list[DeviceBattery]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
payload = {
"updated": time.time(),
"devices": [asdict(r) for r in results],
}
fd, tmp_path = tempfile.mkstemp(
dir=path.parent, prefix=".battery-", suffix=".tmp"
)
try:
with os.fdopen(fd, "w") as f:
json.dump(payload, f, indent=2)
os.replace(tmp_path, path)
except Exception:
Path(tmp_path).unlink(missing_ok=True)
raise
@app.command()
def daemon(
rescan_interval: float = typer.Option(
60.0,
"--rescan-interval",
"-i",
help=(
"Seconds between fallback rescans. This is NOT a battery poll "
"interval -- battery updates are event-driven and written as "
"soon as a device reports a change. New/removed devices "
"(e.g. plugging a keyboard in to charge) are normally noticed "
"within milliseconds via a udev hotplug watcher; this interval "
"is only a safety net in case a udev event is ever missed."
),
),
out: Path = typer.Option(
DEFAULT_OUTPUT,
"--out",
"-o",
help="Where to write the JSON status file.",
),
lock_file: Path = typer.Option(
DEFAULT_LOCK,
"--lock-file",
help="Path used to ensure only one daemon runs at a time.",
),
verbose: bool = typer.Option(
False,
"--verbose",
"-v",
help="Log every listener start and battery-change write to stderr.",
),
):
lock_fd = _acquire_singleton_lock(lock_file)
if lock_fd is None:
try:
holder_pid = lock_file.read_text().strip()
except OSError:
holder_pid = "unknown"
typer.echo(
f"error: a battery daemon is already running (pid {holder_pid}, lock: {lock_file})",
err=True,
)
raise typer.Exit(code=1)
if verbose:
logging.basicConfig(
level=logging.INFO, format="%(asctime)s %(name)s: %(message)s"
)
write_lock = threading.Lock()
last_payload = None
pending_write_timer: list[threading.Timer | None] = [None]
pending_write_lock = threading.Lock()
WRITE_SETTLE_SECONDS = 0.3
write_seq_lock = threading.Lock()
write_seq = 0
last_written_seq = 0
def write_if_changed():
nonlocal last_payload, write_seq, last_written_seq
with write_seq_lock:
write_seq += 1
my_seq = write_seq
results = _snapshot(registry)
with write_lock:
with write_seq_lock:
if my_seq <= last_written_seq:
return
last_written_seq = my_seq
payload = [asdict(r) for r in results]
if payload != last_payload:
_write_json_atomic(out, results)
last_payload = payload
logger.info(
"battery status changed, wrote %d device(s)", len(results)
)
def write_if_changed_coalesced():
with pending_write_lock:
existing = pending_write_timer[0]
if existing is not None:
existing.cancel()
timer = threading.Timer(WRITE_SETTLE_SECONDS, write_if_changed)
timer.daemon = True
pending_write_timer[0] = timer
timer.start()
def on_change(_device_or_receiver, alert=None, reason=None):
write_if_changed_coalesced()
registry = _ListenerRegistry(on_change)
def rescan_and_write():
try:
registry.rescan()
except Exception:
logger.exception("rescan failed")
write_if_changed_coalesced()
try:
registry.rescan()
except RuntimeError as e:
typer.echo(f"error: {e}", err=True)
raise typer.Exit(code=1) from None
write_if_changed()
hotplug_observer = _start_hotplug_watcher(rescan_and_write)
running = True
def _stop(signum, frame):
nonlocal running
running = False
signal.signal(signal.SIGINT, _stop)
signal.signal(signal.SIGTERM, _stop)
typer.echo(
f"Listening for battery events, writing to {out} (lock: {lock_file})"
)
try:
while running:
remaining = rescan_interval
while running and remaining > 0:
step = min(0.2, remaining)
time.sleep(step)
remaining -= step
if running:
rescan_and_write()
finally:
if hotplug_observer is not None:
with contextlib.suppress(Exception):
hotplug_observer.stop()
with contextlib.suppress(Exception):
_clean_json(out)
registry.stop()
typer.echo("Stopped.")
if __name__ == "__main__":
app()
+4 -3
View File
@@ -10,11 +10,12 @@ dependencies = [
"typer",
"pillow",
"jinja2",
"materialyoucolor",
"solaar",
"pytest"
"materialyoucolor"
]
[project.optional-dependencies]
dev = ["pytest"]
[project.scripts]
zshell-cli = "zshell:main"