Merge pull request #45 from tiagocoutinho/gpio

Add modprobe info to github actions

.github/workflows/ci.yml

@@ -14,7 +14,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
 
     steps:
     - name: Checkout
@@ -24,16 +24,20 @@ jobs:
         sudo apt-get install -y linux-image-$(uname -r) linux-modules-extra-$(uname -r)
     - name: Setup user groups
       run: |
-        echo KERNEL==\"uinput\", SUBSYSTEM==\"misc\" GROUP=\"docker\", MODE:=\"0666\" | sudo tee /etc/udev/rules.d/99-$USER.rules
-        echo KERNEL==\"event[0-9]*\", SUBSYSTEM==\"input\" GROUP=\"docker\", MODE:=\"0666\" | sudo tee -a /etc/udev/rules.d/99-$USER.rules
-        echo SUBSYSTEM==\"video4linux\" GROUP=\"docker\", MODE:=\"0666\" | sudo tee -a /etc/udev/rules.d/99-$USER.rules
-        cat /etc/udev/rules.d/99-$USER.rules
+        echo KERNEL==\"uinput\", SUBSYSTEM==\"misc\" GROUP=\"docker\", MODE=\"0666\" | sudo tee /etc/udev/rules.d/99-$USER.rules
+        echo KERNEL==\"event[0-9]*\", SUBSYSTEM==\"input\" GROUP=\"docker\", MODE=\"0666\" | sudo tee -a /etc/udev/rules.d/99-$USER.rules
+        echo SUBSYSTEM==\"video4linux\" GROUP=\"docker\", MODE=\"0666\" | sudo tee -a /etc/udev/rules.d/99-$USER.rules
+        echo KERNEL==\"gpiochip[0-9]*\", SUBSYSTEM==\"gpio\", GROUP=\"docker\", MODE=\"0666\" | sudo tee -a /etc/udev/rules.d/99-$USER.rules
         sudo udevadm control --reload-rules
         sudo udevadm trigger
         sudo modprobe -a uinput
         sudo modprobe vivid n_devs=1 node_types=0xe1d3d vid_cap_nr=190 vid_out_nr=191 meta_cap_nr=192 meta_out_nr=193
-        lsmod
-        ls -lsa /dev/video*
+        sudo modprobe gpio-sim
+        sudo modprobe gpio-aggregator
+        sudo python scripts/setup-gpio-sim.py
+        ls -lsa /dev/video* /dev/uinput /dev/gpio* /dev/inp*
+
+
     - name: Set up Python ${{ matrix.python-version }}
       id: setuppy
       uses: actions/setup-python@v5

linuxpy/configfs.py

@@ -0,0 +1,13 @@
+#
+# This file is part of the linuxpy project
+#
+# Copyright (c) 2024 Tiago Coutinho
+# Distributed under the GPLv3 license. See LICENSE for more info.
+
+from pathlib import Path
+
+from linuxpy.types import Optional
+
+from .mounts import configfs
+
+CONFIGFS_PATH: Optional[Path] = configfs()

linuxpy/gpio/device.py

@@ -131,6 +131,7 @@ def set_values(req_fd: FDLike, mask: int, bits: int) -> raw.gpio_v2_line_values:
         bits: The bits is a bitmap containing the value of the lines, set to 1 for active and 0
               for inactive.
     """
+    print(mask, bits)
     result = raw.gpio_v2_line_values(mask=mask, bits=bits)
     return ioctl(req_fd, IOC.LINE_SET_VALUES, result)
 
@@ -182,27 +183,27 @@ def __init__(
         self.lines = lines
         self.blocking = blocking
         self.indexes = {line: index for index, line in enumerate(lines)}
-        self.fd = None
+        self.fd = -1
         super().__init__()
 
-    def fileno(self):
+    def fileno(self) -> int:
         return self.fd
 
     def close(self):
-        if self.fd is None:
+        if self.fd < 0:
             return
         os.close(self.fd)
-        self.fd = None
+        self.fd = -1
 
     def open(self):
-        self.fd = request_line(self.device, self.name, self.lines, self.flags, self.blocking).fd
+        self.fd: int = request_line(self.device, self.name, self.lines, self.flags, self.blocking).fd
 
     def get_values(self, lines: Collection[int]) -> dict[int, int]:
         mask = functools.reduce(operator.or_, (1 << self.indexes[line] for line in lines), 0)
         values = get_values(self, mask)
         return {line: (values.bits >> self.indexes[line]) & 1 for line in lines}
 
-    def set_values(self, values: dict[int, Union[int, bool]]):
+    def set_values(self, values: dict[int, Union[int, bool]]) -> raw.gpio_v2_line_values:
         mask, bits = 0, 0
         for line, value in values.items():
             index = self.indexes[line]
@@ -234,7 +235,7 @@ def __getitem__(self, key: Union[int, tuple, slice]) -> Union[int, dict]:
         return self.get_values(lines)
 
     def __setitem__(self, key: Union[int, tuple, slice], value: Union[int, Sequence[int]]):
-        if isinstance(key, int):
+        if isinstance(key, int) and isinstance(value, int):
             values = {key: value}
         else:
             lines = expand_from_list(key, self.min_line, self.max_line + 1)

linuxpy/gpio/sim.py

@@ -0,0 +1,47 @@
+#
+# This file is part of the linuxpy project
+#
+# Copyright (c) 2024 Tiago Coutinho
+# Distributed under the GPLv3 license. See LICENSE for more info.
+
+"""
+## Configfs GPIO Simulator
+
+The configfs GPIO Simulator (gpio-sim) provides a way to create simulated GPIO chips
+for testing purposes. The lines exposed by these chips can be accessed using the
+standard GPIO character device interface as well as manipulated using sysfs attributes.
+
+To be able to create GPIOSim chips, the user will need permissions.
+
+Here is an example of what to put in /etc/udev/rules.d/80-gpio-sim.rules that gives
+access to users belonging to the `input` group:
+
+```
+KERNEL=="gpio-sim", SUBSYSTEM=="config", RUN+="/bin/chown -R root:input /sys/kernel/config/gpio-sim"
+KERNEL=="gpio-sim", SUBSYSTEM=="config", RUN+="/bin/chmod -R 775 /sys/kernel/config/gpio-sim"
+```
+
+See https://www.kernel.org/doc/html/latest/admin-guide/gpio/gpio-sim.html for details
+"""
+
+from pathlib import Path
+
+from linuxpy.configfs import CONFIGFS_PATH
+from linuxpy.gpio.device import get_chip_info
+from linuxpy.types import Optional
+
+GPIOSIM_PATH: Optional[Path] = None if CONFIGFS_PATH is None else CONFIGFS_PATH / "gpio-sim"
+
+
+def find_gpio_sim_file(num_lines=None) -> Optional[Path]:
+    """Best effort to find
+
+    Returns:
+        _type_: _description_
+    """
+    for path in sorted(Path("/dev").glob("gpiochip*")):
+        with path.open("rb") as fobj:
+            info = get_chip_info(fobj)
+            if "gpio-sim" in info.label:
+                if num_lines is None or info.lines == num_lines:
+                    return path

linuxpy/mounts.py

@@ -0,0 +1,49 @@
+#
+# This file is part of the linuxpy project
+#
+# Copyright (c) 2024 Tiago Coutinho
+# Distributed under the GPLv3 license. See LICENSE for more info.
+
+import functools
+from pathlib import Path
+
+from .proc import PROC_PATH
+from .types import Generator, NamedTuple, Optional
+
+MOUNTS_PATH: Path = PROC_PATH / "mounts"
+
+
+class MountInfo(NamedTuple):
+    dev_type: str
+    mount_point: str
+    fs_type: str
+    attrs: list[str]
+
+
+def gen_read() -> Generator[MountInfo, None, None]:
+    data = MOUNTS_PATH.read_text()
+    for line in data.splitlines():
+        dev_type, mount_point, fs_type, attrs, *_ = line.split()
+        yield MountInfo(dev_type, mount_point, fs_type, attrs.split(","))
+
+
+@functools.cache
+def cache() -> tuple[MountInfo, ...]:
+    return tuple(gen_read())
+
+
+@functools.cache
+def get_mount_point(dev_type, fs_type=None) -> Optional[Path]:
+    if fs_type is None:
+        fs_type = dev_type
+    for _dev_type, mount_point, _fs_type, *_ in cache():
+        if dev_type == _dev_type and fs_type == _fs_type:
+            return Path(mount_point)
+
+
+def sysfs() -> Optional[Path]:
+    return get_mount_point("sysfs")
+
+
+def configfs() -> Optional[Path]:
+    return get_mount_point("configfs")

linuxpy/proc.py

@@ -0,0 +1,50 @@
+from pathlib import Path
+
+from linuxpy.util import try_numeric
+
+PROC_PATH = Path("/proc")
+
+CPU_INFO_PATH: Path = PROC_PATH / "cpuinfo"
+MEM_INFO_PATH: Path = PROC_PATH / "meminfo"
+MODULES_PATH: Path = PROC_PATH / "modules"
+
+
+def iter_cpu_info():
+    data = CPU_INFO_PATH.read_text()
+    for cpu in data.split("\n\n"):
+        info = {}
+        for line in cpu.splitlines():
+            key, value = map(str.strip, line.split(":", 1))
+            if "flags" in key or key == "bugs":
+                value = value.split()
+            else:
+                value = try_numeric(value)
+            info[key] = value
+        yield info
+
+
+def iter_mem_info():
+    data = MEM_INFO_PATH.read_text()
+    for line in data.splitlines():
+        key, value = map(str.strip, line.split(":", 1))
+        if value.endswith(" kB"):
+            value = try_numeric(value[:-3]) * 1024
+        else:
+            value = try_numeric(value)
+        yield key, value
+
+
+def iter_modules():
+    data = MODULES_PATH.read_text()
+    for line in data.splitlines():
+        fields = line.split()
+        mod = {
+            "name": fields[0],
+            "size": int(fields[1]),
+            "use_count": int(fields[2]),
+            "dependencies": [] if fields[3] == "-" else [dep for dep in fields[3].split(",") if dep],
+        }
+        if len(fields) > 5:
+            mod["state"] = fields[4]
+            mod["offset"] = int(fields[5], 16)
+        yield mod

linuxpy/sysfs.py

@@ -12,13 +12,14 @@
 from linuxpy.types import Callable
 from linuxpy.util import make_find
 
+from . import mounts
 from .magic import Magic
 from .statfs import get_fs_type
 from .types import Iterable, Optional
 
 MAGIC = Magic.SYSFS
 
-MOUNT_PATH = pathlib.Path("/sys")
+MOUNT_PATH = mounts.sysfs()
 DEVICE_PATH = MOUNT_PATH / "bus/usb/devices"
 CLASS_PATH = MOUNT_PATH / "class"
 THERMAL_PATH = CLASS_PATH / "thermal"

linuxpy/types.py

@@ -51,6 +51,6 @@ def fileno(self) -> int: ...
 Callable = collections.abc.Callable
 Sequence = collections.abc.Sequence
 Collection = collections.abc.Collection
-
+NamedTuple = typing.NamedTuple
 
 T = typing.TypeVar("T")

linuxpy/util.py

@@ -8,6 +8,7 @@
 
 import asyncio
 import contextlib
+import functools
 import random
 import selectors
 import string
@@ -78,6 +79,28 @@ def to_fd(fd: FDLike):
     return fd
 
 
+int16 = functools.partial(int, base=16)
+
+
+def try_numeric(text: str):
+    """
+    Try to translate given text into int, int base 16 or float.
+    Returns the orig and return the original text if it fails.
+
+    Args:
+        text (str): text to be translated
+
+    Returns:
+        int, float or str: The converted text
+    """
+    for func in (int, int16, float):
+        try:
+            return func(text)
+        except ValueError:
+            pass
+    return text
+
+
 @contextlib.contextmanager
 def add_reader_asyncio(fd: FDLike, callback: Callable, *args, loop: Optional[asyncio.AbstractEventLoop] = None):
     """Add reader during the context and remove it after"""

pyproject.toml

@@ -23,6 +23,7 @@ classifiers = [
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "Topic :: Multimedia :: Video",
     "Topic :: Multimedia :: Video :: Capture",
 ]
@@ -91,8 +92,8 @@ show-slowest = 3
 branch = true
 omit = ["*test*"]
 report_type = ["html", "json", "xml"]
-report = {skip_empty = true, show_missing = true}
-#html = {directory = "docs/htmlcov"}
+#report = {skip_empty = true, show_missing = true}
+
 
 [tool.ruff]
 lint.select = [