#!/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. # from gi.repository import GObject, Gio, GLib import argparse import errno import os import logging import re import xdg.BaseDirectory logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger('tuhi') CONFIG_PATH = os.path.join(xdg.BaseDirectory.xdg_data_home, 'tuhi-kete') TUHI_DBUS_NAME = 'org.freedesktop.tuhi1' ORG_FREEDESKTOP_TUHI1_MANAGER = 'org.freedesktop.tuhi1.Manager' ORG_FREEDESKTOP_TUHI1_DEVICE = 'org.freedesktop.tuhi1.Device' ROOT_PATH = '/org/freedesktop/tuhi1' ORG_BLUEZ_DEVICE1 = 'org.bluez.Device1' class DBusError(Exception): def __init__(self, message): self.message = message class _DBusObject(GObject.Object): _connection = None def __init__(self, name, interface, objpath): GObject.GObject.__init__(self) # this is not handled asynchronously because if we fail to # get the session bus, we have other issues if _DBusObject._connection is None: self._connect_to_session() self.interface = interface self.objpath = objpath self._connected = False self._name = name try: self._connect() except DBusError: self._reconnect_timer = GObject.timeout_add_seconds(2, self._on_reconnect_timer) def _connect(self): try: self.proxy = Gio.DBusProxy.new_sync(self._connection, Gio.DBusProxyFlags.NONE, None, self._name, self.objpath, self.interface, None) if self.proxy.get_name_owner() is None: raise DBusError(f'No-one is handling {self._name}, is the daemon running?') self._connected = True self.notify('connected') except GLib.Error as e: if (e.domain == 'g-io-error-quark' and e.code == Gio.IOErrorEnum.DBUS_ERROR): raise DBusError(e.message) else: raise e self.proxy.connect('g-properties-changed', self._on_properties_changed) self.proxy.connect('g-signal', self._on_signal_received) def _on_reconnect_timer(self): try: logger.debug('reconnecting') self._connect() return False except DBusError: return True @GObject.Property def connected(self): return self._connected def _connect_to_session(self): try: _DBusObject._connection = Gio.bus_get_sync(Gio.BusType.SESSION, None) except GLib.Error as e: if (e.domain == 'g-io-error-quark' and e.code == Gio.IOErrorEnum.DBUS_ERROR): raise DBusError(e.message) else: raise e def _on_properties_changed(self, proxy, changed_props, invalidated_props): # Implement this in derived classes to respond to property changes pass def _on_signal_received(self, proxy, sender, signal, parameters): # Implement this in derived classes to respond to signals pass def property(self, name): p = self.proxy.get_cached_property(name) if p is not None: return p.unpack() return p def terminate(self): del(self.proxy) class _DBusSystemObject(_DBusObject): ''' Same as the _DBusObject, but connects to the system bus instead ''' def __init__(self, name, interface, objpath): self._connect_to_system() super().__init__(name, interface, objpath) def _connect_to_system(self): try: self._connection = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) except GLib.Error as e: if (e.domain == 'g-io-error-quark' and e.code == Gio.IOErrorEnum.DBUS_ERROR): raise DBusError(e.message) else: raise e class BlueZDevice(_DBusSystemObject): def __init__(self, objpath): super().__init__('org.bluez', ORG_BLUEZ_DEVICE1, objpath) self.proxy.connect('g-properties-changed', self._on_properties_changed) @GObject.Property def connected(self): return self.proxy.get_cached_property('Connected').unpack() def _on_properties_changed(self, obj, properties, invalidated_properties): properties = properties.unpack() if 'Connected' in properties: self.notify('connected') class TuhiKeteDevice(_DBusObject): __gsignals__ = { 'button-press-required': (GObject.SignalFlags.RUN_FIRST, None, (GObject.TYPE_PYOBJECT,)), 'registered': (GObject.SignalFlags.RUN_FIRST, None, (GObject.TYPE_PYOBJECT,)), } def __init__(self, manager, objpath): _DBusObject.__init__(self, TUHI_DBUS_NAME, ORG_FREEDESKTOP_TUHI1_DEVICE, objpath) self.manager = manager self.is_registering = False self._bluez_device = BlueZDevice(self.property('BlueZDevice')) self._bluez_device.connect('notify::connected', self._on_connected) @classmethod def is_device_address(cls, string): if re.match(r'[0-9a-f]{2}(:[0-9a-f]{2}){5}$', string.lower()): return string raise argparse.ArgumentTypeError(f'"{string}" is not a valid device address') @GObject.Property def address(self): return self._bluez_device.property('Address') @GObject.Property def name(self): return self._bluez_device.property('Name') @GObject.Property def listening(self): return self.property('Listening') @GObject.Property def drawings_available(self): return self.property('DrawingsAvailable') @GObject.Property def battery_percent(self): return self.property('BatteryPercent') @GObject.Property def battery_state(self): return self.property('BatteryState') @GObject.Property def connected(self): return self._bluez_device.connected def _on_connected(self, bluez_device, pspec): self.notify('connected') def register(self): logger.debug(f'{self}: Register') # FIXME: Register() doesn't return anything useful yet, so we wait until # the device is in the Manager's Devices property self.s1 = self.manager.connect('notify::devices', self._on_mgr_devices_updated) self.is_registering = True self.proxy.Register() def start_listening(self): self.proxy.StartListening() def stop_listening(self): try: self.proxy.StopListening() except GLib.Error as e: if (e.domain != 'g-dbus-error-quark' or e.code != Gio.IOErrorEnum.EXISTS or Gio.dbus_error_get_remote_error(e) != 'org.freedesktop.DBus.Error.ServiceUnknown'): raise e def json(self, timestamp): SUPPORTED_FILE_FORMAT = 1 return self.proxy.GetJSONData('(ut)', SUPPORTED_FILE_FORMAT, timestamp) def _on_signal_received(self, proxy, sender, signal, parameters): if signal == 'ButtonPressRequired': logger.info(f'{self}: Press button on device now') self.emit('button-press-required', self) elif signal == 'ListeningStopped': err = parameters[0] if err == -errno.EACCES: logger.error(f'{self}: wrong device, please re-register.') elif err < 0: logger.error(f'{self}: an error occured: {os.strerror(-err)}') self.notify('listening') def _on_properties_changed(self, proxy, changed_props, invalidated_props): if changed_props is None: return changed_props = changed_props.unpack() if 'DrawingsAvailable' in changed_props: self.notify('drawings-available') elif 'Listening' in changed_props: self.notify('listening') elif 'BatteryPercent' in changed_props: self.notify('battery-percent') elif 'BatteryState' in changed_props: self.notify('battery-state') def __repr__(self): return f'{self.address} - {self.name}' def _on_mgr_devices_updated(self, manager, pspec): if not self.is_registering: return for d in manager.devices: if d.address == self.address: self.is_registering = False self.manager.disconnect(self.s1) del(self.s1) logger.info(f'{self}: Registration successful') self.emit('registered', self) def terminate(self): try: self.manager.disconnect(self.s1) except AttributeError: pass self._bluez_device.terminate() super(TuhiKeteDevice, self).terminate() class TuhiKeteManager(_DBusObject): __gsignals__ = { 'unregistered-device': (GObject.SignalFlags.RUN_FIRST, None, (GObject.TYPE_PYOBJECT,)), } def __init__(self): _DBusObject.__init__(self, TUHI_DBUS_NAME, ORG_FREEDESKTOP_TUHI1_MANAGER, ROOT_PATH) self._devices = {} self._unregistered_devices = {} logger.info('starting up') if not self.connected: self.connect('notify::connected', self._init) else: self._init() def _init(self, *args, **kwargs): logger.info('connected') for objpath in self.property('Devices'): device = TuhiKeteDevice(self, objpath) self._devices[device.address] = device @GObject.Property def devices(self): return [v for k, v in self._devices.items()] @GObject.Property def unregistered_devices(self): return [v for k, v in self._unregistered_devices.items()] @GObject.Property def searching(self): return self.proxy.get_cached_property('Searching') def start_search(self): self._unregistered_devices = {} self.proxy.StartSearch() def stop_search(self): try: self.proxy.StopSearch() except GLib.Error as e: if (e.domain != 'g-dbus-error-quark' or e.code != Gio.IOErrorEnum.EXISTS or Gio.dbus_error_get_remote_error(e) != 'org.freedesktop.DBus.Error.ServiceUnknown'): raise e self._unregistered_devices = {} def terminate(self): for dev in self._devices.values(): dev.terminate() self._devices = {} self._unregistered_devices = {} super(TuhiKeteManager, self).terminate() def _on_properties_changed(self, proxy, changed_props, invalidated_props): if changed_props is None: return changed_props = changed_props.unpack() if 'Devices' in changed_props: objpaths = changed_props['Devices'] for objpath in objpaths: try: d = self._unregistered_devices[objpath] self._devices[d.address] = d del self._unregistered_devices[objpath] except KeyError: # if we called Register() on an existing device it's not # in unregistered devices pass self.notify('devices') if 'Searching' in changed_props: self.notify('searching') def _handle_unregistered_device(self, objpath): for addr, dev in self._devices.items(): if dev.objpath == objpath: self.emit('unregistered-device', dev) return device = TuhiKeteDevice(self, objpath) self._unregistered_devices[objpath] = device logger.debug(f'New unregistered device: {device}') self.emit('unregistered-device', device) def _on_signal_received(self, proxy, sender, signal, parameters): if signal == 'SearchStopped': self.notify('searching') elif signal == 'UnregisteredDevice': objpath = parameters[0] self._handle_unregistered_device(objpath) def __getitem__(self, btaddr): return self._devices[btaddr]