maxime/loustique-home
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1.0

Browse Source
This commit is contained in:
maxime
2026-03-21 10:53:02 +01:00
parent 3a6c07513c
commit ccfeb6f2c1
1997 changed files with 361936 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

588
venv/lib/python3.11/site-packages/gpiozero/mixins.py Normal file
View File

@@ -0,0 +1,588 @@
# vim: set fileencoding=utf-8:
#
# GPIO Zero: a library for controlling the Raspberry Pi's GPIO pins
#
# Copyright (c) 2016-2023 Dave Jones <dave@waveform.org.uk>
# Copyright (c) 2018-2021 Ben Nuttall <ben@bennuttall.com>
# Copyright (c) 2020 Fangchen Li <fangchen.li@outlook.com>
# Copyright (c) 2016 Andrew Scheller <github@loowis.durge.org>
#
# SPDX-License-Identifier: BSD-3-Clause
import inspect
import weakref
import warnings
from functools import wraps, partial
from threading import Event
from collections import deque
from statistics import median
from .threads import GPIOThread
from .exc import (
BadEventHandler,
BadWaitTime,
BadQueueLen,
DeviceClosed,
CallbackSetToNone,
)
callback_warning = (
'The callback was set to None. This may have been unintentional '
'e.g. btn.when_pressed = pressed() instead of btn.when_pressed = pressed'
)
class ValuesMixin:
"""
Adds a :attr:`values` property to the class which returns an infinite
generator of readings from the :attr:`~Device.value` property. There is
rarely a need to use this mixin directly as all base classes in GPIO Zero
include it.
.. note::
Use this mixin *first* in the parent class list.
"""
@property
def values(self):
"""
An infinite iterator of values read from :attr:`value`.
"""
while True:
try:
yield self.value
except DeviceClosed:
break
class SourceMixin:
"""
Adds a :attr:`source` property to the class which, given an iterable or a
:class:`ValuesMixin` descendent, sets :attr:`~Device.value` to each member
of that iterable until it is exhausted. This mixin is generally included in
novel output devices to allow their state to be driven from another device.
.. note::
Use this mixin *first* in the parent class list.
"""
def __init__(self, *args, **kwargs):
self._source = None
self._source_thread = None
self._source_delay = 0.01
super().__init__(*args, **kwargs)
def close(self):
self.source = None
super().close()
def _copy_values(self, source):
for v in source:
self.value = v
if self._source_thread.stopping.wait(self._source_delay):
break
@property
def source_delay(self):
"""
The delay (measured in seconds) in the loop used to read values from
:attr:`source`. Defaults to 0.01 seconds which is generally sufficient
to keep CPU usage to a minimum while providing adequate responsiveness.
"""
return self._source_delay
@source_delay.setter
def source_delay(self, value):
if value < 0:
raise BadWaitTime('source_delay must be 0 or greater')
self._source_delay = float(value)
@property
def source(self):
"""
The iterable to use as a source of values for :attr:`value`.
"""
return self._source
@source.setter
def source(self, value):
if getattr(self, '_source_thread', None):
self._source_thread.stop()
self._source_thread = None
if isinstance(value, ValuesMixin):
value = value.values
self._source = value
if value is not None:
self._source_thread = GPIOThread(self._copy_values, (value,))
self._source_thread.start()
class SharedMixin:
"""
This mixin marks a class as "shared". In this case, the meta-class
(GPIOMeta) will use :meth:`_shared_key` to convert the constructor
arguments to an immutable key, and will check whether any existing
instances match that key. If they do, they will be returned by the
constructor instead of a new instance. An internal reference counter is
used to determine how many times an instance has been "constructed" in this
way.
When :meth:`~Device.close` is called, an internal reference counter will be
decremented and the instance will only close when it reaches zero.
"""
_instances = {}
def __del__(self):
self._refs = 0
super().__del__()
@classmethod
def _shared_key(cls, *args, **kwargs):
"""
This is called with the constructor arguments to generate a unique
key (which must be storable in a :class:`dict` and, thus, immutable
and hashable) representing the instance that can be shared. This must
be overridden by descendents.
"""
raise NotImplementedError
class event:
"""
A descriptor representing a callable event on a class descending from
:class:`EventsMixin`.
Instances of this class are very similar to a :class:`property` but also
deal with notifying the owning class when events are assigned (or
unassigned) and wrapping callbacks implicitly as appropriate.
"""
def __init__(self, doc=None):
self.handlers = {}
self.__doc__ = doc
def _wrap_callback(self, instance, fn):
if not callable(fn):
raise BadEventHandler('value must be None or a callable')
# If fn is wrapped with partial (i.e. partial, partialmethod, or wraps
# has been used to produce it) we need to dig out the "real" function
# that's been wrapped along with all the mandatory positional args
# used in the wrapper so we can test the binding
args = ()
wrapped_fn = fn
while isinstance(wrapped_fn, partial):
args = wrapped_fn.args + args
wrapped_fn = wrapped_fn.func
if inspect.isbuiltin(wrapped_fn):
# We can't introspect the prototype of builtins. In this case we
# assume that the builtin has no (mandatory) parameters; this is
# the most reasonable assumption on the basis that pre-existing
# builtins have no knowledge of gpiozero, and the sole parameter
# we would pass is a gpiozero object
return fn
else:
# Try binding ourselves to the argspec of the provided callable.
# If this works, assume the function is capable of accepting no
# parameters
try:
inspect.getcallargs(wrapped_fn, *args)
return fn
except TypeError:
try:
# If the above fails, try binding with a single parameter
# (ourselves). If this works, wrap the specified callback
inspect.getcallargs(wrapped_fn, *(args + (instance,)))
@wraps(fn)
def wrapper():
return fn(instance)
return wrapper
except TypeError:
raise BadEventHandler(
'value must be a callable which accepts up to one '
'mandatory parameter')
def __get__(self, instance, owner=None):
if instance is None:
return self
else:
return self.handlers.get(id(instance))
def __set__(self, instance, value):
if value is None:
try:
del self.handlers[id(instance)]
except KeyError:
warnings.warn(CallbackSetToNone(callback_warning))
else:
self.handlers[id(instance)] = self._wrap_callback(instance, value)
enabled = any(
obj.handlers.get(id(instance))
for name in dir(type(instance))
for obj in (getattr(type(instance), name),)
if isinstance(obj, event)
)
instance._start_stop_events(enabled)
class EventsMixin:
"""
Adds edge-detected :meth:`when_activated` and :meth:`when_deactivated`
events to a device based on changes to the :attr:`~Device.is_active`
property common to all devices. Also adds :meth:`wait_for_active` and
:meth:`wait_for_inactive` methods for level-waiting.
.. note::
Note that this mixin provides no means of actually firing its events;
call :meth:`_fire_events` in sub-classes when device state changes to
trigger the events. This should also be called once at the end of
initialization to set initial states.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._active_event = Event()
self._inactive_event = Event()
self._last_active = None
self._last_changed = self.pin_factory.ticks()
def _all_events(self):
"""
Generator function which yields all :class:`event` instances defined
against this class.
"""
for name in dir(type(self)):
obj = getattr(type(self), name)
if isinstance(obj, event):
yield obj
def close(self):
for ev in self._all_events():
try:
del ev.handlers[id(self)]
except KeyError:
pass
super().close()
def wait_for_active(self, timeout=None):
"""
Pause the script until the device is activated, or the timeout is
reached.
:type timeout: float or None
:param timeout:
Number of seconds to wait before proceeding. If this is
:data:`None` (the default), then wait indefinitely until the device
is active.
"""
return self._active_event.wait(timeout)
def wait_for_inactive(self, timeout=None):
"""
Pause the script until the device is deactivated, or the timeout is
reached.
:type timeout: float or None
:param timeout:
Number of seconds to wait before proceeding. If this is
:data:`None` (the default), then wait indefinitely until the device
is inactive.
"""
return self._inactive_event.wait(timeout)
when_activated = event(
"""
The function to run when the device changes state from inactive to
active.
This can be set to a function which accepts no (mandatory) parameters,
or a Python function which accepts a single mandatory parameter (with
as many optional parameters as you like). If the function accepts a
single mandatory parameter, the device that activated it will be passed
as that parameter.
Set this property to :data:`None` (the default) to disable the event.
""")
when_deactivated = event(
"""
The function to run when the device changes state from active to
inactive.
This can be set to a function which accepts no (mandatory) parameters,
or a Python function which accepts a single mandatory parameter (with
as many optional parameters as you like). If the function accepts a
single mandatory parameter, the device that deactivated it will be
passed as that parameter.
Set this property to :data:`None` (the default) to disable the event.
""")
@property
def active_time(self):
"""
The length of time (in seconds) that the device has been active for.
When the device is inactive, this is :data:`None`.
"""
if self._active_event.is_set():
return self.pin_factory.ticks_diff(self.pin_factory.ticks(),
self._last_changed)
else:
return None
@property
def inactive_time(self):
"""
The length of time (in seconds) that the device has been inactive for.
When the device is active, this is :data:`None`.
"""
if self._inactive_event.is_set():
return self.pin_factory.ticks_diff(self.pin_factory.ticks(),
self._last_changed)
else:
return None
def _fire_activated(self):
# These methods are largely here to be overridden by descendents
if self.when_activated:
self.when_activated()
def _fire_deactivated(self):
# These methods are largely here to be overridden by descendents
if self.when_deactivated:
self.when_deactivated()
def _fire_events(self, ticks, new_active):
"""
This method should be called by descendents whenever the
:attr:`~Device.is_active` property is likely to have changed (for
example, in response to a pin's :attr:`~gpiozero.Pin.state` changing).
The *ticks* parameter must be set to the time when the change occurred;
this can usually be obtained from the pin factory's
:meth:`gpiozero.Factory.ticks` method but some pin implementations will
implicitly provide the ticks when an event occurs as part of their
reporting mechanism.
The *new_active* parameter must be set to the device's
:attr:`~Device.is_active` value at the time indicated by *ticks* (which
is not necessarily the value of :attr:`~Device.is_active` right now, if
the pin factory provides means of reporting a pin's historical state).
"""
old_active, self._last_active = self._last_active, new_active
if old_active is None:
# Initial "indeterminate" state; set events but don't fire
# callbacks as there's not necessarily an edge
if new_active:
self._active_event.set()
else:
self._inactive_event.set()
elif old_active != new_active:
self._last_changed = ticks
if new_active:
self._inactive_event.clear()
self._active_event.set()
self._fire_activated()
else:
self._active_event.clear()
self._inactive_event.set()
self._fire_deactivated()
def _start_stop_events(self, enabled):
"""
This is a stub method that only exists to be overridden by descendents.
It is called when :class:`event` properties are assigned (including
when set to :data:`None) to permit the owning instance to activate or
deactivate monitoring facilities.
For example, if a descendent requires a background thread to monitor a
device, it would be preferable to only run the thread if event handlers
are present to respond to it.
The *enabled* parameter is :data:`False` when all :class:`event`
properties on the owning class are :data:`None`, and :data:`True`
otherwise.
"""
pass
class HoldMixin(EventsMixin):
"""
Extends :class:`EventsMixin` to add the :attr:`when_held` event and the
machinery to fire that event repeatedly (when :attr:`hold_repeat` is
:data:`True`) at internals defined by :attr:`hold_time`.
"""
def __init__(self, *args, **kwargs):
self._hold_thread = None
super().__init__(*args, **kwargs)
self._when_held = None
self._held_from = None
self._hold_time = 1
self._hold_repeat = False
self._hold_thread = HoldThread(self)
def close(self):
if self._hold_thread is not None:
self._hold_thread.stop()
self._hold_thread = None
super().close()
def _fire_activated(self):
super()._fire_activated()
self._hold_thread.holding.set()
def _fire_deactivated(self):
self._held_from = None
super()._fire_deactivated()
def _fire_held(self):
if self.when_held:
self.when_held()
when_held = event(
"""
The function to run when the device has remained active for
:attr:`hold_time` seconds.
This can be set to a function which accepts no (mandatory) parameters,
or a Python function which accepts a single mandatory parameter (with
as many optional parameters as you like). If the function accepts a
single mandatory parameter, the device that activated will be passed
as that parameter.
Set this property to :data:`None` (the default) to disable the event.
""")
@property
def hold_time(self):
"""
The length of time (in seconds) to wait after the device is activated,
until executing the :attr:`when_held` handler. If :attr:`hold_repeat`
is True, this is also the length of time between invocations of
:attr:`when_held`.
"""
return self._hold_time
@hold_time.setter
def hold_time(self, value):
if value < 0:
raise BadWaitTime('hold_time must be 0 or greater')
self._hold_time = float(value)
@property
def hold_repeat(self):
"""
If :data:`True`, :attr:`when_held` will be executed repeatedly with
:attr:`hold_time` seconds between each invocation.
"""
return self._hold_repeat
@hold_repeat.setter
def hold_repeat(self, value):
self._hold_repeat = bool(value)
@property
def is_held(self):
"""
When :data:`True`, the device has been active for at least
:attr:`hold_time` seconds.
"""
return self._held_from is not None
@property
def held_time(self):
"""
The length of time (in seconds) that the device has been held for.
This is counted from the first execution of the :attr:`when_held` event
rather than when the device activated, in contrast to
:attr:`~EventsMixin.active_time`. If the device is not currently held,
this is :data:`None`.
"""
if self._held_from is not None:
return self.pin_factory.ticks_diff(self.pin_factory.ticks(),
self._held_from)
else:
return None
class HoldThread(GPIOThread):
"""
Extends :class:`GPIOThread`. Provides a background thread that repeatedly
fires the :attr:`HoldMixin.when_held` event as long as the owning
device is active.
"""
def __init__(self, parent):
super().__init__(
target=self.held, args=(weakref.proxy(parent),))
self.holding = Event()
self.start()
def held(self, parent):
try:
while not self.stopping.is_set():
if self.holding.wait(0.1):
self.holding.clear()
while not (
self.stopping.is_set() or
parent._inactive_event.wait(parent.hold_time)
):
if parent._held_from is None:
parent._held_from = parent.pin_factory.ticks()
parent._fire_held()
if not parent.hold_repeat:
break
except ReferenceError:
# Parent is dead; time to die!
pass
class GPIOQueue(GPIOThread):
"""
Extends :class:`GPIOThread`. Provides a background thread that monitors a
device's values and provides a running *average* (defaults to median) of
those values. If the *parent* device includes the :class:`EventsMixin` in
its ancestry, the thread automatically calls
:meth:`~EventsMixin._fire_events`.
"""
def __init__(
self, parent, queue_len=5, sample_wait=0.0, partial=False,
average=median, ignore=None):
assert callable(average)
if queue_len < 1:
raise BadQueueLen('queue_len must be at least one')
if sample_wait < 0:
raise BadWaitTime('sample_wait must be 0 or greater')
if ignore is None:
ignore = set()
super().__init__(target=self.fill)
self.queue = deque(maxlen=queue_len)
self.partial = bool(partial)
self.sample_wait = float(sample_wait)
self.full = Event()
self.parent = weakref.proxy(parent)
self.average = average
self.ignore = ignore
@property
def value(self):
if not self.partial:
self.full.wait()
try:
return self.average(self.queue)
except (ZeroDivisionError, ValueError):
# No data == inactive value
return 0.0
def fill(self):
try:
while not self.stopping.wait(self.sample_wait):
value = self.parent._read()
if value not in self.ignore:
self.queue.append(value)
if not self.full.is_set() and len(self.queue) >= self.queue.maxlen:
self.full.set()
if (self.partial or self.full.is_set()) and isinstance(self.parent, EventsMixin):
self.parent._fire_events(self.parent.pin_factory.ticks(), self.parent.is_active)
except ReferenceError:
# Parent is dead; time to die!
pass