fegger
4a47277c86
Add systemd service to bridge the Pi Unix socket over SSH, make fan.py fall back to the TCP tunnel, and rename the module from `custom/fan` to `custom/cooling_pad`. Update README install steps and connection details.
140 lines
4.1 KiB
Python
Executable File
140 lines
4.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Waybar widget for the cooling-pad fan controller.
|
|
|
|
Talks to fan_controller over its Unix socket (/tmp/fan_controller.sock).
|
|
|
|
Usage:
|
|
fan.py print status as Waybar JSON (return-type: "json")
|
|
fan.py cycle auto -> 25% -> 50% -> 75% -> 100% -> auto (on-click)
|
|
fan.py auto switch back to auto mode (on-right-click)
|
|
fan.py duty 75 set manual duty, 0-100
|
|
fan.py setpoint 55 set auto-mode setpoint, 35-85
|
|
|
|
Connection: the daemon's command socket lives on the Pi. This script uses the
|
|
local Unix socket if present, otherwise a TCP tunnel to it (default
|
|
127.0.0.1:10250, provided by the fan-tunnel systemd service:
|
|
ssh -N -L 10250:/tmp/fan_controller.sock root@10.55.0.1).
|
|
Override with FAN_CONTROLLER_ENDPOINT: "unix:/path" or "host:port".
|
|
"""
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
|
|
UNIX_PATH = "/tmp/fan_controller.sock"
|
|
TUNNEL_HOST, TUNNEL_PORT = "127.0.0.1", 10250
|
|
STEPS = [25, 50, 75, 100]
|
|
|
|
|
|
def make_socket():
|
|
endpoint = os.environ.get("FAN_CONTROLLER_ENDPOINT")
|
|
if endpoint is None:
|
|
if os.path.exists(UNIX_PATH):
|
|
endpoint = f"unix:{UNIX_PATH}"
|
|
else:
|
|
endpoint = f"{TUNNEL_HOST}:{TUNNEL_PORT}"
|
|
if endpoint.startswith("unix:"):
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
s.connect(endpoint[len("unix:"):])
|
|
else:
|
|
host, port = endpoint.rsplit(":", 1)
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.connect((host, int(port)))
|
|
return s
|
|
|
|
|
|
def cmd(command: str) -> str:
|
|
with make_socket() as s:
|
|
s.settimeout(2)
|
|
s.sendall(command.encode() + b"\n")
|
|
chunks = []
|
|
while True:
|
|
data = s.recv(1024)
|
|
if not data:
|
|
break
|
|
chunks.append(data)
|
|
return b"".join(chunks).decode().strip()
|
|
|
|
|
|
def status():
|
|
return json.loads(cmd("STATUS"))
|
|
|
|
|
|
def emit(text, tooltip=None, cls=None, alt=None):
|
|
out = {"text": text}
|
|
if tooltip:
|
|
out["tooltip"] = tooltip
|
|
if cls:
|
|
out["class"] = " ".join(cls)
|
|
if alt:
|
|
out["alt"] = alt
|
|
print(json.dumps(out))
|
|
|
|
|
|
def main():
|
|
arg = sys.argv[1] if len(sys.argv) > 1 else None
|
|
try:
|
|
if arg is None:
|
|
st = status()
|
|
mode = st.get("mode", "?")
|
|
if mode == "auto":
|
|
label = f"auto {st['setpoint']:.0f}°"
|
|
else:
|
|
label = f"{st['duty']:.0f}%"
|
|
text = f"{st['temp']:.1f}° {label}"
|
|
tooltip = (
|
|
f"Temp: {st['temp']:.1f}°C\n"
|
|
f"Fan: {st['duty']:.0f}% / {st['rpm']} rpm\n"
|
|
f"Mode: {mode}"
|
|
)
|
|
if mode == "auto":
|
|
tooltip += f" (setpoint {st['setpoint']:.0f}°C)"
|
|
if st.get("safe"):
|
|
tooltip += "\nSafe mode: ON"
|
|
cls = ["fan"]
|
|
if st.get("safe"):
|
|
cls.append("safe")
|
|
emit(text, tooltip, cls, alt=f"{st['rpm']} rpm")
|
|
|
|
elif arg == "cycle":
|
|
st = status()
|
|
if st["mode"] == "manual":
|
|
d = float(st["duty"])
|
|
nxt = next((x for x in STEPS if x > d + 0.5), None)
|
|
if nxt is None:
|
|
cmd("SET_MODE auto")
|
|
else:
|
|
cmd(f"SET_DUTY {nxt}")
|
|
else:
|
|
cmd("SET_MODE manual")
|
|
cmd(f"SET_DUTY {STEPS[0]}")
|
|
|
|
elif arg == "auto":
|
|
cmd("SET_MODE auto")
|
|
|
|
elif arg == "duty":
|
|
d = float(sys.argv[2])
|
|
if not 0 <= d <= 100:
|
|
sys.exit("duty must be 0-100")
|
|
cmd("SET_MODE manual")
|
|
cmd(f"SET_DUTY {d:.0f}")
|
|
|
|
elif arg == "setpoint":
|
|
sp = float(sys.argv[2])
|
|
if not 35 <= sp <= 85:
|
|
sys.exit("setpoint must be 35-85")
|
|
cmd(f"SET_SETPOINT {sp:.1f}")
|
|
|
|
else:
|
|
sys.exit(__doc__)
|
|
|
|
except (OSError, ValueError) as e:
|
|
if arg is None:
|
|
emit("fan: offline", "fan_controller not reachable", ["fan", "offline"])
|
|
else:
|
|
sys.exit(f"error: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|