revil-O/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

Added support for continuous checks on distance sensor and DistanceSensorEvent

Browse source
This commit is contained in:
Fabio Manganiello 2019-12-22 19:09:02 +01:00
parent 9c4f917b53
commit d6515ed991
4 changed files with 130 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
platypush/message/event/__init__.py
View file

@ -11,6 +11,7 @@ from platypush.config import Config
from platypush.message import Message
from platypush.utils import get_event_class_by_type
class Event(Message):
""" Event message class """
@ -39,6 +40,9 @@ class Event(Message):
self.args = kwargs
self.disable_logging = disable_logging
for arg, value in self.args.items():
self.__setattr__(arg, value)
@classmethod
def build(cls, msg):
""" Builds an event message from a JSON UTF-8 string/bytearray, a
@ -55,7 +59,6 @@ class Event(Message):
args['timestamp'] = msg['_timestamp'] if '_timestamp' in msg else time.time()
return event_class(**args)
@staticmethod
def _generate_id():
""" Generate a unique event ID """
@ -64,7 +67,6 @@ class Event(Message):
id += '%.2x' % random.randint(0, 255)
return id
def matches_condition(self, condition):
"""
If the event matches an event condition, it will return an EventMatchResult
@ -75,7 +77,8 @@ class Event(Message):
result = EventMatchResult(is_match=False, parsed_args=self.args)
match_scores = []
if not isinstance(self, condition.type): return result
if not isinstance(self, condition.type):
return result
for (attr, value) in condition.args.items():
if attr not in self.args:
@ -100,7 +103,6 @@ class Event(Message):
return result
def _matches_argument(self, argname, condition_value):
"""
Returns an EventMatchResult if the event argument [argname] matches
@ -147,9 +149,8 @@ class Event(Message):
else:
result.parsed_args[argname] += ' ' + event_token
if (len(condition_tokens) == 1 and len(event_tokens) == 1) \
or (len(event_tokens) > 1 and len(condition_tokens) > 1 \
or (len(event_tokens) > 1 and len(condition_tokens) > 1
and event_tokens[1] == condition_tokens[1]):
# Stop appending tokens to this argument, as the next
# condition will be satisfied as well
@ -164,7 +165,6 @@ class Event(Message):
result.is_match = len(condition_tokens) == 0
return result
def __str__(self):
"""
Overrides the str() operator and converts
@ -234,10 +234,8 @@ def flatten(args):
elif isinstance(args, list):
for i in range(0, len(args)):
if isinstance(args[i], date):
args[i] = value.isoformat()
args[i] = args[i].isoformat()
elif isinstance(args[i], dict) or isinstance(args[i], list):
flatten(args[i])
# vim:sw=4:ts=4:et:

12
platypush/message/event/distance.py Normal file
View file

@ -0,0 +1,12 @@
from platypush.message.event import Event
class DistanceSensorEvent(Event):
"""
Event triggered when a new value is processed by a distance sensor.
"""
def __init__(self, distance: float, unit: str = 'mm', *args, **kwargs):
super().__init__(*args, distance=distance, unit=unit, **kwargs)
# vim:sw=4:ts=4:et:

3
platypush/plugins/gpio/__init__.py
View file

@ -176,6 +176,9 @@ class GpioPlugin(Plugin):
Cleanup the state of the GPIO and resets PIN values.
"""
import RPi.GPIO as GPIO
with self._init_lock:
if self._initialized:
GPIO.cleanup()
self._initialized_pins = {}
self._initialized = False

97
platypush/plugins/gpio/sensor/distance/__init__.py
View file

@ -1,5 +1,10 @@
import threading
import time
from typing import Optional
from platypush.context import get_bus
from platypush.message.event.distance import DistanceSensorEvent
from platypush.plugins import action
from platypush.plugins.gpio import GpioPlugin
from platypush.plugins.gpio.sensor import GpioSensorPlugin
@ -13,6 +18,11 @@ class GpioSensorDistancePlugin(GpioPlugin, GpioSensorPlugin):
Requires:
* ``RPi.GPIO`` (``pip install RPi.GPIO``)
Triggers:
* :class:`platypush.message.event.distance.DistanceSensorEvent` when a new distance measurement is available
"""
def __init__(self, trigger_pin: int, echo_pin: int,
@ -31,20 +41,27 @@ class GpioSensorDistancePlugin(GpioPlugin, GpioSensorPlugin):
for the sensor to be ready (default: 2 seconds).
"""
GpioPlugin.__init__(self, *args, **kwargs)
GpioPlugin.__init__(self, pins={'trigger': trigger_pin, 'echo': echo_pin, }, *args, **kwargs)
self.trigger_pin = trigger_pin
self.echo_pin = echo_pin
self.timeout = timeout
self.warmup_time = warmup_time
self._initialized = False
self._init_gpio()
self._measurement_thread: Optional[threading.Thread] = None
self._measurement_thread_lock = threading.RLock()
self._measurement_thread_can_run = False
self._init_board()
def _init_gpio(self):
def _init_board(self):
import RPi.GPIO as GPIO
with self._init_lock:
if self._initialized:
return
import RPi.GPIO as GPIO
GPIO.setmode(self.mode)
GpioPlugin._init_board(self)
self._initialized = False
GPIO.setup(self.trigger_pin, GPIO.OUT)
GPIO.setup(self.echo_pin, GPIO.IN)
GPIO.output(self.trigger_pin, GPIO.LOW)
@ -59,7 +76,7 @@ class GpioSensorDistancePlugin(GpioPlugin, GpioSensorPlugin):
pulse_start = pulse_on = time.time()
self._init_gpio()
self._init_board()
GPIO.output(self.trigger_pin, GPIO.HIGH)
time.sleep(0.00001) # 1 us pulse to trigger echo measurement
GPIO.output(self.trigger_pin, GPIO.LOW)
@ -94,7 +111,10 @@ class GpioSensorDistancePlugin(GpioPlugin, GpioSensorPlugin):
"""
try:
return self._get_data()
distance = self._get_data()
bus = get_bus()
bus.post(DistanceSensorEvent(distance=distance, unit='mm'))
return distance
except TimeoutError as e:
self.logger.warning(str(e))
return
@ -104,16 +124,67 @@ class GpioSensorDistancePlugin(GpioPlugin, GpioSensorPlugin):
@action
def close(self):
import RPi.GPIO as GPIO
if self._initialized:
GPIO.cleanup()
self._initialized = False
return self.cleanup()
def __enter__(self):
self._init_gpio()
self._init_board()
def __exit__(self):
self.close()
def _get_measurement_thread(self, duration: float):
def _thread():
with self:
start_time = time.time()
try:
while self._measurement_thread_can_run and (
not duration or time.time() - start_time <= duration):
self.get_measurement()
finally:
self._measurement_thread = None
return _thread
def _is_measurement_thread_running(self):
with self._measurement_thread_lock:
return self._measurement_thread is not None
@action
def start_measurement(self, duration: Optional[float] = None):
"""
Start the measurement thread. It will trigger :class:`platypush.message.event.distance.DistanceSensorEvent`
events when new measurements are available.
:param duration: If set, then the thread will run for the specified amount of seconds (default: None)
"""
with self._measurement_thread_lock:
if self._is_measurement_thread_running():
self.logger.warning('A measurement thread is already running')
return
thread_func = self._get_measurement_thread(duration=duration)
self._measurement_thread = threading.Thread(target=thread_func)
self._measurement_thread_can_run = True
self._measurement_thread.start()
@action
def stop_measurement(self):
"""
Stop the running measurement thread.
"""
with self._measurement_thread_lock:
if not self._is_measurement_thread_running():
self.logger.warning('No measurement thread is running')
return
self._measurement_thread_can_run = False
self.logger.info('Waiting for the measurement thread to end')
if self._measurement_thread:
self._measurement_thread.join(timeout=self.timeout)
self.logger.info('Measurement thread terminated')
# vim:sw=4:ts=4:et: