# SPDX-License-Identifier: LGPL-2.1-or-later
# SPDX-FileCopyrightText: 2022 Bartosz Golaszewski <brgl@bgdev.pl>
from . import _ext
from .edge_event import EdgeEvent
from .exception import RequestReleasedError
from .internal import poll_fd
from .line import Value
from .line_settings import LineSettings, _line_settings_to_ext
from collections.abc import Iterable
from datetime import timedelta
from typing import Optional, Union
__all__ = "LineRequest"
class LineRequest:
"""
Stores the context of a set of requested GPIO lines.
"""
def __init__(self, req: _ext.Request):
"""
Note: LineRequest objects can only be instantiated by a Chip parent.
LineRequest.__init__() is not part of stable API.
"""
self._req = req
def __bool__(self) -> bool:
"""
Boolean conversion for GPIO line requests.
Returns:
True if the request is live and False if it's been released.
"""
return True if self._req else False
def __enter__(self):
"""
Controlled execution enter callback.
"""
self._check_released()
return self
def __exit__(self, exc_type, exc_value, traceback):
"""
Controlled execution exit callback.
"""
self.release()
def _check_released(self) -> None:
if not self._req:
raise RequestReleasedError()
def release(self) -> None:
"""
Release this request and free all associated resources. The object must
not be used after a call to this method.
"""
self._check_released()
self._req.release()
self._req = None
def get_value(self, line: Union[int, str]) -> Value:
"""
Get a single GPIO line value.
Args:
line:
Offset or name of the line to get value for.
Returns:
Logical value of the line.
"""
return self.get_values([line])[0]
def _check_line_name(self, line):
if isinstance(line, str):
if line not in self._name_map:
raise ValueError("unknown line name: {}".format(line))
return True
return False
def get_values(
self, lines: Optional[Iterable[Union[int, str]]] = None
) -> list[Value]:
"""
Get values of a set of GPIO lines.
Args:
lines:
List of names or offsets of GPIO lines to get values for. Can be
None in which case all requested lines will be read.
Returns:
List of logical line values.
"""
self._check_released()
lines = lines or self._lines
offsets = [
self._name_map[line] if self._check_line_name(line) else line
for line in lines
]
buf = [None] * len(lines)
self._req.get_values(offsets, buf)
return buf
def set_value(self, line: Union[int, str], value: Value) -> None:
"""
Set the value of a single GPIO line.
Args:
line:
Offset or name of the line to set.
value:
New value.
"""
self.set_values({line: value})
def set_values(self, values: dict[Union[int, str], Value]) -> None:
"""
Set the values of a subset of GPIO lines.
Args:
values:
Dictionary mapping line offsets or names to desired values.
"""
self._check_released()
mapped = {
self._name_map[line] if self._check_line_name(line) else line: values[line]
for line in values
}
self._req.set_values(mapped)
def reconfigure_lines(
self, config: dict[tuple[Union[int, str]], LineSettings]
) -> None:
"""
Reconfigure requested lines.
Args:
config
Dictionary mapping offsets or names (or tuples thereof) to
LineSettings. If no entry exists, or a None is passed as the
settings, then the configuration for that line is not changed.
Any settings for non-requested lines are ignored.
"""
self._check_released()
line_cfg = _ext.LineConfig()
line_settings = {}
for lines, settings in config.items():
if isinstance(lines, int) or isinstance(lines, str):
lines = [lines]
for line in lines:
offset = self._name_map[line] if self._check_line_name(line) else line
line_settings[offset] = settings
for offset in self.offsets:
settings = line_settings.get(offset) or LineSettings()
line_cfg.add_line_settings([offset], _line_settings_to_ext(settings))
self._req.reconfigure_lines(line_cfg)
def wait_edge_events(
self, timeout: Optional[Union[timedelta, float]] = None
) -> bool:
"""
Wait for edge events on any of the requested lines.
Args:
timeout:
Wait time limit expressed as either a datetime.timedelta object
or the number of seconds stored in a float. If set to 0, the
method returns immediately, if set to None it blocks indefinitely.
Returns:
True if events are ready to be read. False on timeout.
"""
self._check_released()
return poll_fd(self.fd, timeout)
def read_edge_events(self, max_events: Optional[int] = None) -> list[EdgeEvent]:
"""
Read a number of edge events from a line request.
Args:
max_events:
Maximum number of events to read.
Returns:
List of read EdgeEvent objects.
"""
self._check_released()
return self._req.read_edge_events(max_events)
def __str__(self):
"""
Return a user-friendly, human-readable description of this request.
"""
if not self._req:
return "<LineRequest RELEASED>"
return '<LineRequest chip="{}" num_lines={} offsets={} fd={}>'.format(
self.chip_name, self.num_lines, self.offsets, self.fd
)
@property
def chip_name(self) -> str:
"""
Name of the chip this request was made on.
"""
self._check_released()
return self._chip_name
@property
def num_lines(self) -> int:
"""
Number of requested lines.
"""
self._check_released()
return len(self._offsets)
@property
def offsets(self) -> list[int]:
"""
List of requested offsets. Lines requested by name are mapped to their
offsets.
"""
self._check_released()
return self._offsets
@property
def lines(self) -> list[Union[int, str]]:
"""
List of requested lines. Lines requested by name are listed as such.
"""
self._check_released()
return self._lines
@property
def fd(self) -> int:
"""
File descriptor associated with this request.
"""
self._check_released()
return self._req.fd