#!/usr/bin/env python3 # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # import argparse import json import logging import sys from gi.repository import GObject from tuhi.dbusserver import TuhiDBusServer from tuhi.ble import BlueZDeviceManager from tuhi.wacom import WacomDevice, Stroke logging.basicConfig(format='%(levelname)s: %(name)s: %(message)s', level=logging.INFO) logger = logging.getLogger('tuhi') WACOM_COMPANY_ID = 0x4755 class TuhiDrawing(object): class Stroke(object): def __init__(self): self.points = [] def to_dict(self): d = {} d['points'] = [p.to_dict() for p in self.points] return d class Point(object): def __init__(self): pass def to_dict(self): d = {} for key in ['toffset', 'position', 'pressure']: val = getattr(self, key, None) if val is not None: d[key] = val return d def __init__(self, name, dimensions, timestamp): self.name = name self.dimensions = dimensions self.timestamp = timestamp self.strokes = [] def json(self): JSON_FILE_FORMAT_VERSION = 1 json_data = { 'version': JSON_FILE_FORMAT_VERSION, 'devicename': self.name, 'dimensions': list(self.dimensions), 'timestamp': self.timestamp, 'strokes': [s.to_dict() for s in self.strokes] } return json.dumps(json_data) class TuhiDevice(GObject.Object): """ Glue object to combine the backend bluez DBus object (that talks to the real device) with the frontend DBusServer object that exports the device over Tuhi's DBus interface """ def __init__(self, bluez_device, tuhi_dbus_device): GObject.Object.__init__(self) self._tuhi_dbus_device = tuhi_dbus_device self._wacom_device = WacomDevice(bluez_device) self._wacom_device.connect('drawing', self._on_drawing_received) self._wacom_device.connect('done', self._on_fetching_finished, bluez_device) self.drawings = [] self.pairing_mode = False bluez_device.connect('connected', self._on_bluez_device_connected) bluez_device.connect('disconnected', self._on_bluez_device_disconnected) self._bluez_device = bluez_device def connect_device(self): self._bluez_device.connect_device() def _on_bluez_device_connected(self, bluez_device): logger.debug('{}: connected'.format(bluez_device.address)) self._wacom_device.start(self.pairing_mode) self.pairing_mode = False def _on_bluez_device_disconnected(self, bluez_device): logger.debug('{}: disconnected'.format(bluez_device.address)) def _on_drawing_received(self, device, drawing): logger.debug('Drawing received') d = TuhiDrawing(device.name, (0, 0), drawing.timestamp) for s in drawing: stroke = TuhiDrawing.Stroke() lastx, lasty, lastp = None, None, None for type, x, y, p in s.points: if x is not None: if type == Stroke.RELATIVE: x += lastx lastx = x if y is not None: if type == Stroke.RELATIVE: y += lasty lasty = y if p is not None: if type == Stroke.RELATIVE: p += lastp lastp = p lastx, lasty, lastp = x, y, p point = TuhiDrawing.Point() point.position = (lastx, lasty) point.pressure = lastp stroke.points.append(point) d.strokes.append(stroke) self._tuhi_dbus_device.add_drawing(d) def _on_fetching_finished(self, device, bluez_device): bluez_device.disconnect_device() class Tuhi(GObject.Object): __gsignals__ = { "device-added": (GObject.SIGNAL_RUN_FIRST, None, (GObject.TYPE_PYOBJECT,)), "device-connected": (GObject.SIGNAL_RUN_FIRST, None, (GObject.TYPE_PYOBJECT,)), } def __init__(self): GObject.Object.__init__(self) self.server = TuhiDBusServer() self.server.connect('bus-name-acquired', self._on_tuhi_bus_name_acquired) self.server.connect('pairing-start-requested', self._on_start_pairing_requested) self.server.connect('pairing-stop-requested', self._on_stop_pairing_requested) self.server.connect('pair-device-requested', self._on_pair_device_requested) self.bluez = BlueZDeviceManager() self.bluez.connect('device-added', self._on_bluez_device_added) self.bluez.connect('device-updated', self._on_bluez_device_updated) self.bluez.connect('discovery-started', self._on_bluez_discovery_started) self.bluez.connect('discovery-stopped', self._on_bluez_discovery_stopped) self.devices = {} self._pairing_stop_handler = None def _on_tuhi_bus_name_acquired(self, dbus_server): self.bluez.connect_to_bluez() def _on_start_pairing_requested(self, dbus_server, stop_handler): self._pairing_stop_handler = stop_handler self.bluez.start_discovery(timeout=30) def _on_stop_pairing_requested(self, dbus_server): # If you request to stop, you get a successful stop and we ignore # anything the server does underneath self._pairing_stop_handler(0) self._pairing_stop_handler = None self.bluez.stop_discovery() self._pairable_device_handler = None def _on_pair_device_requested(self, dbusserver, bluez_device): tuhi_dbus_device = self.server.create_device(bluez_device) d = TuhiDevice(bluez_device, tuhi_dbus_device) d.pairing_mode = True self.devices[bluez_device.address] = d d.connect_device() @classmethod def _is_pairing_device(cls, bluez_device): if bluez_device.vendor_id != WACOM_COMPANY_ID: return False manufacturer_data = bluez_device.get_manufacturer_data(WACOM_COMPANY_ID) return manufacturer_data is not None and len(manufacturer_data) == 4 def _on_bluez_device_added(self, manager, bluez_device): if bluez_device.vendor_id != WACOM_COMPANY_ID: return if Tuhi._is_pairing_device(bluez_device): return tuhi_dbus_device = self.server.create_device(bluez_device) d = TuhiDevice(bluez_device, tuhi_dbus_device) self.devices[bluez_device.address] = d def _on_bluez_discovery_started(self, manager): # Something else may turn discovery mode on, we don't care about # it then if not self._pairing_stop_handler: return def _on_bluez_discovery_stopped(self, manager): if self._pairing_stop_handler is not None: self._pairing_stop_handler(0) def _on_bluez_device_updated(self, manager, bluez_device): if bluez_device.vendor_id != WACOM_COMPANY_ID: return if Tuhi._is_pairing_device(bluez_device): self.server.notify_pairable_device(bluez_device) def main(args): desc = "Daemon to extract the pen stroke data from Wacom SmartPad devices" parser = argparse.ArgumentParser(description=desc) parser.add_argument('-v', '--verbose', help='Show some debugging informations', action='store_true', default=False) ns = parser.parse_args(args[1:]) if ns.verbose: logger.setLevel(logging.DEBUG) Tuhi() try: GObject.MainLoop().run() except KeyboardInterrupt: pass finally: pass if __name__ == "__main__": main(sys.argv)