| #!/usr/bin/env python3 |
| # -*- Mode: python; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- |
| |
| from __future__ import print_function |
| |
| import gi |
| gi.require_version('ModemManager', '1.0') |
| from gi.repository import GLib, ModemManager |
| import argparse |
| import sys |
| import dbus |
| import dbus.service |
| import dbus.mainloop.glib |
| import random |
| import collections |
| |
| mainloop = GLib.MainLoop() |
| |
| ######################################################### |
| IFACE_DBUS = 'org.freedesktop.DBus' |
| |
| class UnknownInterfaceException(dbus.DBusException): |
| def __init__(self, *args, **kwargs): |
| self._dbus_error_name = '{}.UnknownInterface'.format(IFACE_DBUS) |
| super().__init__(*args, **kwargs) |
| |
| class UnknownPropertyException(dbus.DBusException): |
| def __init__(self, *args, **kwargs): |
| self._dbus_error_name = '{}.UnknownProperty'.format(IFACE_DBUS) |
| super().__init__(*args, **kwargs) |
| |
| class MobileEquipmentException(dbus.DBusException): |
| _dbus_error_name = '{}.Error.MobileEquipment'.format(IFACE_DBUS) |
| def __init__(self, *args, **kwargs): |
| equipment_error_num = kwargs.pop('equipment_error', None) |
| if equipment_error_num is not None: |
| equipment_error_except = ModemManager.MobileEquipmentError(equipment_error_num) |
| self._dbus_error_name = '{}.Error.MobileEquipment.{}'.format(IFACE_DBUS, equipment_error_except.value_nick) |
| super().__init__(*args, **kwargs) |
| |
| def log(msg): |
| if log_file: |
| try: |
| log_file.write(msg + "\n") |
| log_file.flush() |
| except Exception: |
| pass |
| else: |
| print(msg) |
| |
| def to_path_array(src): |
| array = dbus.Array([], signature=dbus.Signature('o')) |
| for o in src: |
| array.append(to_path(o)) |
| return array |
| |
| def to_path(src): |
| if src: |
| return dbus.ObjectPath(src.path) |
| return dbus.ObjectPath("/") |
| |
| class ExportedObj(dbus.service.Object): |
| |
| DBusInterface = collections.namedtuple('DBusInterface', ['dbus_iface', 'get_props_func', 'set_props_func', 'prop_changed_func']) |
| |
| def __init__(self, bus, object_path): |
| super(ExportedObj, self).__init__(bus, object_path) |
| self._bus = bus |
| self.path = object_path |
| self.__ensure_dbus_ifaces() |
| log("Will add object with path '%s' to object manager" % object_path) |
| object_manager.add_object(self) |
| |
| def __ensure_dbus_ifaces(self): |
| if not hasattr(self, '_ExportedObj__dbus_ifaces'): |
| self.__dbus_ifaces = {} |
| |
| def add_dbus_interface(self, dbus_iface, get_props_func, set_props_func, prop_changed_func): |
| self.__ensure_dbus_ifaces() |
| self.__dbus_ifaces[dbus_iface] = ExportedObj.DBusInterface(dbus_iface, get_props_func, set_props_func, prop_changed_func) |
| |
| def __dbus_interface_get(self, dbus_iface): |
| if dbus_iface not in self.__dbus_ifaces: |
| raise UnknownInterfaceException() |
| return self.__dbus_ifaces[dbus_iface] |
| |
| def _dbus_property_get(self, dbus_iface, propname=None): |
| props = self.__dbus_interface_get(dbus_iface).get_props_func() |
| if propname is None: |
| return props |
| if propname not in props: |
| raise UnknownPropertyException() |
| return props[propname] |
| |
| def _dbus_property_set(self, dbus_iface, propname, value): |
| props = self.__dbus_interface_get(dbus_iface).get_props_func() |
| |
| try: |
| if props[propname] == value: |
| return |
| except KeyError: |
| raise UnknownPropertyException() |
| |
| if self.__dbus_interface_get(dbus_iface).set_props_func is not None: |
| self.__dbus_interface_get(dbus_iface).set_props_func(propname, value) |
| self._dbus_property_notify(dbus_iface, propname) |
| |
| def _dbus_property_notify(self, dbus_iface, propname): |
| prop = self._dbus_property_get(dbus_iface, propname) |
| self.__dbus_interface_get(dbus_iface).prop_changed_func(self, {propname: prop}) |
| ExportedObj.PropertiesChanged(self, dbus_iface, {propname: prop}, []) |
| |
| @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') |
| def PropertiesChanged(self, iface, changed, invalidated): |
| pass |
| |
| @dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') |
| def GetAll(self, dbus_iface): |
| return self._dbus_property_get(dbus_iface) |
| |
| @dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') |
| def Get(self, dbus_iface, name): |
| return self._dbus_property_get(dbus_iface, name) |
| |
| @dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='ssv', out_signature='') |
| def Set(self, dbus_iface, name, value): |
| return self._dbus_property_set(dbus_iface, name, value) |
| |
| def get_managed_ifaces(self): |
| my_ifaces = {} |
| for iface in self.__dbus_ifaces: |
| my_ifaces[iface] = self.__dbus_ifaces[iface].get_props_func() |
| return self.path, my_ifaces |
| |
| |
| ################################################################### |
| IFACE_SIM = 'org.freedesktop.ModemManager1.Sim' |
| |
| PS_IMSI = "Imsi" |
| PS_OPERATOR_IDENTIFIER = "OperatorIdentifier" |
| PS_OPERATOR_NAME = "OperatorName" |
| PS_SIM_IDENTIFIER = "SimIdentifier" |
| |
| class Sim(ExportedObj): |
| def __init__(self, bus, counter, iccid, modem): |
| object_path = "/org/freedesktop/ModemManager1/SIM/%d" % counter |
| self.iccid = iccid |
| self.modem = modem |
| |
| self.add_dbus_interface(IFACE_SIM, self.__get_props, None, Sim.PropertiesChanged) |
| super(Sim, self).__init__(bus, object_path) |
| |
| # Properties interface |
| def __get_props(self): |
| props = {} |
| props[PS_IMSI] = "Imsi_1" |
| props[PS_OPERATOR_IDENTIFIER] = "OperatorIdentifier_1" |
| props[PS_OPERATOR_NAME] = "OperatorName_1" |
| props[PS_SIM_IDENTIFIER] = self.iccid |
| return props |
| |
| # methods |
| @dbus.service.method(dbus_interface=IFACE_SIM, in_signature='ss', out_signature='') |
| def ChangePin(self, old_pin, new_pin): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_SIM, in_signature='sb', out_signature='ao') |
| def EnablePin(self, pin, enabled): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_SIM, in_signature='s', out_signature='') |
| def SendPin(self, pin): |
| if self.modem.equipmentError is not None: |
| raise MobileEquipmentException(equipment_error=self.modem.equipmentError) |
| self.modem.unlock() |
| |
| @dbus.service.method(dbus_interface=IFACE_SIM, in_signature='ss', out_signature='') |
| def SendPuk(self, puk, pin): |
| self.modem.unlock() |
| |
| # signals |
| @dbus.service.signal(IFACE_SIM, signature='a{sv}') |
| def PropertiesChanged(self, changed): |
| pass |
| |
| |
| ################################################################### |
| IFACE_MODEM = 'org.freedesktop.ModemManager1.Modem' |
| |
| PM_SIM = "Sim" |
| PM_BEARERS = "Bearers" |
| PM_SUPPORTED_CAPABILITIES = "SupportedCapabilities" |
| PM_CURRENT_CAPABILITIES = "CurrentCapabilities" |
| PM_MAX_BEARERS = "MaxBearers" |
| PM_MAX_ACTIVE_BEARERS = "MaxActiveBearers" |
| PM_MANUFACTURER = "Manufacturer" |
| PM_MODEL = "Model" |
| PM_REVISION = "Revision" |
| PM_DEVICE_IDENTIFIER = "DeviceIdentifier" |
| PM_DEVICE = "Device" |
| PM_DRIVERS = "Drivers" |
| PM_PLUGIN = "Plugin" |
| PM_PRIMARY_PORT = "PrimaryPort" |
| PM_PORTS = "Ports" |
| PM_EQUIPMENT_IDENTIFIER = "EquipmentIdentifier" |
| PM_UNLOCK_REQUIRED = "UnlockRequired" |
| PM_UNLOCK_RETRIES = "UnlockRetries" |
| PM_STATE = "State" |
| PM_STATE_FAILED_REASON = "StateFailedReason" |
| PM_ACCESS_TECHNOLOGIES = "AccessTechnologies" |
| PM_SIGNAL_QUALITY = "SignalQuality" |
| PM_OWN_NUMBERS = "OwnNumbers" |
| PM_POWER_STATE = "PowerState" |
| PM_SUPPORTED_MODES = "SupportedModes" |
| PM_CURRENT_MODES = "CurrentModes" |
| PM_SUPPORTED_BANDS = "SupportedBands" |
| PM_CURRENT_BANDS = "CurrentBands" |
| PM_SUPPORTED_IP_FAMILIES = "SupportedIpFamilies" |
| |
| class Modem(ExportedObj): |
| counter = 0 |
| |
| def __init__(self, bus, add_sim, iccid): |
| object_path = "/org/freedesktop/ModemManager1/Modem/%d" % Modem.counter |
| self.sim_object = None |
| if add_sim: |
| self.sim_object = Sim(bus, Modem.counter, iccid, self) |
| self.sim_path = to_path(self.sim_object) |
| self.equipmentError = None |
| self.reset_status = True |
| self.reset_status_clear = False |
| |
| self.__props = self.__init_default_props() |
| |
| Modem.counter = Modem.counter + 1 |
| |
| self.add_dbus_interface(IFACE_MODEM, self.__get_props, self.__set_prop, Modem.PropertiesChanged) |
| super(Modem, self).__init__(bus, object_path) |
| |
| # Properties interface |
| def __init_default_props(self): |
| props = {} |
| props[PM_SIM] = dbus.ObjectPath(self.sim_path) |
| props[PM_DEVICE] = dbus.String("/fake/path") |
| props[PM_UNLOCK_REQUIRED] = dbus.UInt32(ModemManager.ModemLock.NONE) |
| props[PM_STATE] = dbus.Int32(ModemManager.ModemState.UNKNOWN) |
| props[PM_STATE_FAILED_REASON] = dbus.UInt32(ModemManager.ModemStateFailedReason.UNKNOWN) |
| # Not already used properties |
| #props[PM_BEARERS] = None |
| #props[PM_SUPPORTED_CAPABILITIES] = None |
| #props[PM_CURRENT_CAPABILITIES] = None |
| #props[PM_MAX_BEARERS] = None |
| #props[PM_MAX_ACTIVE_BEARERS] = None |
| #props[PM_MANUFACTURER] = None |
| #props[PM_MODEL] = None |
| #props[PM_REVISION] = None |
| #props[PM_DEVICE_IDENTIFIER] = None |
| #props[PM_DRIVERS] = None |
| #props[PM_PLUGIN] = None |
| #props[PM_PRIMARY_PORT] = None |
| #props[PM_PORTS] = None |
| #props[PM_EQUIPMENT_IDENTIFIER] = None |
| #props[PM_UNLOCK_RETRIES] = dbus.UInt32(0) |
| #props[PM_ACCESS_TECHNOLOGIES] = None |
| #props[PM_SIGNAL_QUALITY] = None |
| #props[PM_OWN_NUMBERS] = None |
| #props[PM_POWER_STATE] = None |
| #props[PM_SUPPORTED_MODES] = None |
| #props[PM_CURRENT_MODES] = None |
| #props[PM_SUPPORTED_BANDS] = None |
| #props[PM_CURRENT_BANDS] = None |
| #props[PM_SUPPORTED_IP_FAMILIES] = None |
| return props |
| |
| def __get_props(self): |
| return self.__props |
| |
| def __set_prop(self, name, value): |
| try: |
| self.__props[name] = value |
| except KeyError: |
| pass |
| |
| def unlock(self): |
| self._dbus_property_set(IFACE_MODEM, PM_UNLOCK_REQUIRED , dbus.UInt32(ModemManager.ModemLock.NONE)) |
| |
| # methods |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='b', out_signature='') |
| def Enable(self, enable): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='', out_signature='ao') |
| def ListBearers(self): |
| return None |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='a{sv}', out_signature='o') |
| def CreateBearer(self, properties): |
| return None |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='o', out_signature='') |
| def DeleteBearer(self, bearer): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='', out_signature='') |
| def Reset(self): |
| if not self.reset_status: |
| if self.reset_status_clear: |
| self.reset_status = True |
| self.reset_status_clear = False |
| |
| raise Exception("Fake reset exception") |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='s', out_signature='') |
| def FactoryReset(self, code): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='u', out_signature='') |
| def SetPowerState(self, state): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='u', out_signature='') |
| def SetCurrentCapabilities(self, capabilites): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='(uu)', out_signature='') |
| def SetCurrentModes(self, modes): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='au', out_signature='') |
| def SetCurrentBands(self, bands): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_MODEM, in_signature='su', out_signature='s') |
| def Command(self, cmd, timeout): |
| return None |
| |
| # signals |
| @dbus.service.signal(IFACE_MODEM, signature='a{sv}') |
| def PropertiesChanged(self, changed): |
| pass |
| |
| @dbus.service.signal(IFACE_MODEM, signature='iiu') |
| def StateChanged(self, old_state, new_state, reason): |
| pass |
| |
| |
| ################################################################### |
| IFACE_OBJECT_MANAGER = 'org.freedesktop.DBus.ObjectManager' |
| |
| PATH_OBJECT_MANAGER = '/org/freedesktop/ModemManager1' |
| |
| IFACE_TEST = 'org.freedesktop.ModemManager1.LibmmGlibTest' |
| IFACE_MM = 'org.freedesktop.ModemManager1' |
| |
| class ObjectManager(dbus.service.Object): |
| def __init__(self, bus, object_path): |
| super(ObjectManager, self).__init__(bus, object_path) |
| self.objs = [] |
| self.bus = bus |
| self.modem = None |
| |
| @dbus.service.method(dbus_interface=IFACE_OBJECT_MANAGER, |
| in_signature='', out_signature='a{oa{sa{sv}}}', |
| sender_keyword='sender') |
| def GetManagedObjects(self, sender=None): |
| managed_objects = {} |
| for obj in self.objs: |
| name, ifaces = obj.get_managed_ifaces() |
| managed_objects[name] = ifaces |
| return managed_objects |
| |
| def add_object(self, obj): |
| self.objs.append(obj) |
| name, ifaces = obj.get_managed_ifaces() |
| self.InterfacesAdded(name, ifaces) |
| |
| def remove_object(self, obj): |
| self.objs.remove(obj) |
| name, ifaces = obj.get_managed_ifaces() |
| self.InterfacesRemoved(name, ifaces.keys()) |
| |
| @dbus.service.signal(IFACE_OBJECT_MANAGER, signature='oa{sa{sv}}') |
| def InterfacesAdded(self, name, ifaces): |
| pass |
| |
| @dbus.service.signal(IFACE_OBJECT_MANAGER, signature='oas') |
| def InterfacesRemoved(self, name, ifaces): |
| pass |
| |
| # ModemManager methods |
| @dbus.service.method(dbus_interface=IFACE_MM, in_signature='', out_signature='') |
| def ScanDevices(self): |
| pass |
| |
| @dbus.service.method(dbus_interface=IFACE_MM, in_signature='s', out_signature='') |
| def SetLogging(self, logging): |
| pass |
| |
| # Testing methods |
| @dbus.service.method(IFACE_TEST, in_signature='', out_signature='') |
| def Quit(self): |
| mainloop.quit() |
| |
| @dbus.service.method(IFACE_TEST, in_signature='bs', out_signature='o') |
| def AddModem(self, add_sim, iccid): |
| self.modem = Modem(self.bus, add_sim, iccid) |
| return dbus.ObjectPath(self.modem.path) |
| |
| @dbus.service.method(IFACE_TEST, in_signature='iiu', out_signature='') |
| def EmitStateChanged(self, old_state, new_state, reason): |
| if self.modem is not None: |
| self.modem.StateChanged(old_state, new_state, reason) |
| |
| @dbus.service.method(IFACE_TEST, in_signature='ub', out_signature='') |
| def SetMobileEquipmentError(self, error, clear): |
| if self.modem is not None: |
| if clear: |
| self.modem.equipmentError = None |
| else: |
| self.modem.equipmentError = error |
| |
| @dbus.service.method(IFACE_TEST, in_signature='bb', out_signature='') |
| def SetResetStatus(self, status, clear): |
| if self.modem is not None: |
| self.modem.reset_status = status |
| self.modem.reset_status_clear = clear |
| |
| @dbus.service.method(dbus_interface=IFACE_TEST, in_signature='', out_signature='') |
| def Restart(self): |
| bus.release_name("org.freedesktop.ModemManager1") |
| bus.request_name("org.freedesktop.ModemManager1") |
| |
| ################################################################### |
| def stdin_cb(io, condition): |
| mainloop.quit() |
| |
| def quit_cb(user_data): |
| mainloop.quit() |
| |
| def main(): |
| parser = argparse.ArgumentParser(description="ModemManager dbus interface stub utility") |
| parser.add_argument("-f", "--log-file", help="Path of a file to log things into") |
| |
| cfg = parser.parse_args() |
| |
| global log_file |
| |
| if cfg.log_file: |
| try: |
| log_file = open(cfg.log_file, "w") |
| except Exception: |
| log_file = None |
| else: |
| log_file = None |
| |
| log("Starting mainloop") |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) |
| |
| random.seed() |
| |
| global object_manager, bus |
| |
| bus = dbus.SessionBus() |
| log("Creating object manager for /org/freedesktop/ModemManager1") |
| object_manager = ObjectManager(bus, "/org/freedesktop/ModemManager1") |
| |
| log("Requesting name org.freedesktop.ModemManager1") |
| if not bus.request_name("org.freedesktop.ModemManager1"): |
| log("Unable to acquire the DBus name") |
| sys.exit(1) |
| |
| # Watch stdin; if it closes, assume our parent has crashed, and exit |
| id1 = GLib.io_add_watch(0, GLib.IOCondition.HUP, stdin_cb) |
| |
| log("Starting the main loop") |
| try: |
| mainloop.run() |
| except (Exception, KeyboardInterrupt): |
| pass |
| |
| GLib.source_remove(id1) |
| |
| log("Ending the stub") |
| if log_file: |
| log_file.close() |
| sys.exit(0) |
| |
| |
| if __name__ == '__main__': |
| main() |
