chongming-plugins/aibox_sophon_adapter/diwatch/backends.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""DI 通道的底层读取后端。

每个后端只负责：open / close / read。
电平语义（0=低，1=高）由上层 DIChannel 统一处理。

后端类型：
  - EventBackend：/dev/input/eventN，事件驱动，可 poll。SE5 的 IN1。
  - SysfsBackend：/sys/.../gpio_status 或 /dev/tv_gpioin_N，轮询 + 抗抖。
                   SE9 的 IN1、所有设备的 IN2。
"""

import errno
import glob
import os
import struct
import time

from .config import (
    EV_KEY, VALUE_RELEASE, VALUE_PRESS, VALUE_REPEAT,
    DEFAULT_EVENT_DEVICE, GPIO_KEYS_NAME, DEFAULT_IN2_KEY,
    DEBOUNCE, DEFAULT_INTERVAL,
    find_status_file,
)

EVENT_STRUCT = struct.Struct("llHHi")  # aarch64 / x86_64 都是 24 字节


# ---------------------------------------------------------------------------
# 基类
# ---------------------------------------------------------------------------
class Backend:
    """后端接口。"""
    can_poll = False

    def open(self):
        raise NotImplementedError

    def close(self):
        raise NotImplementedError

    def read(self):
        """非阻塞读一次，返回本批事件列表 [(level, ts), ...]；无事件返回 []。

        level 是原始电平 0/1，尚未做 active_low 翻转。
        同一批数据里的多个事件必须全部返回，不能只留最后一个，
        否则快速抖动（一次 read 里按下+释放）会丢失中间跳变。
        """
        raise NotImplementedError

    @property
    def name(self):
        raise NotImplementedError


# ---------------------------------------------------------------------------
# /dev/input/eventN 后端（SE5 的 IN1）
# ---------------------------------------------------------------------------
class EventBackend(Backend):
    """基于 input 事件设备，事件驱动。

    原始 value 语义：
        VALUE_PRESS(1)   -> 硬件“按下”
        VALUE_RELEASE(0) -> 硬件“释放”
    本后端返回原始 value，极性翻转留给 DIChannel。
    """

    can_poll = True

    def __init__(self, device=DEFAULT_EVENT_DEVICE, key=None,
                 find_by_name=GPIO_KEYS_NAME):
        self.device = device
        self.key = key
        self.find_by_name = find_by_name
        self.fd = None
        self.buf = b""

    # ---- 自动查找 ----
    @staticmethod
    def find_event_device(name=GPIO_KEYS_NAME):
        """按设备名找 eventN，找不到返回 None。"""
        for path in sorted(glob.glob("/dev/input/event*")):
            ev = path.rsplit("/", 1)[-1]
            namefile = f"/sys/class/input/{ev}/device/name"
            try:
                with open(namefile) as f:
                    if f.read().strip() == name:
                        return path
            except OSError:
                continue
        return None

    def resolve_device(self):
        if not self.device or self.device == "auto":
            return self.find_event_device(self.find_by_name) or DEFAULT_EVENT_DEVICE
        return self.device

    @property
    def name(self):
        dev = self.device or "(auto)"
        return f"event:{dev}"

    # ---- 生命周期 ----
    def open(self):
        if self.fd is None:
            dev = self.resolve_device()
            if not dev or not os.path.exists(dev):
                raise FileNotFoundError(
                    errno.ENOENT,
                    f"event 设备不存在：{dev}",
                    dev,
                )
            self.device = dev
            self.fd = os.open(dev, os.O_RDONLY | os.O_NONBLOCK)

    def close(self):
        if self.fd is not None:
            try:
                os.close(self.fd)
            finally:
                self.fd = None

    def fileno(self):
        return self.fd

    # ---- 读取 ----
    def read(self):
        """返回 [(level, ts), ...]；只汇报关心的 key，批内事件按序全部返回。"""
        if self.fd is None:
            return []
        try:
            data = os.read(self.fd, EVENT_STRUCT.size * 64)
        except OSError:
            return []
        self.buf += data
        events = []
        while len(self.buf) >= EVENT_STRUCT.size:
            chunk = self.buf[:EVENT_STRUCT.size]
            self.buf = self.buf[EVENT_STRUCT.size:]
            sec, usec, etype, code, value = EVENT_STRUCT.unpack(chunk)
            if etype != EV_KEY:
                continue
            if self.key is not None and code != self.key:
                continue
            if value == VALUE_REPEAT:
                continue
            if value not in (VALUE_RELEASE, VALUE_PRESS):
                continue
            ts = sec + usec / 1e6
            events.append((value, ts))
        return events


# ---------------------------------------------------------------------------
# sysfs 状态文件后端（SE9 的 IN1、所有 IN2）
# ---------------------------------------------------------------------------
class SysfsBackend(Backend):
    """基于 /sys/.../gpio_status 或 /dev/tv_gpioin_N，轮询采样 + 抗抖。"""

    can_poll = False

    def __init__(self, path, debounce=DEBOUNCE, interval=DEFAULT_INTERVAL):
        self.path = path
        self.debounce = max(1, int(debounce))
        self.interval = interval
        self._cand = None
        self._cand_n = 0
        self._reported = None   # 上次上报的稳定值（跳变只报一次）

    @property
    def name(self):
        return f"sysfs:{self.path}"

    def open(self):
        if not os.access(self.path, os.R_OK):
            raise FileNotFoundError(
                errno.ENOENT,
                f"状态文件不可读：{self.path}",
                self.path,
            )

    def close(self):
        pass

    # ---- 原始电平 ----
    def read_raw(self):
        try:
            with open(self.path, "r") as f:
                v = f.read()
        except OSError:
            return None
        v = v.strip().strip("\t\r\n ")
        return v if v in ("0", "1") else None

    # ---- 带抗抖的读取 ----
    def read(self):
        """返回 [(level, ts)]；抗抖未确认或电平未变化时返回 []。

        连续 debounce 次采样一致才确认稳定值；确认值与上次上报不同
        （真实跳变）才上报一次，避免稳态期间每个抗抖周期重复产出事件。
        """
        v = self.read_raw()
        if v is None:
            return []
        ts = time.time()

        if self._cand is None or v == self._cand:
            self._cand = v
            self._cand_n += 1
        else:
            self._cand = v
            self._cand_n = 1

        if self._cand_n < self.debounce:
            return []
        confirmed = int(v)
        if self._reported == confirmed:
            return []
        self._reported = confirmed
        return [(confirmed, ts)]


# ---------------------------------------------------------------------------
# 工厂
# ---------------------------------------------------------------------------
def make_backend(kind: str, **kwargs):
    """kind = 'event' | 'sysfs'。"""
    if kind == "event":
        return EventBackend(**kwargs)
    if kind == "sysfs":
        return SysfsBackend(**kwargs)
    raise ValueError(f"未知后端类型：{kind}")


def make_in1_backend(device=None, key=None, in_index=1):
    """自动为 IN1 选择后端，适配 SE5 / SE9。

    优先级：
      1) 显式 device（非 auto） -> EventBackend
      2) 自动查找 gpio-keys 的 event 设备 -> EventBackend
      3) 查找 gpio-in-<in_index> 的状态文件 -> SysfsBackend
      4) 都找不到 -> 返回 EventBackend（open 时抛出可诊断的错误）
    """
    # 1) 显式指定
    if device and device != "auto":
        return EventBackend(device, key=key)

    # 2) 自动查找 gpio-keys
    ev = EventBackend.find_event_device(GPIO_KEYS_NAME)
    if ev:
        return EventBackend(ev, key=key)

    # 3) 回退到 sysfs 状态文件
    status = find_status_file(in_index)
    if status:
        return SysfsBackend(status)

    # 4) 都找不到，返回一个 open 时会报错的 EventBackend
    return EventBackend(DEFAULT_EVENT_DEVICE, key=key)


def make_in2_backend(device=None, key=None, in_index=2,
                     interval=DEFAULT_INTERVAL, debounce=DEBOUNCE):
    """自动为 IN2 选择后端：优先 sysfs（SE9），回退 gpio-keys event（SE5）。

    SE5 无 gpio-in-2 状态文件，IN2 与 IN1 同走 gpio-keys 事件设备（不同键码）。

    优先级：
      1) 显式 device（event 设备路径） -> EventBackend
      2) 查找 gpio-in-<in_index> 的状态文件 -> SysfsBackend
      3) 自动查找 gpio-keys 的 event 设备 -> EventBackend（key 缺省 DEFAULT_IN2_KEY）
      4) 都找不到 -> 返回 None（由调用方给出可诊断错误）
    """
    # 1) 显式指定 event 设备
    if device and device != "auto":
        return EventBackend(device, key=key if key is not None else DEFAULT_IN2_KEY)

    # 2) sysfs 状态文件（SE9）
    status = find_status_file(in_index)
    if status:
        return SysfsBackend(status, debounce=debounce, interval=interval)

    # 3) 回退到 gpio-keys 事件设备（SE5）
    ev = EventBackend.find_event_device(GPIO_KEYS_NAME)
    if ev:
        return EventBackend(ev, key=key if key is not None else DEFAULT_IN2_KEY)

    # 4) 都找不到
    return None

def find_event_device(name=GPIO_KEYS_NAME):
    """模块级别别名，等价于 EventBackend.find_event_device."""
    return EventBackend.find_event_device(name)

__all__ = [
    "Backend",
    "EventBackend",
    "SysfsBackend",
    "make_backend",
    "make_in1_backend",
    "make_in2_backend",
    "find_status_file",
    "find_event_device",
]