Source code for pisak.cursor

Module handles cursor-style (stream of coordinates) input in JSON layout.
import time

from gi.repository import GObject, Clutter

from pisak import logger, scanning, configurator, layout, unit, tracker

_LOG = logger.get_logger(__name__)

[docs]class Sprite(layout.Bin, configurator.Configurable): """ Sprite (virtual cursor) object. It displays a big dot in a bright color on the screen which follows input coordinates and selects GUI controls on a timeouted hover. """ __gtype_name__ = "PisakSprite" __gproperties__ = { "timeout": ( GObject.TYPE_UINT, "", "", 0, GObject.G_MAXUINT, 1600, GObject.PARAM_READWRITE) } def __init__(self): super().__init__() self._timeout = 1 self.container = None self.clickables = None self._running = False self.hover_start = None self.hover_actor = None self.initial_position = (-10, -10) self.all_rescan = [] self.coords = (0, 0) self.set_x_expand(True) self.set_y_expand(True) self._init_sprite() self.tracker_client = tracker.TrackerClient(self) self.tracker_client.connect() self.apply_props() def _init_sprite(self): self.sprite = Clutter.Actor() self.sprite.set_size(20, 20) self.sprite.set_background_color(, 255, 0, 255)) self.add_actor(self.sprite) self.update_sprite(self.initial_position) def _rescan(self, *source): self.clickables = None def _do_disconnect(self, obj, func): try: obj.disconnect_by_func(func) except TypeError as e: _LOG.warning(e) def _disconnect_rescan(self): self._do_disconnect(self.container, self._rescan) for obj in self.all_rescan: self._do_disconnect(obj, self._rescan) def _connect_rescan(self): self.all_rescan = [] self.container.connect("allocation-changed", self._rescan) to_conn = self.container.get_children() while len(to_conn) > 0: current = to_conn.pop() current.connect("allocation-changed", self._rescan) self.all_rescan.append(current) to_conn = to_conn + current.get_children() @property def timeout(self): """ Selection hover timeout, in miliseconds. Default is 1 second. """ return self._timeout * 1000 @timeout.setter def timeout(self, value): self._timeout = int(value) / 1000
[docs] def parse_coords(self, data): """ Parses raw data line into x-y coordinates tuple. :param data: raw data with coordinates, being a single-line string in a format: 'screenWidth% screenHeight%'. :return: tuple with cursor parsed x and y coordinates, in pixels as floats. """ if not data: return self.coords try: coords = tuple(float(x) for x in data.split(' ')) coords = (round(coords[0] * unit.size_pix.width), round(coords[1] * unit.size_pix.height)) self.coords = coords except Exception as ex: raise Exception("Error parsing coordinates data: {}".format(ex)) return self.coords
[docs] def update_sprite(self, coords): """ Changes cursor position on the screen. :param coords: tuple with x and y coordinates. """ x, y = (coords[0] - self.sprite.get_width() / 2), (coords[1] - self.sprite.get_height() / 2) self.sprite.set_position(x, y)
[docs] def scan_clickables(self): """ Detects any widgets that could be possibly clicked and puts them on a list used by :func:`find_actor`. """ if self.container is None: self.clickables = [] return to_scan = self.container.get_children() clickables = [] while len(to_scan) > 0: current = to_scan.pop() if isinstance(current, scanning.Scannable): if not current.is_disabled(): clickables.append(current) to_scan = to_scan + current.get_children() self.clickables = clickables _LOG.debug("clickables: {}".format(clickables))
[docs] def find_actor(self, coords): """ Looks for any widget positioned at a given coordinates. If a widget is found then it is returned, otherwise returns None. :param coords: tuple with x-y coordinates. :return: some found widget or None. """ if self.clickables is None: self.scan_clickables() for clickable in self.clickables: (x, y), (w, h) = clickable.get_transformed_position(), clickable.get_size() if (x <= coords[0]) and (coords[0] <= x + w) \ and (y <= coords[1]) and (coords[1] <= y + h): return clickable return None
[docs] def on_new_coords(self, x, y): """ Takes on any cursor-related actions when the new coordinates arrive. Moves cursor, manages widgets highlight, selects widgets if hovered for long enough. :param x: new x coordinate, in pixels, float. :param y: new y coordinate, in pixels, float. :return: False, in order to avoid this function being called again automatically, what otherwise would be the case as long as this function is registered as a Clutter timeout callback from another Python thread for the sake of better Python-threads vs Clutter-GUI cooperation. """ coords = (x, y) self.update_sprite(coords) actor = self.find_actor(coords) if actor is not None: if actor == self.hover_actor: if time.time() - self.hover_start > self._timeout: actor.emit("clicked") self.hover_start = time.time() + 1.0 # dead time else: # reset timeout if self.hover_actor is not None: self.hover_actor.disable_hilite() self.hover_actor = actor self.hover_actor.enable_hilite() self.hover_start = time.time() else: if self.hover_actor is not None: self.hover_actor.disable_hilite() self.hover_actor = None return False
[docs] def on_new_data(self, data): """ Receives new raw data, parses them and schedules calling the main thread callback. :param data: raw data. """ x, y = self.parse_coords(data) Clutter.threads_add_timeout(-100, 20, self.on_new_coords, x, y)
[docs] def run(self, container): """ Displays and starts the sprite. Runs proper tracker websocket client. :param container: widget that the sprite should be placed above. """ self.container = container self._connect_rescan() self._rescan() self.container.insert_child_above(self, None) self.update_sprite(self.initial_position) self._running = True self.tracker_client.activate()
[docs] def stop(self): """ Stops the sprite. Stops the websocket client. """ self._running = False self.tracker_client.deactivate() self.container.remove_child(self) self._disconnect_rescan() self.container = None self.hover_actor = None
[docs] def is_running(self): """ Checks whether the sprite is in a working state. :return: boolean. """ return self._running