add proof of concept client

.gitignore

@@ -4,3 +4,4 @@
 sensor/build
 sensor/sensor.*
 sensor/compile_commands.json
+__pycache__

client/brightness.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python
+
+import lux
+import ddc
+import time
+
+filter_fact = 0.9
+
+# out = out_b + out_a * in_a * (in_b + in)
+c_in = 1.0, -40.0, # in_a, in_b
+calibration = {
+    "HPN:HP 27xq:CNK1072BJY": [
+        1.0, 30.0, # out_a, out_b
+    ],
+
+    "MSI:MSI G27CQ4:": [
+        1.0, 20.0, # out_a, out_b
+    ],
+}
+
+def cal(v, c):
+    return c[1] + c[0] * c_in[0] * (c_in[1] + v)
+
+def filter_lux(old, new):
+    return (old * filter_fact) + (new * (1.0 - filter_fact))
+
+def lux_to_disp(name, val):
+    if name in calibration:
+        val = cal(int(val), calibration[name])
+    else:
+        raise ValueError("no calibration for \"{}\"".format(name))
+    val = int(val)
+    return min(max(val, 0), 100)
+
+if __name__ == "__main__":
+    usb = lux.usb_init()
+    lux.check_connection(usb)
+
+    disps = ddc.ddc_detect()
+    brightness = lux.read_brightness(usb)
+    print("Brightness:", brightness)
+
+    for d in disps:
+        d["prev"] = ddc.ddc_get(d["id"])
+        print("Display \"{}\" at {}".format(d["name"], d["prev"]))
+
+    print()
+    print("Starting main loop")
+    print()
+
+    while True:
+        brightness = filter_lux(brightness, lux.read_brightness(usb))
+
+        for d in disps:
+            val = lux_to_disp(d["name"], brightness)
+            if val != d["prev"]:
+                d["prev"] = val
+                print("Setting \"{}\" to {}".format(d["name"], val))
+                ddc.ddc_set(d["id"], val)
+
+        time.sleep(1.0)

client/ddc.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+
+import subprocess
+
+def ddc_detect():
+    r = subprocess.run(["ddcutil", "detect", "-t"], capture_output=True)
+    if r.returncode != 0:
+        raise ValueError("ddcutil returned {}".format(r.returncode))
+
+    out = []
+
+    displays = r.stdout.decode("utf-8").split("\n\n")
+    for disp in displays:
+        data = disp.split("\n")
+        field = {}
+
+        if len(data) < 4:
+            continue
+
+        # Display X
+        num = data[0].split()
+        if num[0] != "Display":
+            #raise ValueError("unexpected identifier (\"{}\" != \"Display\")".format(num[0]))
+            continue
+        field["id"] = num[1]
+
+        # Monitor: name ...
+        name = data[3].split()
+        if name[0] != "Monitor:":
+            #raise ValueError("unexpected identifier (\"{}\" != \"Monitor:\")".format(name[0]))
+            continue
+        field["name"] = ' '.join(name[1:])
+
+        out.append(field)
+
+    return out
+
+def ddc_get(dev):
+    r = subprocess.run(["ddcutil", "-d", str(dev), "-t", "getvcp", "10"], capture_output=True)
+    if r.returncode != 0:
+        raise ValueError("ddcutil returned {}".format(r.returncode))
+
+    s = r.stdout.decode("utf-8").split()
+    if (s[0] != "VCP") or (s[1] != "10") or (s[2] != "C") or (s[4] != "100"):
+        raise ValueError("unexpected identifier")
+
+    return int(s[3])
+
+def ddc_set(dev, val):
+    val = int(val)
+    if (val < 0) or (val > 100):
+        raise ValueError("out of range")
+
+    r = subprocess.run(["ddcutil", "-d", str(dev), "-t", "setvcp", "10", str(val)], capture_output=True)
+    if r.returncode != 0:
+        raise ValueError("ddcutil returned {}".format(r.returncode))
+
+if __name__ == "__main__":
+    devs = ddc_detect()
+    for dev in devs:
+        print("Display:", dev["id"], dev["name"])
+
+        b = ddc_get(dev["id"])
+        print("Brightness:", b)
+
+        new = 50
+        if b == 50:
+            new = 60
+        print("Setting to {}...".format(new))
+        ddc_set(dev["id"], new)
+
+        b = ddc_get(dev["id"])
+        print("Brightness:", b)
+
+        print()

client/fetch.py

@@ -1,45 +0,0 @@
-#!/usr/bin/env python
-
-import usb.core
-import usb.util
-import time
-
-CUSTOM_RQ_ECHO = 0 # send back wValue and wIndex, for testing comms reliability
-CUSTOM_RQ_RESET = 1 # reset to bootloader
-CUSTOM_RQ_GET = 2 # get ldr value
-
-max_comm_retries = 5;
-
-def is_target_device(dev):
-    if dev.manufacturer == "xythobuz.de" and dev.product == "AutoBrightness":
-        return True
-    return False
-
-# find our device
-dev = usb.core.find(idVendor=0x16c0, idProduct=0x05dc, custom_match=is_target_device)
-
-# was it found?
-if dev is None:
-    raise ValueError('Device not found')
-
-# configure the device
-dev.set_configuration()
-
-def read_val(req, w, a=0, b=0):
-    for attempts in range(0, max_comm_retries):
-        try:
-            r = dev.ctrl_transfer(usb.util.CTRL_TYPE_VENDOR | usb.util.CTRL_IN, req, a, b, w)
-            return int.from_bytes(r, "little")
-        except usb.core.USBError as e:
-            if attempts >= (max_comm_retries - 1):
-                raise e
-
-# check for proper connectivity
-if read_val(CUSTOM_RQ_ECHO, 4, 42, 23) != (42 | (23 << 16)):
-    raise ValueError("invalid echo response")
-
-# repeatedly print value
-while True:
-    v = read_val(CUSTOM_RQ_GET, 2)
-    print(time.time(), ";", v, flush=True)
-    time.sleep(0.25)

client/lux.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+
+import usb.core
+import usb.util
+import time
+
+CUSTOM_RQ_ECHO = 0 # send back wValue and wIndex, for testing comms reliability
+CUSTOM_RQ_RESET = 1 # reset to bootloader
+CUSTOM_RQ_GET = 2 # get ldr value
+
+max_comm_retries = 5;
+
+def is_target_device(dev):
+    if dev.manufacturer == "xythobuz.de" and dev.product == "AutoBrightness":
+        return True
+    return False
+
+def usb_init():
+    # find our device
+    dev = usb.core.find(idVendor=0x16c0, idProduct=0x05dc, custom_match=is_target_device)
+
+    # was it found?
+    if dev is None:
+        raise ValueError('Device not found')
+
+    # configure the device
+    dev.set_configuration()
+
+    return dev
+
+def read_val(dev, req, w, a=0, b=0):
+    for attempts in range(0, max_comm_retries):
+        try:
+            r = dev.ctrl_transfer(usb.util.CTRL_TYPE_VENDOR | usb.util.CTRL_IN, req, a, b, w)
+            return int.from_bytes(r, "little")
+        except usb.core.USBError as e:
+            if attempts >= (max_comm_retries - 1):
+                raise e
+
+def check_connection(dev):
+    # check for proper connectivity
+    if read_val(dev, CUSTOM_RQ_ECHO, 4, 42, 23) != (42 | (23 << 16)):
+        raise ValueError("invalid echo response")
+
+def read_brightness(dev):
+    return read_val(dev, CUSTOM_RQ_GET, 2)
+
+if __name__ == "__main__":
+    dev = usb_init()
+    check_connection(dev)
+
+    # repeatedly print value
+    while True:
+        v = read_brightness(dev)
+        print(time.time(), ";", v, flush=True)
+        time.sleep(0.25)

client/plot.sh

@@ -5,4 +5,4 @@
 # enter script directory
 cd "$(dirname "$0")"
 
-./fetch.py | ttyplot -t "Brightness"
+./lux.py | ttyplot -t "Brightness"