Source code for platypush.plugins.leap

from threading import Timer
from multiprocessing import Process
from typing import Iterable, Optional

import Leap

from platypush.context import get_bus
from platypush.message.event.sensor.leap import (
    LeapFrameEvent,
    LeapFrameStartEvent,
    LeapFrameStopEvent,
    LeapConnectEvent,
    LeapDisconnectEvent,
)
from platypush.plugins import RunnablePlugin


[docs] class LeapPlugin(RunnablePlugin): """ Integration to handle events from a `Leap Motion <https://www.leapmotion.com/>`_ device to track hands and gestures. Note that the default SDK is not compatible with Python 3. Follow the instructions on https://github.com/BlackLight/leap-sdk-python3 to build the Python 3 module. Also, you'll need the Leap driver and utils installed on your OS (follow instructions at https://www.leapmotion.com/setup/) and the `leapd` daemon running to recognize your controller. Requires: * The Leap Motion SDK compiled with Python 3 support, see my port at https://github.com:BlackLight/leap-sdk-python3.git * The ``leapd`` daemon to be running and your Leap Motion connected """ _listener_proc = None
[docs] def __init__( self, position_ranges: Optional[Iterable[Iterable[float]]] = None, position_tolerance: float = 0.0, poll_interval: Optional[float] = 0.1, **kwargs ): """ :param position_ranges: It specifies how wide the hand space (x, y and z axes) should be in millimiters. Default:: [ [-300.0, 300.0], # x axis [25.0, 600.0], # y axis [-300.0, 300.0], # z axis ] :param position_tolerance: % of change between a frame and the next to really consider the next frame as a new one (default: 0). :param poll_interval: How often the plugin should generate and push events (default: at most one event each 0.1 seconds). Note that a Leap Motion generates events with a very high throughput, so you may want to set a higher value if you want to throttle the events and avoid flooding the bus. """ super().__init__(poll_interval=poll_interval, **kwargs) if position_ranges is None: position_ranges = [ [-300.0, 300.0], # x axis [25.0, 600.0], # y axis [-300.0, 300.0], # z axis ] self.position_ranges = position_ranges self.position_tolerance = position_tolerance
[docs] def main(self): def _listener_process(): listener = LeapListener( position_ranges=self.position_ranges, position_tolerance=self.position_tolerance, frames_throttle_secs=self.poll_interval, logger=self.logger, ) controller = Leap.Controller() if not controller: raise RuntimeError( 'No Leap Motion controller found - is your ' + 'device connected and is leapd running?' ) controller.add_listener(listener) self.logger.info('Leap Motion device initialized') try: self.wait_stop() except KeyboardInterrupt: self.logger.info('Terminating Leap Motion listener') while not self.should_stop(): self.wait_stop(1) self._listener_proc = Process(target=_listener_process) self._listener_proc.start() self._listener_proc.join() self._listener_proc = None
[docs] class LeapFuture(Timer):
[docs] def __init__(self, seconds, listener, event): self.listener = listener self.event = event super().__init__(seconds, self._callback_wrapper())
def _callback_wrapper(self): def _callback(): self.listener._send_event(self.event) # pylint: disable=protected-access return _callback
[docs] class LeapListener(Leap.Listener):
[docs] def __init__( self, position_ranges, position_tolerance, logger, frames_throttle_secs=None ): super().__init__() self.prev_frame = None self.position_ranges = position_ranges self.position_tolerance = position_tolerance self.frames_throttle_secs = frames_throttle_secs self.logger = logger self.running_future = None
def _send_event(self, event): get_bus().post(event) def send_event(self, event): if self.frames_throttle_secs: if not self.running_future or not self.running_future.is_alive(): self.running_future = LeapFuture( seconds=self.frames_throttle_secs, listener=self, event=event ) self.running_future.start() else: self._send_event(event) def on_init(self, *_, **__): self.prev_frame = None self.logger.info('Leap controller listener initialized') def on_connect(self, *_, **__): self.logger.info('Leap controller connected') self.prev_frame = None self.send_event(LeapConnectEvent()) def on_disconnect(self, *_, **__): self.logger.info('Leap controller disconnected') self.prev_frame = None self.send_event(LeapDisconnectEvent()) def on_exit(self, *_, **__): self.logger.info('Leap listener terminated') def on_frame(self, controller): frame = controller.frame() if len(frame.hands) > 0: hands = self._flatten_hands(frame) if hands: if not self.prev_frame: self.send_event(LeapFrameStartEvent()) self.send_event(LeapFrameEvent(hands=hands)) self.prev_frame = frame else: if self.prev_frame: self.send_event(LeapFrameStopEvent()) self.prev_frame = None def _flatten_hands(self, frame): return [ { 'confidence': hand.confidence, 'direction': [hand.direction[0], hand.direction[1], hand.direction[2]], 'id': hand.id, 'is_left': hand.is_left, 'is_right': hand.is_right, 'palm_normal': [ hand.palm_normal[0], hand.palm_normal[1], hand.palm_normal[2], ], 'palm_position': self._normalize_position(hand.palm_position), 'palm_velocity': [ hand.palm_velocity[0], hand.palm_velocity[1], hand.palm_velocity[2], ], 'palm_width': hand.palm_width, 'sphere_center': [ hand.sphere_center[0], hand.sphere_center[1], hand.sphere_center[2], ], 'sphere_radius': hand.sphere_radius, 'stabilized_palm_position': self._normalize_position( hand.stabilized_palm_position ), 'time_visible': hand.time_visible, 'wrist_position': self._normalize_position(hand.wrist_position), } for i, hand in enumerate(frame.hands) if hand.is_valid and ( len(frame.hands) != len(self.prev_frame.hands) or self._position_changed( old_position=self.prev_frame.hands[i].stabilized_palm_position, new_position=hand.stabilized_palm_position, ) if self.prev_frame else True ) ] def _normalize_position(self, position): # Normalize absolute position onto a hemisphere centered in (0,0) # having x_range = z_range = [-100, 100], y_range = [0, 100] return [ self._scale_scalar( value=position[0], range=self.position_ranges[0], new_range=[-100.0, 100.0], ), self._scale_scalar( value=position[1], range=self.position_ranges[1], new_range=[0.0, 100.0] ), self._scale_scalar( value=position[2], range=self.position_ranges[2], new_range=[-100.0, 100.0], ), ] @staticmethod def _scale_scalar(value, range, new_range): if value < range[0]: value = range[0] if value > range[1]: value = range[1] return ((new_range[1] - new_range[0]) / (range[1] - range[0])) * ( value - range[0] ) + new_range[0] def _position_changed(self, old_position, new_position): return ( abs(old_position[0] - new_position[0]) > self.position_tolerance or abs(old_position[1] - new_position[1]) > self.position_tolerance or abs(old_position[2] - new_position[2]) > self.position_tolerance )
# vim:sw=4:ts=4:et: