Merge pull request #224 from bkbilly/Brightness

Wayland support for Brightness

Author: bkbilly

lnxlink/modules/brightness.py

@@ -1,8 +1,6 @@
 """Adjust display luminance globally or for individual monitors via number entities"""
-import re
 import logging
-from shutil import which
-from lnxlink.modules.scripts.helpers import syscommand, get_display_variable
+from lnxlink.modules.scripts.monitor_brightness import MonitorBrightness
 
 logger = logging.getLogger("lnxlink")
 
@@ -14,88 +12,54 @@ def __init__(self, lnxlink=None):
         """Setup addon"""
         self.name = "Brightness"
         self.lnxlink = lnxlink
-        self.display_variable = None
-        if which("xrandr") is None:
-            raise SystemError("System command 'xrandr' not found")
-        self.displays = self._get_displays()
+        self.monitors, issues = MonitorBrightness.list_displays()
+        for issue in issues:
+            logger.warning("Brightness Monitor Issue: %s", issue)
 
     def get_info(self):
         """Gather information from the system"""
-        displays = self._get_displays()
-        if displays != self.displays:
-            self.displays = displays
+        monitors, _ = MonitorBrightness.list_displays()
+        if monitors != self.monitors:
+            self.monitors = monitors
             self.lnxlink.setup_discovery("brightness")
-        values = [disp["brightness"] for disp in displays.values()]
-        avg_brightness = sum(values) / max(1, len(values))
-        avg_brightness = max(0.1, avg_brightness)
 
-        info = {"status": avg_brightness}
-        for display, values in displays.items():
-            info[display] = max(0.1, values["brightness"])
+        info = {}
+        for monitor in self.monitors:
+            monitor.get_brightness()
+            info[monitor.unique_name] = monitor.last_successful_read
+
         return info
 
     def exposed_controls(self):
         """Exposes to home assistant"""
         controls = {}
-        if len(self.displays) > 0:
-            controls = {
-                "Brightness": {
-                    "type": "number",
-                    "icon": "mdi:brightness-7",
-                    "min": 0.1,
-                    "max": 1,
-                    "step": 0.1,
-                    "value_template": "{{ value_json.status }}",
-                }
-            }
-        for display, values in self.displays.items():
-            controls[f"Brightness {values['name']}"] = {
+        for monitor in self.monitors:
+            controls[f"Brightness {monitor.unique_name}"] = {
                 "type": "number",
                 "icon": "mdi:brightness-7",
-                "min": 0.1,
-                "max": 1,
-                "step": 0.1,
-                "value_template": f"{{{{ value_json.{display} }}}}",
-                "enabled": False,
+                "min": 0,
+                "max": 100,
+                "step": 1,
+                "value_template": f"{{{{ value_json.get('{monitor.unique_name}') }}}}",
             }
         return controls
 
     def start_control(self, topic, data):
         """Control system"""
-        disp_env_cmd = ""
-        if self.display_variable is not None:
-            disp_env_cmd = f" --display {self.display_variable}"
-
         if topic[1] == "brightness":
-            for values in self.displays.values():
-                syscommand(
-                    f"xrandr --output {values['name']} --brightness {data} {disp_env_cmd}"
-                )
+            for monitor in self.monitors:
+                monitor.set_brightness(int(data))
         else:
-            display = self.displays[
-                topic[1].replace("brightness_", "").replace("-", "_")
-            ]["name"]
-            syscommand(f"xrandr --output {display} --brightness {data} {disp_env_cmd}")
-
-    def _get_displays(self):
-        """Get all the displays"""
-        self.display_variable = get_display_variable()
-        displays = {}
-        disp_env_cmd = ""
-        if self.display_variable is not None:
-            disp_env_cmd = f" --display {self.display_variable}"
-
-        stdout, _, _ = syscommand(
-            f"xrandr --verbose --current {disp_env_cmd}",
-        )
-        pattern = re.compile(
-            r"(\S+) \bconnected\b.*[\s\S]*?(?=Brightness)Brightness: ([\d\.\d]+)"
-        )
-
-        for match in pattern.findall(stdout):
-            displays[match[0].replace("-", "_").lower()] = {
-                "name": match[0],
-                "brightness": float(match[1]),
-            }
-
-        return displays
+            for monitor in self.monitors:
+                name_query = topic[1].replace("brightness_", "").replace("-", "_")
+                monitor_unique_name = (
+                    monitor.unique_name.lower().replace("-", "_").replace(" ", "_")
+                )
+                if monitor_unique_name == name_query:
+                    logger.info(
+                        "Changing Brightness to %d for %s",
+                        int(data),
+                        monitor.unique_name,
+                    )
+                    monitor.set_brightness(int(data))
+                    break

lnxlink/modules/scripts/monitor_brightness.py

@@ -0,0 +1,227 @@
+"""
+Module to read and set monitor brightness via I2C/DDC and sysfs.
+Based on the work from: https://github.com/Crozzers/screen_brightness_control
+Copyright (c) 2025 Crozzers (https://github.com/Crozzers)
+"""
+
+import os
+import glob
+import time
+import fcntl
+import struct
+import functools
+import operator
+from typing import Optional, List, Set
+
+
+class MonitorBrightness:
+    """Base class for all monitor types."""
+
+    def __init__(self, identifier: str, manufacturer: str, name: str):
+        self.identifier = identifier  # Bus path or sysfs path
+        self.manufacturer = manufacturer
+        self.name = name
+        self.unique_name = f"{manufacturer} {name}"
+        self.last_successful_read = None
+
+    def get_brightness(self, timeout: float = 2.0) -> Optional[int]:
+        """Polls for brightness until timeout. Returns None if max_val is 0."""
+        raise NotImplementedError
+
+    def set_brightness(self, value: int) -> None:
+        """Sets brightness level. Returns True on success."""
+        raise NotImplementedError
+
+    @staticmethod
+    def list_displays() -> tuple[List["MonitorBrightness"], Set[str]]:
+        """Scans I2C buses and /sys/class/backlight to return all monitors."""
+        found_displays = []
+        issues = set()
+
+        # 1. Internal Laptop Displays (Sysfs)
+        for path in glob.glob("/sys/class/backlight/*"):
+            found_displays.append(SysfsMonitor(path))
+            if not os.access(os.path.join(path, "brightness"), os.W_OK):
+                issues.add(
+                    "Backlight Permission Error: Create a udev rule.\n"
+                    'Run: echo \'SUBSYSTEM=="backlight",RUN+="/bin/chmod 666" '
+                    "/sys/class/backlight/%k/brightness /sys/class/backlight/%k/bl_power\"'"
+                    " | sudo tee -a /etc/udev/rules.d/backlight-permissions.rules\n"
+                    "Then run: sudo udevadm control --reload-rules && udevadm trigger"
+                )
+            found_displays.append(SysfsMonitor(path))
+
+        # 2. External Monitors (I2C/DDC)
+        for i2c_path in sorted(glob.glob("/dev/i2c-*")):
+            try:
+                fd = os.open(i2c_path, os.O_RDWR)
+                # Use EDID address to identify the monitor
+                fcntl.ioctl(fd, 0x0703, 0x50)
+                data = os.read(fd, 256)
+                os.close(fd)
+
+                if data.startswith(bytes.fromhex("00 FF FF FF FF FF FF 00")):
+                    mfg, name, _ = DDCIPMonitor.parse_edid(data)
+                    found_displays.append(DDCIPMonitor(i2c_path, mfg, name))
+            except PermissionError:
+                issues.add(
+                    f"I2C Permission Error: User '{os.getlogin()}' needs 'i2c' group access.\n"
+                    f"Run: sudo usermod -aG i2c {os.getlogin()} && reboot"
+                )
+            except Exception:
+                continue
+
+        return found_displays, issues
+
+    def __repr__(self):
+        return (
+            f"<MonitorBrightness {self.manufacturer} {self.name} on {self.identifier}>"
+        )
+
+
+class SysfsMonitor(MonitorBrightness):
+    """Handles laptop backlights via /sys/class/backlight."""
+
+    def __init__(self, path: str):
+        name = os.path.basename(path)
+        super().__init__(path, "Internal", name)
+
+    def get_brightness(self, timeout: float = 1.0) -> Optional[int]:
+        try:
+            with open(
+                os.path.join(self.identifier, "brightness"), "r", encoding="UTF-8"
+            ) as f:
+                cur_val = int(f.read().strip())
+            with open(
+                os.path.join(self.identifier, "max_brightness"), "r", encoding="UTF-8"
+            ) as f:
+                max_val = int(f.read().strip())
+
+            if max_val == 0:
+                return None
+            self.last_successful_read = int((cur_val / max_val) * 100)
+            return self.last_successful_read
+        except (OSError, ValueError):
+            return None
+
+    def set_brightness(self, value: int) -> None:
+        try:
+            target_pct = max(0, min(100, int(value)))
+            with open(
+                os.path.join(self.identifier, "max_brightness"), "r", encoding="UTF-8"
+            ) as f:
+                max_val = int(f.read().strip())
+
+            actual_value = int((target_pct / 100) * max_val)
+            with open(
+                os.path.join(self.identifier, "brightness"), "w", encoding="UTF-8"
+            ) as f:
+                f.write(str(actual_value))
+        except (OSError, PermissionError):
+            print("Error setting brightness for", self)
+
+
+class DDCIPMonitor(MonitorBrightness):
+    """Handles external monitors via I2C/DDC logic."""
+
+    I2C_SLAVE = 0x0703
+    DDC_ADDR = 0x37
+    HOST_ADDR_W = 0x51
+    DEST_ADDR_W = 0x6E
+    GET_VCP_CMD = 0x01
+    SET_VCP_CMD = 0x03
+    VCP_BRIGHTNESS = 0x10
+
+    @staticmethod
+    def parse_edid(data: bytes):
+        """Parses manufacturer and name from EDID block."""
+        try:
+            m_bytes = struct.unpack(">H", data[8:10])[0]
+            manufacturer = "".join(
+                [
+                    chr(((m_bytes >> 10) & 31) + 64),
+                    chr(((m_bytes >> 5) & 31) + 64),
+                    chr((m_bytes & 31) + 64),
+                ]
+            )
+            name, serial = "Unknown", "Unknown"
+            for position_from in range(54, 109, 18):
+                position_to = position_from + 18
+                block = data[position_from:position_to]
+                if block[0:4] == b"\x00\x00\x00\xfc":
+                    name = (
+                        block[5:]
+                        .split(b"\x0a")[0]
+                        .decode("ascii", errors="ignore")
+                        .strip()
+                    )
+                elif block[0:4] == b"\x00\x00\x00\xff":
+                    serial = (
+                        block[5:]
+                        .split(b"\x0a")[0]
+                        .decode("ascii", errors="ignore")
+                        .strip()
+                    )
+            return manufacturer, name, serial
+        except Exception:
+            return "Unknown", "Unknown", "Unknown"
+
+    def _ddc_command(self, payload: list, read_length: int = 0) -> Optional[bytes]:
+        """Low-level I2C write/read transaction."""
+        fd = None
+        try:
+            fd = os.open(self.identifier, os.O_RDWR)
+            fcntl.ioctl(fd, self.I2C_SLAVE, self.DDC_ADDR)
+
+            # Construct DDC packet with checksum
+            packet = bytearray([self.HOST_ADDR_W, len(payload) | 0x80] + payload)
+            checksum = functools.reduce(operator.xor, packet, self.DEST_ADDR_W)
+            packet.append(checksum)
+
+            os.write(fd, packet)
+            time.sleep(0.05)
+
+            if read_length > 0:
+                return os.read(fd, read_length)
+            return None
+        except (OSError, IOError):
+            return None
+        finally:
+            if fd is not None:
+                os.close(fd)
+
+    def get_brightness(self, timeout: float = 2.0) -> Optional[int]:
+        """Polls for external brightness via VCP."""
+        start_time = time.monotonic()
+        while (time.monotonic() - start_time) < timeout:
+            data = self._ddc_command([self.GET_VCP_CMD, self.VCP_BRIGHTNESS], 11)
+            # DDC reply: [source, length, result, code, max_h, max_l, cur_h, cur_l, checksum]
+            if data and len(data) >= 10:
+                max_val = int.from_bytes(data[6:8], "big")
+                cur_val = int.from_bytes(data[8:10], "big")
+
+                if max_val != 0:
+                    self.last_successful_read = int((cur_val / max_val) * 100)
+                    return self.last_successful_read
+
+            time.sleep(0.1)  # Retry interval
+        return None
+
+    def set_brightness(self, value: int, retries: int = 2) -> None:
+        """Sets brightness and verifies the hardware change."""
+        target = max(0, min(100, int(value)))
+        # Send set command
+        for _attempt in range(retries):
+            self._ddc_command([self.SET_VCP_CMD, self.VCP_BRIGHTNESS, 0x00, target])
+
+
+if __name__ == "__main__":
+    monitors, myissues = MonitorBrightness.list_displays()
+    for issue in myissues:
+        print("Issue:", issue)
+        print("---------")
+    for mon in monitors:
+        print(f"Checking {mon}...")
+        current = mon.get_brightness()
+        print(f"Current Brightness: {current}%")
+        mon.set_brightness(20)