#!/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# -*- coding: utf-8 -*-
#
# Copyright (c) 2017 Benjamin Tissoires <[email protected]>
# Copyright (c) 2017 Red Hat, Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import fcntl
import functools
import libevdev
import os
try:
import pyudev
except ImportError:
raise ImportError("UHID is not supported due to missing pyudev dependency")
import logging
import hidtools.hid as hid
from hidtools.uhid import UHIDDevice
from hidtools.util import BusType
from pathlib import Path
from typing import Any, ClassVar, Dict, List, Optional, Tuple, Type, Union
logger = logging.getLogger("hidtools.device.base_device")
class SysfsFile(object):
def __init__(self, path):
self.path = path
def __set_value(self, value):
with open(self.path, "w") as f:
return f.write(f"{value}\n")
def __get_value(self):
with open(self.path) as f:
return f.read().strip()
@property
def int_value(self) -> int:
return int(self.__get_value())
@int_value.setter
def int_value(self, v: int) -> None:
self.__set_value(v)
@property
def str_value(self) -> str:
return self.__get_value()
@str_value.setter
def str_value(self, v: str) -> None:
self.__set_value(v)
class LED(object):
def __init__(self, sys_path):
self.max_brightness = SysfsFile(sys_path / "max_brightness").int_value
self.__brightness = SysfsFile(sys_path / "brightness")
@property
def brightness(self) -> int:
return self.__brightness.int_value
@brightness.setter
def brightness(self, value: int) -> None:
self.__brightness.int_value = value
class PowerSupply(object):
"""Represents Linux power_supply_class sysfs nodes."""
def __init__(self, sys_path):
self._capacity = SysfsFile(sys_path / "capacity")
self._status = SysfsFile(sys_path / "status")
self._type = SysfsFile(sys_path / "type")
@property
def capacity(self) -> int:
return self._capacity.int_value
@property
def status(self) -> str:
return self._status.str_value
@property
def type(self) -> str:
return self._type.str_value
class HIDIsReady(object):
"""
Companion class that binds to a kernel mechanism
and that allows to know when a uhid device is ready or not.
See :meth:`is_ready` for details.
"""
def __init__(self: "HIDIsReady", uhid: UHIDDevice) -> None:
self.uhid = uhid
def is_ready(self: "HIDIsReady") -> bool:
"""
Overwrite in subclasses: should return True or False whether
the attached uhid device is ready or not.
"""
return False
class UdevHIDIsReady(HIDIsReady):
_pyudev_context: ClassVar[Optional[pyudev.Context]] = None
_pyudev_monitor: ClassVar[Optional[pyudev.Monitor]] = None
_uhid_devices: ClassVar[Dict[int, Tuple[bool, int]]] = {}
def __init__(self: "UdevHIDIsReady", uhid: UHIDDevice) -> None:
super().__init__(uhid)
self._init_pyudev()
@classmethod
def _init_pyudev(cls: Type["UdevHIDIsReady"]) -> None:
if cls._pyudev_context is None:
cls._pyudev_context = pyudev.Context()
cls._pyudev_monitor = pyudev.Monitor.from_netlink(cls._pyudev_context)
cls._pyudev_monitor.filter_by("hid")
cls._pyudev_monitor.start()
UHIDDevice._append_fd_to_poll(
cls._pyudev_monitor.fileno(), cls._cls_udev_event_callback
)
@classmethod
def _cls_udev_event_callback(cls: Type["UdevHIDIsReady"]) -> None:
if cls._pyudev_monitor is None:
return
event: pyudev.Device
for event in iter(functools.partial(cls._pyudev_monitor.poll, 0.02), None):
if event.action not in ["bind", "remove", "unbind"]:
return
logger.debug(f"udev event: {event.action} -> {event}")
id = int(event.sys_path.strip().split(".")[-1], 16)
device_ready, count = cls._uhid_devices.get(id, (False, 0))
ready = event.action == "bind"
if not device_ready and ready:
count += 1
cls._uhid_devices[id] = (ready, count)
def is_ready(self: "UdevHIDIsReady") -> Tuple[bool, int]:
try:
return self._uhid_devices[self.uhid.hid_id]
except KeyError:
return (False, 0)
class EvdevMatch(object):
def __init__(
self: "EvdevMatch",
*,
requires: List[Any] = [],
excludes: List[Any] = [],
req_properties: List[Any] = [],
excl_properties: List[Any] = [],
) -> None:
self.requires = requires
self.excludes = excludes
self.req_properties = req_properties
self.excl_properties = excl_properties
def is_a_match(self: "EvdevMatch", evdev: libevdev.Device) -> bool:
for m in self.requires:
if not evdev.has(m):
return False
for m in self.excludes:
if evdev.has(m):
return False
for p in self.req_properties:
if not evdev.has_property(p):
return False
for p in self.excl_properties:
if evdev.has_property(p):
return False
return True
class EvdevDevice(object):
"""
Represents an Evdev node and its properties.
This is a stub for the libevdev devices, as they are relying on
uevent to get the data, saving us some ioctls to fetch the names
and properties.
"""
def __init__(self: "EvdevDevice", sysfs: Path) -> None:
self.sysfs = sysfs
self.event_node: Any = None
self.libevdev: Optional[libevdev.Device] = None
self.uevents = {}
# all of the interesting properties are stored in the input uevent, so in the parent
# so convert the uevent file of the parent input node into a dict
with open(sysfs.parent / "uevent") as f:
for line in f.readlines():
key, value = line.strip().split("=")
self.uevents[key] = value.strip('"')
# we open all evdev nodes in order to not miss any event
self.open()
@property
def name(self: "EvdevDevice") -> str:
assert "NAME" in self.uevents
return self.uevents["NAME"]
@property
def evdev(self: "EvdevDevice") -> Path:
return Path("/dev/input") / self.sysfs.name
def matches_application(
self: "EvdevDevice", application: str, matches: Dict[str, EvdevMatch]
) -> bool:
if self.libevdev is None:
return False
if application in matches:
return matches[application].is_a_match(self.libevdev)
logger.error(
f"application '{application}' is unknown, please update/fix hid-tools"
)
assert False # hid-tools likely needs an update
def open(self: "EvdevDevice") -> libevdev.Device:
self.event_node = open(self.evdev, "rb")
self.libevdev = libevdev.Device(self.event_node)
assert self.libevdev.fd is not None
fd = self.libevdev.fd.fileno()
flag = fcntl.fcntl(fd, fcntl.F_GETFD)
fcntl.fcntl(fd, fcntl.F_SETFL, flag | os.O_NONBLOCK)
return self.libevdev
def close(self: "EvdevDevice") -> None:
if self.libevdev is not None and self.libevdev.fd is not None:
self.libevdev.fd.close()
self.libevdev = None
if self.event_node is not None:
self.event_node.close()
self.event_node = None
class BaseDevice(UHIDDevice):
# default _application_matches that matches nothing. This needs
# to be set in the subclasses to have get_evdev() working
_application_matches: Dict[str, EvdevMatch] = {}
def __init__(
self,
name,
application,
rdesc_str: Optional[str] = None,
rdesc: Optional[Union[hid.ReportDescriptor, str, bytes]] = None,
input_info=None,
) -> None:
self._kernel_is_ready: HIDIsReady = UdevHIDIsReady(self)
if rdesc_str is None and rdesc is None:
raise Exception("Please provide at least a rdesc or rdesc_str")
super().__init__()
if name is None:
name = f"uhid gamepad test {self.__class__.__name__}"
if input_info is None:
input_info = (BusType.USB, 1, 2)
self.name = name
self.info = input_info
self.default_reportID = None
self.opened = False
self.started = False
self.application = application
self._input_nodes: Optional[list[EvdevDevice]] = None
if rdesc is None:
assert rdesc_str is not None
self.rdesc = hid.ReportDescriptor.from_human_descr(rdesc_str) # type: ignore
else:
self.rdesc = rdesc # type: ignore
@property
def power_supply_class(self: "BaseDevice") -> Optional[PowerSupply]:
ps = self.walk_sysfs("power_supply", "power_supply/*")
if ps is None or len(ps) < 1:
return None
return PowerSupply(ps[0])
@property
def led_classes(self: "BaseDevice") -> List[LED]:
leds = self.walk_sysfs("led", "**/max_brightness")
if leds is None:
return []
return [LED(led.parent) for led in leds]
@property
def kernel_is_ready(self: "BaseDevice") -> bool:
return self._kernel_is_ready.is_ready()[0] and self.started
@property
def kernel_ready_count(self: "BaseDevice") -> int:
return self._kernel_is_ready.is_ready()[1]
@property
def input_nodes(self: "BaseDevice") -> List[EvdevDevice]:
if self._input_nodes is not None:
return self._input_nodes
if not self.kernel_is_ready or not self.started:
return []
self._input_nodes = [
EvdevDevice(path)
for path in self.walk_sysfs("input", "input/input*/event*")
]
return self._input_nodes
def match_evdev_rule(self, application, evdev):
"""Replace this in subclasses if the device has multiple reports
of the same type and we need to filter based on the actual evdev
node.
returning True will append the corresponding report to
`self.input_nodes[type]`
returning False will ignore this report / type combination
for the device.
"""
return True
def open(self):
self.opened = True
def _close_all_opened_evdev(self):
if self._input_nodes is not None:
for e in self._input_nodes:
e.close()
def __del__(self):
self._close_all_opened_evdev()
def close(self):
self.opened = False
def start(self, flags):
self.started = True
def stop(self):
self.started = False
self._close_all_opened_evdev()
def next_sync_events(self, application=None):
evdev = self.get_evdev(application)
if evdev is not None:
return list(evdev.events())
return []
@property
def application_matches(self: "BaseDevice") -> Dict[str, EvdevMatch]:
return self._application_matches
@application_matches.setter
def application_matches(self: "BaseDevice", data: Dict[str, EvdevMatch]) -> None:
self._application_matches = data
def get_evdev(self, application=None):
if application is None:
application = self.application
if len(self.input_nodes) == 0:
return None
assert self._input_nodes is not None
if len(self._input_nodes) == 1:
evdev = self._input_nodes[0]
if self.match_evdev_rule(application, evdev.libevdev):
return evdev.libevdev
else:
for _evdev in self._input_nodes:
if _evdev.matches_application(application, self.application_matches):
if self.match_evdev_rule(application, _evdev.libevdev):
return _evdev.libevdev
def is_ready(self):
"""Returns whether a UHID device is ready. Can be overwritten in
subclasses to add extra conditions on when to consider a UHID
device ready. This can be:
- we need to wait on different types of input devices to be ready
(Touch Screen and Pen for example)
- we need to have at least 4 LEDs present
(len(self.uhdev.leds_classes) == 4)
- or any other combinations"""
return self.kernel_is_ready
Codebase Browser
linux