#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: 2024 Bartosz Golaszewski <bartosz.golaszewski@linaro.org>
# SPDX-FileCopyrightText: 2026 Qualcomm Technologies, Inc. and/or its subsidiaries

import errno
import os
import re
import select
import stat
import sys
import time
import traceback
from argparse import ArgumentParser
from importlib.metadata import PackageNotFoundError, version
from threading import Lock, Thread

try:
    __version__ = version("gpiod-sysfs-proxy")
except PackageNotFoundError:
    __version__ = "unknown"

import pyfuse3
import trio
import gpiod
import pyudev

from pyfuse3 import (
    EntryAttributes,
    FileHandleT,
    FileInfo,
    FUSEError,
    InodeT,
    PollHandle,
    ReaddirToken,
    RequestContext,
    SetattrFields,
)
from gpiod.line import Direction, Edge, Value


class Range:

    def __init__(self, base, size):
        self._base = base
        self._size = size

    @property
    def base(self):
        return self._base

    @property
    def size(self):
        return self._size

    @property
    def end(self):
        return self.base + self.size


class RangeManager:

    BASE = 512

    def __init__(self):
        self._ranges = list()

    def get_new_base(self, size):
        # If there are no other ranges, just add the first one.
        if not self._ranges:
            new = Range(RangeManager.BASE, size)
            self._ranges.append(new)
        else:
            nranges = len(self._ranges)
            base = RangeManager.BASE

            for i in range(nranges):
                curr = self._ranges[i]

                if base + size < curr.base:
                    # This range fits before the current element.
                    new = Range(base, size)
                    self._ranges.insert(i, new)
                else:
                    base = curr.end

                    if i == nranges - 1:
                        # This was the last element so just append the new range.
                        new = Range(base, size)
                        self._ranges.append(new)

        return new.base

    def free_range(self, base):
        for entry in self._ranges:
            if base == entry.base:
                self._ranges.remove(entry)
                return

        raise ValueError(f"Range not found for base {base}")


_inode_counter = pyfuse3.ROOT_INODE


def _alloc_inode():
    global _inode_counter
    _inode_counter += 1
    return InodeT(_inode_counter)


_now_ns = int(time.time() * 1e9)


def _make_entry_attr(mode, nlink, size=4096):
    attr = EntryAttributes()
    attr.st_ino = _alloc_inode()
    attr.st_mode = mode
    attr.st_nlink = nlink
    attr.st_uid = 0
    attr.st_gid = 0
    attr.st_rdev = 0
    attr.st_size = size
    attr.st_blksize = 512
    attr.st_blocks = (size + 511) // 512
    attr.st_atime_ns = _now_ns
    attr.st_mtime_ns = _now_ns
    attr.st_ctime_ns = _now_ns
    attr.generation = 0
    attr.entry_timeout = 0
    attr.attr_timeout = 0
    return attr


class Entry:

    def __init__(self, parent):
        self._parent = parent
        self._attr = self._make_attr()

    def _make_attr(self):
        raise NotImplementedError

    def lookup(self, name):
        raise FUSEError(errno.ENOENT)

    def readdir(self):
        raise FUSEError(errno.ENOTDIR)

    def open(self, flags):
        raise FUSEError(errno.ENOENT)

    def read(self, fh, offset, size):
        raise FUSEError(errno.EBADF)

    def write(self, fh, offset, buf):
        raise FUSEError(errno.EPERM)

    def poll(self, fh, pollhandle):
        raise FUSEError(errno.ENOSYS)

    def readlink(self):
        raise FUSEError(errno.EINVAL)

    def rmdir(self, name):
        raise FUSEError(errno.EPERM)

    def chmod(self, mode):
        self._attr.st_mode = mode
        return self._attr

    def chown(self, uid, gid):
        self._attr.st_uid = uid
        self._attr.st_gid = gid
        return self._attr

    def truncate(self, size):
        return self._attr

    @property
    def parent(self):
        return self._parent

    @property
    def attr(self):
        return self._attr

    @property
    def inode(self):
        return self._attr.st_ino


class Directory(Entry):

    def __init__(self, parent):
        self._children = dict()
        Entry.__init__(self, parent)

    def _make_attr(self):
        return _make_entry_attr(
            stat.S_IFDIR
            | stat.S_IRUSR
            | stat.S_IWUSR
            | stat.S_IXUSR
            | stat.S_IRGRP
            | stat.S_IXGRP
            | stat.S_IROTH
            | stat.S_IXOTH,
            1,
        )

    def lookup(self, name):
        if name in self._children:
            return self._children[name]
        raise FUSEError(errno.ENOENT)

    def readdir(self):
        return list(self._children.items())

    def rmdir(self, name):
        raise FUSEError(errno.ENOTDIR)

    @property
    def children(self):
        return self._children


class Attribute(Entry):

    def _make_attr(self):
        return _make_entry_attr(stat.S_IFREG, 1)

    def open(self, flags):
        return None


class ConstRoAttr(Attribute):

    def __init__(self, parent, value):
        self._value = value
        Attribute.__init__(self, parent)

    def _make_attr(self):
        a = _make_entry_attr(
            stat.S_IFREG | stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH, 1
        )
        return a

    def read(self, fh, offset, size):
        data = f"{self._value}\n".encode()
        return data[offset : offset + size]


class RwAttr(Attribute):

    def _make_attr(self):
        return _make_entry_attr(
            stat.S_IFREG | stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH,
            1,
        )

    def write(self, fh, offset, buf):
        try:
            self.do_write(buf.strip().decode())
        except FUSEError:
            raise
        except ValueError:
            raise FUSEError(errno.EINVAL)
        return len(buf)

    def do_write(self, buf):
        raise NotImplementedError


class Link(Entry):

    def __init__(self, parent, path):
        self._path = path
        Entry.__init__(self, parent)

    def _make_attr(self):
        return _make_entry_attr(
            stat.S_IFLNK
            | stat.S_IRUSR
            | stat.S_IWUSR
            | stat.S_IXUSR
            | stat.S_IRGRP
            | stat.S_IXGRP
            | stat.S_IROTH
            | stat.S_IXOTH,
            2,
            0,
        )

    def readlink(self):
        return self._path.encode()


class ExportBase(RwAttr):

    def _make_attr(self):
        return _make_entry_attr(stat.S_IFREG | stat.S_IWUSR, 1)

    def do_write(self, buf):
        if not buf.isdigit():
            raise ValueError


class Export(ExportBase):

    def do_write(self, buf):
        ExportBase.do_write(self, buf)

        gpio = int(buf)

        for name, entry in self.parent.children.items():
            if not isinstance(entry, Gpiochip):
                continue

            if entry.has_gpio(gpio):
                entry.request(gpio)
                return

        raise ValueError


class Unexport(ExportBase):

    def do_write(self, buf):
        ExportBase.do_write(self, buf)

        gpio = f"gpio{int(buf)}"

        if gpio not in self.parent.children:
            raise ValueError

        entry = self.parent.children[gpio]
        entry.unexport()
        self.parent.children.pop(gpio)


class UeventAttr(RwAttr):

    CMDS = "(add|remove|change|move|online|offline|bind|unbind)"
    UUID = "[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"
    VARS = "(\\s+[A-Za-z0-9_]+\\=[A-Za-z0-9_]+)*"
    PATTERN = f"^{CMDS}\\s+{UUID}{VARS}$"

    def read(self, fh, offset, size):
        return b""

    def do_write(self, buf):
        if not re.match(UeventAttr.PATTERN, buf):
            raise ValueError


class Gpiochip(Directory):

    def __init__(self, parent, handle, base, syspath):
        Directory.__init__(self, parent)
        self._handle = handle
        self._info = self._handle.get_info()
        self._base = base

        self.children["base"] = ConstRoAttr(self, base)
        self.children["label"] = ConstRoAttr(self, self._info.label)
        self.children["ngpio"] = ConstRoAttr(self, self._info.num_lines)
        self.children["uevent"] = UeventAttr(self)
        self.children["device"] = Link(self, syspath)
        self.children["power"] = Link(self, f"{syspath}/power")
        self.children["subsystem"] = Link(self, self.parent.mountpoint)

    def has_gpio(self, gpio):
        return gpio >= self._base and gpio < (self._base + self._info.num_lines)

    def request(self, gpio):
        offset = gpio - self._base

        request = self._handle.request_lines(
            consumer="sysfs",
            config={offset: gpiod.LineSettings(direction=Direction.AS_IS)},
        )

        self.parent.children[f"gpio{gpio}"] = Gpio(
            self.parent, request, self._handle.get_line_info(offset), self
        )

    def close(self):
        self._handle.close()

    @property
    def name(self):
        return self._info.name

    @property
    def base(self):
        return self._base


class RwAttrWithVal(RwAttr):

    def __init__(self, parent, init_val):
        self._value = init_val
        RwAttr.__init__(self, parent)

    @property
    def value(self):
        return self._value

    @value.setter
    def value(self, val):
        self._value = val


class DirectionAttr(RwAttrWithVal):

    def read(self, fh, offset, size):
        dirstr = "in" if self.value == Direction.INPUT else "out"
        data = f"{dirstr}\n".encode()
        return data[offset : offset + size]

    def do_write(self, buf):
        if buf == "in":
            self.value = Direction.INPUT
        elif buf == "out":
            self.value = Direction.OUTPUT
        else:
            raise ValueError

        self.parent.reconfigure()


class EdgeAttr(RwAttrWithVal):

    def read(self, fh, offset, size):
        if self.value == Edge.RISING:
            edgestr = "rising"
        elif self.value == Edge.FALLING:
            edgestr = "falling"
        elif self.value == Edge.BOTH:
            edgestr = "both"
        else:
            edgestr = "none"

        data = f"{edgestr}\n".encode()
        return data[offset : offset + size]

    def do_write(self, buf):
        if buf == "none":
            self.value = Edge.NONE
        elif buf == "rising":
            self.value = Edge.RISING
        elif buf == "falling":
            self.value = Edge.FALLING
        elif buf == "both":
            self.value = Edge.BOTH
        else:
            raise ValueError

        self.parent.reconfigure()


class ActiveLowAttr(RwAttrWithVal):

    def read(self, fh, offset, size):
        val = "1" if self.value else "0"
        data = f"{val}\n".encode()
        return data[offset : offset + size]

    def do_write(self, buf):
        if not buf.isdigit():
            raise ValueError

        self.value = False if buf == "0" else True
        self.parent.reconfigure()


class ValueAttr(RwAttrWithVal):

    def __init__(self, parent):
        RwAttrWithVal.__init__(self, parent, None)
        self._event = False
        self._pollhandle = None

    def read(self, fh, offset, size):
        val = "1" if self.parent.get_value() == Value.ACTIVE else "0"
        data = f"{val}\n".encode()
        return data[offset : offset + size]

    def do_write(self, buf):
        if not buf.isdigit():
            raise ValueError

        val = Value.INACTIVE if buf == "0" else Value.ACTIVE
        try:
            self.parent.set_value(val)
        except PermissionError:
            raise FUSEError(errno.EPERM)

    def poll(self, fh, pollhandle):
        event = self._event
        self._event = False
        if pollhandle is not None:
            self._pollhandle = pollhandle
        # sysfs never blocks on POLLIN and POLLOUT; raise POLLPRI on edge event
        return select.POLLIN | select.POLLOUT | (select.POLLPRI if event else 0)

    def notify_poll(self):
        if self._pollhandle:
            self._event = True
            self._pollhandle.notify()
            self._pollhandle = None


class Gpio(Directory):

    def __init__(self, parent, handle, info, chip):
        Directory.__init__(self, parent)
        self._handle = handle
        self._offset = info.offset
        self._chip = chip

        val = ValueAttr(self)

        self.children["active_low"] = ActiveLowAttr(self, info.active_low)
        self.children["direction"] = DirectionAttr(self, info.direction)
        self.children["edge"] = EdgeAttr(self, info.edge_detection)
        self.children["value"] = val
        self.children["device"] = Link(self, f"/sys/bus/gpio/devices/{chip.name}")
        self.children["power"] = Link(self, f"/sys/bus/gpio/devices/{chip.name}/power")
        self.children["subsystem"] = Link(self, self.parent.mountpoint)
        self.children["uevent"] = UeventAttr(self)

        self.parent.watch_gpio(self._handle, val)

    def unexport(self):
        self.parent.unwatch_gpio(self._handle)
        self._handle.release()

    def reconfigure(self):
        self._handle.reconfigure_lines(
            {
                self._offset: gpiod.LineSettings(
                    direction=self.children["direction"].value,
                    edge_detection=self.children["edge"].value,
                    active_low=self.children["active_low"].value,
                )
            }
        )

    def get_value(self):
        return self._handle.get_values()[0]

    def set_value(self, val):
        self._handle.set_values({self._offset: val})

    @property
    def chip_name(self):
        return self._chip.name


class EventThread(Thread):

    def __init__(self):
        Thread.__init__(self)
        self._watched = dict()
        self._rdfd, self._wrfd = os.pipe()
        self._keep_going = True
        self._lock = Lock()

    def run(self):
        while self._keep_going:
            with self._lock:
                fds = list(self._watched.keys()) + [self._rdfd]

            readable, _, _ = select.select(fds, [], [], 60)
            for fd in readable:
                if fd == self._rdfd:
                    os.read(self._rdfd, 1024)
                    continue

                with self._lock:
                    value = self._watched.get(fd, None)
                    if not value:
                        continue

                    try:
                        os.read(fd, 4096)
                    except Exception:
                        self._watched.pop(fd)

                    value.notify_poll()

    def _intr_poll(self):
        os.write(self._wrfd, b".")

    def stop(self):
        self._keep_going = False
        self._intr_poll()

    def watch_gpio(self, request, value):
        with self._lock:
            self._watched[request.fd] = value

        self._intr_poll()

    def unwatch_gpio(self, request):
        with self._lock:
            self._watched.pop(request.fd)

        self._intr_poll()


class Root(Directory):

    def _make_attr(self):
        a = _make_entry_attr(
            stat.S_IFDIR
            | stat.S_IRUSR
            | stat.S_IWUSR
            | stat.S_IXUSR
            | stat.S_IRGRP
            | stat.S_IXGRP
            | stat.S_IROTH
            | stat.S_IXOTH,
            2,
        )
        a.st_ino = pyfuse3.ROOT_INODE
        return a

    def _add_chip(self, device):
        chip = gpiod.Chip(device.device_node)
        base = self._ranges.get_new_base(chip.get_info().num_lines)
        self.children[f"gpiochip{base}"] = Gpiochip(
            self, chip, base, os.path.dirname(device.sys_path)
        )

    def _remove_chip_gpios(self, chip):
        for name, entry in list(self.children.items()):
            if isinstance(entry, Gpio) and entry.chip_name == chip.name:
                self.children.pop(name)

    def _remove_chip(self, device):
        for name, entry in list(self.children.items()):
            if isinstance(entry, Gpiochip) and entry.name == device.sys_name:
                chip = self.children.pop(name)
                self._remove_chip_gpios(chip)
                chip.close()
                self._ranges.free_range(chip.base)
                break

    def _handle_udev_event(self, action, device):
        if not device.device_node:
            return None

        try:
            if action == "add":
                self._add_chip(device)
            elif action == "remove":
                self._remove_chip(device)
        except OSError as ex:
            if ex.errno not in [errno.ENOENT, errno.ENODEV]:
                print(traceback.format_exc(), file=sys.stderr)
                sys.exit(1)

    def _setup_udev(self):
        context = pyudev.Context()
        monitor = pyudev.Monitor.from_netlink(context)
        monitor.filter_by("gpio")
        self._observer = pyudev.MonitorObserver(monitor, self._handle_udev_event)
        self._observer.start()

        for device in context.list_devices(subsystem="gpio"):
            if device.device_node:
                self._add_chip(device)

    def __init__(self, mountpoint):
        Directory.__init__(self, None)
        self._attr.st_nlink = 2
        self._mountpoint = mountpoint

        self._ranges = RangeManager()
        self._evthread = EventThread()
        self._setup_udev()
        self.children["export"] = Export(self)
        self.children["unexport"] = Unexport(self)

        self._evthread.start()

    def stop(self):
        self._evthread.stop()
        self._observer.stop()
        self._evthread.join()
        self._observer.join()

    def watch_gpio(self, request, value):
        self._evthread.watch_gpio(request, value)

    def unwatch_gpio(self, request):
        self._evthread.unwatch_gpio(request)

    @property
    def mountpoint(self):
        return self._mountpoint


class GpioSysfsOperations(pyfuse3.Operations):

    def __init__(self, mountpoint):
        super().__init__()
        self._root = Root(mountpoint)
        self._inode_map = {pyfuse3.ROOT_INODE: self._root}
        self._fh_map = {}
        self._fh_counter = 0

    def stop(self):
        self._root.stop()

    def _alloc_fh(self, entry):
        self._fh_counter += 1
        fh = FileHandleT(self._fh_counter)
        self._fh_map[fh] = entry
        return fh

    def _get_entry_by_inode(self, inode):
        entry = self._inode_map.get(inode)
        if entry is None:
            raise FUSEError(errno.ENOENT)
        return entry

    def _register_tree(self, entry):
        """Register entry and all its descendants in the inode map."""
        self._inode_map[entry.inode] = entry
        if isinstance(entry, Directory):
            for child in entry.children.values():
                self._register_tree(child)

    def _lookup_and_register(self, parent_entry, name):
        child = parent_entry.lookup(name)
        # Register child and descendants (handles newly exported GPIOs)
        self._register_tree(child)
        return child

    async def lookup(self, parent_inode, name, ctx):
        parent = self._get_entry_by_inode(parent_inode)
        name_str = name.decode()
        child = self._lookup_and_register(parent, name_str)
        attr = EntryAttributes()
        self._copy_attr(child.attr, attr)
        return attr

    def _copy_attr(self, src, dst):
        for field in (
            "st_ino",
            "st_mode",
            "st_nlink",
            "st_uid",
            "st_gid",
            "st_rdev",
            "st_size",
            "st_blksize",
            "st_blocks",
            "st_atime_ns",
            "st_mtime_ns",
            "st_ctime_ns",
        ):
            setattr(dst, field, getattr(src, field))
        dst.generation = 0
        dst.entry_timeout = 0
        dst.attr_timeout = 0

    async def getattr(self, inode, ctx=None):
        entry = self._get_entry_by_inode(inode)
        attr = EntryAttributes()
        self._copy_attr(entry.attr, attr)
        return attr

    async def setattr(self, inode, attr, fields, fh, ctx):
        entry = self._get_entry_by_inode(inode)

        if fields.update_mode:
            entry.chmod(attr.st_mode)
        if fields.update_uid or fields.update_gid:
            uid = attr.st_uid if fields.update_uid else entry.attr.st_uid
            gid = attr.st_gid if fields.update_gid else entry.attr.st_gid
            entry.attr.st_uid = uid
            entry.attr.st_gid = gid
        if fields.update_size:
            entry.truncate(attr.st_size)

        result = EntryAttributes()
        self._copy_attr(entry.attr, result)
        return result

    async def readlink(self, inode, ctx):
        entry = self._get_entry_by_inode(inode)
        return entry.readlink()

    async def opendir(self, inode, ctx):
        entry = self._get_entry_by_inode(inode)
        if not isinstance(entry, Directory):
            raise FUSEError(errno.ENOTDIR)
        return FileHandleT(inode)

    async def readdir(self, fh, start_id, token):
        entry = self._get_entry_by_inode(InodeT(fh))
        children = entry.readdir()

        # Build list of (name, child_entry) sorted by child inode for stable ordering
        entries = []
        for name, child in children:
            self._register_tree(child)
            entries.append((child.inode, name, child))

        entries.sort(key=lambda x: x[0])

        for ino, name, child in entries:
            if ino <= start_id:
                continue
            attr = EntryAttributes()
            self._copy_attr(child.attr, attr)
            if not pyfuse3.readdir_reply(token, name.encode(), attr, ino):
                break

    async def open(self, inode, flags, ctx):
        entry = self._get_entry_by_inode(inode)
        entry.open(flags)
        fh = self._alloc_fh(entry)
        return FileInfo(fh=fh, direct_io=True)

    async def read(self, fh, offset, size):
        entry = self._fh_map.get(fh)
        if entry is None:
            raise FUSEError(errno.EBADF)
        return entry.read(fh, offset, size)

    async def write(self, fh, offset, buf):
        entry = self._fh_map.get(fh)
        if entry is None:
            raise FUSEError(errno.EBADF)
        return entry.write(fh, offset, buf)

    async def release(self, fh):
        self._fh_map.pop(fh, None)

    async def releasedir(self, fh):
        pass

    async def flush(self, fh):
        pass

    async def poll(self, inode, fh, poll_handle, ctx):
        entry = self._fh_map.get(fh)
        if entry is None:
            raise FUSEError(errno.EBADF)
        return entry.poll(fh, poll_handle)

    async def mknod(self, parent_inode, name, mode, rdev, ctx):
        raise FUSEError(errno.EACCES)

    async def mkdir(self, parent_inode, name, mode, ctx):
        raise FUSEError(errno.EPERM)

    async def rmdir(self, parent_inode, name, ctx):
        parent = self._get_entry_by_inode(parent_inode)
        parent.rmdir(name.decode())

    async def unlink(self, parent_inode, name, ctx):
        raise FUSEError(errno.EPERM)

    async def statfs(self, ctx):
        data = pyfuse3.StatvfsData()
        data.f_bsize = 4096
        data.f_frsize = 4096
        data.f_blocks = 0
        data.f_bfree = 0
        data.f_bavail = 0
        data.f_files = len(self._inode_map)
        data.f_ffree = 0
        data.f_favail = 0
        data.f_namemax = 255
        return data


def parse_args():
    parser = ArgumentParser(
        prog="gpiod-sysfs-proxy",
        description=(
            "Emulate the deprecated Linux GPIO sysfs interface "
            "using the modern libgpiod character device API."
        ),
    )
    parser.add_argument(
        "mountpoint",
        help="directory to mount the virtual GPIO sysfs filesystem over "
        "(typically /sys/class/gpio)",
    )
    parser.add_argument(
        "-f",
        "--foreground",
        action="store_true",
        default=False,
        help="run in the foreground instead of daemonising",
    )
    parser.add_argument(
        "-o",
        "--options",
        action="append",
        default=[],
        metavar="OPT[,OPT...]",
        help="pass comma-separated mount options through to FUSE; may be "
        "given multiple times (e.g. -o allow_other,default_permissions)",
    )
    parser.add_argument(
        "-d",
        "--debug-fuse",
        action="store_true",
        default=False,
        help="enable FUSE debug output (implies foreground operation)",
    )
    parser.add_argument(
        "-V",
        "--version",
        action="version",
        version=f"%(prog)s {__version__}",
        help="show the program version and exit",
    )

    # Capitalise group titles.
    # This is to allow help2man to promote the lists to dedicated sections in
    # the generated manual page.
    parser._positionals.title = "POSITIONAL ARGUMENTS"
    parser._optionals.title = "OPTIONS"

    return parser.parse_args()


def main():
    args = parse_args()

    fuse_options = set(pyfuse3.default_options)
    fuse_options.add("fsname=gpiod-sysfs-proxy")

    for opt_group in args.options:
        for opt in opt_group.split(","):
            opt = opt.strip()
            if opt:
                fuse_options.add(opt)

    if args.debug_fuse:
        fuse_options.add("debug")

    ops = GpioSysfsOperations(args.mountpoint)
    pyfuse3.init(ops, args.mountpoint, fuse_options)

    try:
        trio.run(pyfuse3.main)
    except Exception:
        pyfuse3.close(unmount=False)
        ops.stop()
        raise

    pyfuse3.close()
    ops.stop()


if __name__ == "__main__":
    main()
