Source code for gearthonic.methods

"""Includes all functions provided by the XML RPC API, split into logical entities."""


class BaseCollection(object):
    """Provide the interface to call XML RPC methods."""

    def __init__(self, caller):
        """
        :param caller: interface to call XML RPC methods
        """
        self._caller = caller

    def call(self, method_name, *args, **kwargs):
        """Make a call to the XML RPC API.

        :param method_name: name of the method to call
        :type method_name: str
        :param args: arguments to pass to the method
        :param kwargs: keyword arguments to pass to the method
        """
        return self._caller.call(method_name, *args, **kwargs)


[docs]class SystemMethodsCollection(BaseCollection): """ All XML RPC server provide a set of standard methods. """
[docs] def get_capabilities(self): """Lists server's XML RPC capabilities. Example output:: { 'xmlrpc': {'specUrl': 'http://example.com', 'specVersion': 1} 'faults_interop': {'specUrl': 'http://example2.com', 'specVersion': 101} 'introspection': {'specUrl': 'http://example3.com', 'specVersion': 42} } :return: A dict containing all information :rtype: dict """ return self.call('system.getCapabilities')
[docs] def list_methods(self): """Lists servers's XML RPC methods. :return: A list of available methods :rtype: list """ return self.call('system.listMethods')
[docs] def method_help(self, method_name): """Returns the description of a method. :param method_name: The name of the method :type method_name: str :return: The description of the method :rtype: str """ return self.call('system.methodHelp', method_name)
[docs] def method_signature(self, method_name): """Returns the signature of a method. :param method_name: The name of the method :type method_name: str :return: The signature of the method :rtype: list """ return self.call('system.methodSignature', method_name)
[docs] def multicall(self, methods): """Call multiple methods at once. Example list of ``methods``:: [ {'methodName': 'getValue', 'params': [13, 4, 'TEMPERATURE']}, {'methodName': 'getValue', 'params': [3, 3, 'HUMIDITY']} ] Return value of the multicall:: [22.0, 58] :param methods: A list of methods and their parameters :type methods: list :return: A list of method responses. :rtype: list """ return self.call('system.multicall', methods)
[docs]class GeneralMethodsCollection(BaseCollection): """All general methods."""
[docs] def write_log(self, message, log_level=None): """Write a message to the Homegear log. This method writes a message to Homegear's log. It's possible to set the log level. Valid values are: * 1: Critical * 2: Error * 3: Warning * 4: Info * 5: Debug :param message: This is the message you want to write to the log file. The date is automatically prepended. :type message: str :param log_level: (optional) This is the log level of the message. If Homegear's log level value is lower than this value, no message is logged. :type log_level: int """ return self.call('writeLog', message, log_level=log_level)
[docs] def get_service_messages(self, return_id): """Return all service messages. This method returns all service messages that are currently active in Homegear (device unreachable, config pending, low battery, sabotage, ...). :param return_id: Recommended. If true, Homegear returns the peer ID instead of the "address" (serial number and channel separated by a colon). By default, the address is returned for compatibility reasons. :type return_id: bool """ return self.call('getServiceMessages', return_id)
[docs] def log_level(self, level=None): """Get or set the current log level. Valid values are: * 0: Log nothing * 1: Critical * 2: Error * 3: Warning * 4: Info * 5: Debug .. warning:: Don't use "debug" for normal operation as it slows down Homegear. :param level: (optional) This is the log level that you want to set. :type level: int """ return self.call('logLevel', level=level)
[docs] def get_version(self): """Return Homegear's version number. """ return self.call('getVersion')
[docs]class DeviceMethodsCollection(BaseCollection): """All device related methods."""
[docs] def list_devices(self): """Return a list of devices. :return: List of devices :rtype: list """ return self.call('listDevices')
[docs] def get_value(self, peer_id, channel, key, request_from_device=False, asynchronous=False): """Return the value of the device, specified by channel and key (parameterName). Per default the value is read from the local cache of Homegear. If the value should be read from the device, use ``request_from_device``. If the value should be read from the device, this can be done asynchronously. The method returns immediately and doesn't wait for the current value. The value will be sent as an event as soon as it's returned by the device. Error codes: * Returns ``-2`` when the device or channel is unknown * Returns ``-5`` when the key (parameter) is unknown :param peer_id: ID of the device :type peer_id: int :param channel: Channel of the device to get the value for :type channel: int :param key: Name of the parameter to get the value for :type key: str :param request_from_device: If true value is read from the device :type request_from_device: bool :param asynchronous: If true value is read asynchronously :type asynchronous: bool :return: The value of the parameter or error code :rtype: unknown """ return self.call('getValue', peer_id, channel, key, request_from_device, asynchronous)
[docs] def set_value(self, peer_id, channel, key, value): """Set the value of the device, specified by channel and key (parameterName). :param peer_id: ID of the device :type peer_id: int :param channel: Channel of the device to set the value for :type channel: int :param key: Name of the parameter to get the value for :type key: str :param value: The value to set :type value: unknown :return: * ``None`` on success * ``-2`` when the device or channel is unknown * ``-5`` when the key (parameter) is unknown * ``-100`` when the device did not respond * ``-101`` when the device returns an error """ return self.call('setValue', peer_id, channel, key, value)
[docs] def get_all_config(self, peer_id=None): """Return all peer configuration parameters and some additional metadata. This method returns all configuration parameter values and information about all configuration parameters for one or all peers. Variables are not returned. To get all variables, call :func:`~gearthonic.methods.DeviceMethodsCollection.get_all_values`. :param peer_id: (optional) When specified, only variables of this peer are returned. :type peer_id: int """ return self.call('getAllConfig', peer_id=peer_id)
[docs] def get_all_values(self, peer_id=None, return_write_only_variables=False): """Return all peer configuration parameters and some additional metadata. This method returns all variable values and information about all variables for one or all peers. Configuration parameters are not returned. To get all configuration parameters, call :func:`~gearthonic.methods.DeviceMethodsCollection.get_all_config`. :param peer_id: (optional) When specified, only variables of this peer are returned. :type peer_id: int :param return_write_only_variables: (optional) When specifed, write only variables are also returned. :type return_write_only_variables: bool """ return self.call('getAllValues', peer_id=peer_id, return_write_only_variables=return_write_only_variables)
[docs]class PairingMethodsCollection(BaseCollection): """All pairing related methods."""
[docs] def add_device(self, serial_number, family_id=None): """Pair a device without enabling pairing mode. This method pairs a device by its serial number, but this does not work for all devices. :param serial_number: The serial number of the device to be paired :type serial_number: str :param family_id: (optional) ID of the family you want to add the device to; if not specified, "addDevice" is executed for all device families that support it. :type family_id: int """ return self.call('addDevice', serial_number, family_id=family_id)
[docs] def search_devices(self, family_id=None): """Start a device search for all supported device families. When you use this method, Homegear searches for new devices in all device families that support the method. :param family_id: (optional) This is the ID of the family that you want to search for devices. :type family_id: int """ return self.call('searchDevices', family_id=family_id)
[docs] def set_install_mode(self, on, family_id=None, duration=60): """Enable pairing mode. This method enables or disables pairing mode for all device families if it is supported by the device family. :param on: When this is true, pairing mode is enabled. Otherwise, pairing mode is disabled. :type on: bool :param family_id: (optional) This is the ID of the family for which you want to enable pairing mode. If it is not specified, pairing mode will be enabled for all device families. :type family_id: int :param duration: (optional) This is the duration in seconds that the central should remain in pairing mode. The minimum duration is 5 seconds, and the maximum duration is 3600 seconds. The default duration is 60 seconds. :type duration: int """ return self.call('setInstallMode', on=on, family_id=family_id, duration=duration)
[docs] def get_install_mode(self, family_id=None): """Return the time left in pairing mode. This method returns the remaining amount of time the central will be in pairing mode. :param family_id: (optional) This is the ID of the family for which you want to get the remaining time in pairing mode. If not specified, the remaining time in pairing mode of the first central for which pairing mode enabled is returned. :type family_id: int """ return self.call('getInstallMode', family_id=family_id)
[docs] def create_device(self, family_id, device_type, serial_number, address, firmware_version): """Create a device manually. This method manually creates a new device. It is not supported by all device families, and it is also not supported for all devices. createDevice can be used to create virtual devices in the family "Miscellaneous". :param family_id: This is the ID of the family you want to create the device in. See: :func:`~gearthonic.methods.FamilyMethodsCollection.list_families`. :type family_id: int :param device_type: The type ID of the device as specified in the device's XML file :type device_type: int :param serial_number: The serial number of the new device :type serial_number: str :param address: This is the physical address of the new device. Depending on the device family, this parameter might be optional. If it is not needed, set it to "-1". :type address: int :param firmware_version: This is the firmware version of the new device. Depending on the device family, this parameter might be optional. If the firmware version is "1.2", set this variable to 0x12 = 18. If it is not needed, set it to "-1". :type firmware_version: int """ return self.call('createDevice', family_id, device_type, serial_number, address, firmware_version)
[docs] def get_pairing_methods(self, family_id): """Return the pairing methods supported by a device family. This method returns all pairing methods supported by the specified device family. :param family_id: The ID of the family for which you want to get the supported pairing methods :type family_id: int """ return self.call('getPairingMethods', family_id)
[docs]class FamilyMethodsCollection(BaseCollection): """All pairing related methods."""
[docs] def list_families(self): """Return information about all device families (ID, name, pairing methods). This method returns information about all available device families. Use this method to get the ID of a family if you have only the name or only the ID. You can also use this method to get the pairing methods supported by the family. """ return self.call('listFamilies')
[docs]class EventServerMethodsCollection(BaseCollection): """All EventServer related methods"""
[docs] def unsubscribe_peers(self, event_server_id): """Unsubscribe peer events. This method is used to unsubscribe peer events after :func:`~gearthonic.methods.EventServerMethodsCollection.subscribe_peers` has been called. :param event_server_id: This is either the url specified in :func:`~gearthonic.methods.EventServerMethodsCollection.init` or the WebSocket client ID. :type event_server_id: str """ return self.call('unsubscribePeers', event_server_id)
[docs] def client_server_initialized(self, interface_id): """Check if an RPC client's RPC server was successfully registered and if it still is registered. This method checks if an RPC client's RPC server is registered and connected to Homegear. You can register your RPC "event" server by calling :func:`~gearthonic.methods.EventServerMethodsCollection.init`. :param interface_id: The interface ID as specified in :func:`~gearthonic.methods.EventServerMethodsCollection.init` :type interface_id: str """ return self.call('clientServerInitialized', interface_id)
[docs] def subscribe_peers(self, event_server_id): """Subscribe peer events that are to be sent to an event server. This method is used to subscribe peer events after calling :func:`~gearthonic.methods.EventServerMethodsCollection.init` with the `subscribe_peers` flag set. :param event_server_id: This is either the url specified in :func:`~gearthonic.methods.EventServerMethodsCollection.init` or the WebSocket client ID. :type event_server_id: str """ return self.call('subscribePeers', event_server_id)
[docs] def trigger_rpc_event(self, event_method): """Send an RPC event to all RPC event servers. This method manually calls an RPC event method on all RPC event servers. Currently supported methods are `deleteDevices`, `newDevices` and `updateDevice`. :param event_method: This is the method you want to call. :type event_method: str """ return self.call('triggerRpcEvent', event_method)
[docs] def list_client_servers(self, interface_id=None): """Return information about all RPC servers registered with Homegear by clients. This method returns an array with one entry for each RPC server registered with Homegear. :param interface_id: (optional) This is the interface ID of the RPC server as it was passed to :func:`~gearthonic.methods.EventServerMethodsCollection.init`. If it is specified, only the information for this server is returned. :type interface_id: str """ return self.call('listClientServers', interface_id=interface_id)
[docs] def init(self, url, interface_id, flags=None): """Register a client's RPC server with Homegear to receive events. This method is used to register or unregister an RPC event server with Homegear. After calling this method, Homegear's RPC client starts sending events and device updates to the registered server. It is not necessary to call "init" for MQTT or WebSockets. It's possible to configure the communication between Homegear and the client's RPC server by using `flags`. The following (bitmask) flags are available: * 0x01: keepAlive: Do not close the connection after each packet. * 0x02: binaryMode: Send RPC data in binary format. Equivalent to "binary://" or "binarys://". * 0x04: newFormat: (Recommended) Send device's ID in broadcast methods instead of the serial number and activates variable types ARRAY and STRUCT. This is recommended because serial numbers are not necessarily unique. * 0x08: subscribePeers: If this is set, Homegear will send events only for peers subscribed with `subscribePeers` to the event server. * 0x10: jsonMode: Send RPC data in JSON format. So if you want to enable `binaryMode` and `subscribePeers`, you have to provide `10`. If you want to set `newFormat` additionally, provide `14`. And if you want to enable `jsonMode` additionally, provide `30`. :param url: The URL of the event server that you want to register, including "http://" and the port. If you use "binary://", RPC data is sent in binary format. If you pass "https://" or "binarys://", SSL is enabled. :type url: str :param interface_id: This is an arbitrary name for the interface. To unregister an event server, pass an empty string to interfaceId. :type interface_id: str :param flags: (optional) Used to configure the communication between Homegear and the registered server. :type flags: int """ return self.call('init', url, interface_id, flags=flags)
[docs]class PhysicalInterfaceMethodsCollection(BaseCollection): """All methods related to the physical interface."""
[docs] def list_bidcos_interfaces(self): """Exist only for compatibility reasons. This method exists only for reasons of backward compatibility with the CCU and has no real function. """ return self.call('listBidcosInterfaces')
[docs] def set_interface(self, peer_id, interface_id): """Set the physical interface Homegear uses to communicate with a peer. This method sets the physical interface that Homegear is to use to communicate with a peer. :param peer_id: The ID of the peer you want to set the interface for :type peer_id: int :param interface_id: This is the ID of the physical interface as defined in the family interface settings. If it is empty, the physical interface is reset to the default interface. :type interface_id: str """ return self.call('setInterface', peer_id, interface_id)
[docs] def list_interfaces(self, family_id=None): """List all physical interfaces with status information. This method returns a list of all physical interfaces. It can be used to determine if an interface is available. :param family_id: (optional) The ID of the family for which you want to get interfaces :type family_id: int """ return self.call('listInterfaces', family_id=family_id)
[docs]class MetadataMethodsCollection(BaseCollection): """All metadata related methods."""
[docs] def set_metadata(self, peer_id, data_id, value): """Store metadata for a peer. This method can be used to store metadata for devices in Homegear's database. You can retrieve this metadata later by calling :func:`~gearthonic.methods.MetadataMethodsCollection.get_metadata`. :param peer_id: The ID of the peer for which you want to store metadata :type peer_id: int :param data_id: A name of your choice :type data_id: str :param value: The value you want to store """ return self.call('setMetadata', peer_id, data_id, value)
[docs] def get_all_metadata(self, peer_id): """Return all the metadata of one peer. :param peer_id: The ID of the peer for which you want to get metadata :type peer_id: int """ return self.call('getAllMetadata', peer_id)
[docs] def get_metadata(self, peer_id, data_id): """Retrieve previously stored metadata. This method returns metadata that was previously stored with :func:`~gearthonic.methods.MetadataMethodsCollection.setMetadata`. :param peer_id: The ID of the peer for which you want to get the metadata :type peer_id: int :param data_id: The data ID :type data_id: str """ return self.call('getMetadata', peer_id, data_id)
[docs] def delete_metadata(self, peer_id, data_id=None): """Delete previously stored metadata. :param peer_id: The ID of the peer for which the metadata is stored. :type peer_id: int :param data_id: (optional) The dataId :type data_id: str """ return self.call('deleteMetadata', peer_id, data_id=data_id)
[docs]class SystemVariableMethodsCollection(BaseCollection): """All system variables related methods."""
[docs] def delete_system_variable(self, name): """Delete a system variable. This method deletes a system variable created with :func:`~gearthonic.methods.SystemVariableMethodsCollection.set_system_variable`. :param name: The name of the system variable to be deleted :type name: str """ return self.call('deleteSystemVariable', name)
[docs] def set_system_variable(self, name, value): """Create or update a system variable. This method can be used to store arbitrary data in Homegear's database. You can retrieve this data later by calling :func:`~gearthonic.methods.SystemVariableMethodsCollection.getSystemVariable`. :param name: A name of your choice :type name: str :param value: The value to be stored """ return self.call('setSystemVariable', name, value)
[docs] def get_system_variable(self, name): """Get the value of a system variable. This method returns a system variable's value that was previously stored with :func:`~gearthonic.methods.SystemVariableMethodsCollection.set_system_variable`. :param name: The name of the system variable :type name: str """ return self.call('getSystemVariable', name)
[docs] def get_all_system_variables(self): """Return all system variables.""" return self.call('getAllSystemVariables')