How to use the pyhomematic.HMConnection function in pyhomematic

To help you get started, we’ve selected a few pyhomematic examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github danielperna84 / pyhomematic / tests / test_pyhomematic.py View on Github external
def test_1_pyhomematic_init(self):
        LOG.info("TestPyhomematicBase.test_1_pyhomematic_init")
        client = HMConnection(
            interface_id=DEFAULT_INTERFACE_CLIENT,
            autostart=False,
            remotes={
                DEFAULT_REMOTE: {
                    "ip": DEFAULT_IP,
                    "port": self.localport,
                    "connect": True
                }
            }
        )
        client.start()
        time.sleep(STARTUP_DELAY)
        servicemessages = client.getServiceMessages(DEFAULT_REMOTE)
        self.assertEqual(len(servicemessages), 1)
        self.assertEqual(servicemessages[0][0], 'VCU0000001:1')
        self.assertIsInstance(client.devices, dict)
github danielperna84 / pyhomematic / tests / test_pyhomematic.py View on Github external
def test_0_pyhomematic_noinit(self):
        LOG.info("TestPyhomematicBase.test_0_pyhomematic_noinit")
        client = HMConnection(
            interface_id=DEFAULT_INTERFACE_CLIENT,
            autostart=False,
            remotes={
                DEFAULT_REMOTE: {
                    "ip": DEFAULT_IP,
                    "port": self.localport,
                    "connect": False
                }
            }
        )
        client.start()
        time.sleep(STARTUP_DELAY)
        servicemessages = client.getServiceMessages(DEFAULT_REMOTE)
        self.assertEqual(len(servicemessages), 1)
        self.assertEqual(servicemessages[0][0], 'VCU0000001:1')
        client.stop()
github danielperna84 / pyhomematic / tests / test_pyhomematic.py View on Github external
def setUp(self):
        LOG.debug("TestPyhomematicDevices.setUp")
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind(("", 0))
        self.localport = s.getsockname()[1]
        s.close()
        self.vccu = vccu.ServerThread(local=DEFAULT_IP,
                                      localport=self.localport)
        self.vccu.start()
        time.sleep(0.5)
        self.client = HMConnection(
            interface_id=DEFAULT_INTERFACE_CLIENT,
            autostart=False,
            remotes={
                DEFAULT_REMOTE: {
                    "ip": DEFAULT_IP,
                    "port": self.localport,
                    "connect": True
                }
            }
        )
        self.client.start()
        time.sleep(STARTUP_DELAY)
github danielperna84 / pyhomematic / example.py View on Github external
DEVICE1 = 'address_of_rollershutter_device'  # e.g. KEQ7654321
DEVICE2 = 'address_of_doorcontact'  # e.g. LEQ1234567
DEVICE3 = 'address_of_thermostat'

def systemcallback(src, *args):
    print(src)
    for arg in args:
        print(arg)

try:
    # Create a server that listens on 127.0.0.1:7080 and identifies itself as myserver.
    # Connect to Homegear at 127.0.0.1:2001
    # Automatically start everything. Without autostart, pyhomematic.start() can be called.
    # We add a systemcallback so we can see what else happens besides the regular events.
    pyhomematic = HMConnection(interface_id="myserver",
                               autostart=True,
                               systemcallback=systemcallback,
                               remotes={"rf":{
                                   "ip":"127.0.0.1",
                                   "port": 2001}})
except Exception:
    sys.exit(1)

sleepcounter = 0

def eventcallback(address, interface_id, key, value):
    print("CALLBACK: %s, %s, %s, %s" % (address, interface_id, key, value))

while not pyhomematic.devices and sleepcounter < 20:
    print("Waiting for devices")
    sleepcounter += 1
github smarthomeNG / plugins / homematic / __init__.py View on Github external
remotes={self.hm_id:{"ip": self.host, "port": self.port}})
#                                    remotes={self.hm_id:{"ip": self.host, "port": self.port}, self.hmip_id:{"ip": self.host, "port": self.port_hmip}})
        except:
            self.logger.error("{}: Unable to create HomeMatic object".format(self.get_fullname()))
            self._init_complete = False
            return


        # build dict identifier for the homematicIP ccu of this plugin instance
        if self.port_hmip != 0:
            self.hmip_id = 'ip'
            if self.get_instance_name() != '':
                self.hmip_id += '_' + self.get_instance_name()
            # create HomeMaticIP object
            try:
                 self.hmip = HMConnection(interface_id="myserver_ip", autostart=False, 
                                          eventcallback=self.eventcallback, systemcallback=self.systemcallback, 
                                          remotes={self.hmip_id:{"ip": self.host, "port": self.port_hmip}})
            except:
                self.logger.error("{}: Unable to create HomeMaticIP object".format(self.get_fullname()))
#                self._init_complete = False
#                return


        # set the name of the thread that got created by pyhomematic to something meaningfull 
        self.hm._server.name = self.get_fullname()

        # start communication with HomeMatic ccu
        try:
            self.hm.start()
            self.connected = True
        except:
github smarthomeNG / plugins / homematic / __init__.py View on Github external
self.host = self.get_parameter_value('host')
#        self.port = self.get_parameter_value('port')
#        self.port_hmip = self.get_parameter_value('port_hmip')
        self.port = 2001
        self.port_hmip = 2010

        # build dict identifier for the homematic ccu of this plugin instance
        self.hm_id = 'rf'
        if self.get_instance_name() != '':
            self.hm_id += '_' + self.get_instance_name()
            self.log_instance_str = ' ('+self.get_instance_name()+')'
        else:
            self.log_instance_str = ''
        # create HomeMatic object
        try:
             self.hm = HMConnection(interface_id="myserver", autostart=False, 
                                    eventcallback=self.eventcallback, systemcallback=self.systemcallback, 
                                    remotes={self.hm_id:{"ip": self.host, "port": self.port}})
#                                    remotes={self.hm_id:{"ip": self.host, "port": self.port}, self.hmip_id:{"ip": self.host, "port": self.port_hmip}})
        except:
            self.logger.error("{}: Unable to create HomeMatic object".format(self.get_fullname()))
            self._init_complete = False
            return


        # build dict identifier for the homematicIP ccu of this plugin instance
        if self.port_hmip != 0:
            self.hmip_id = 'ip'
            if self.get_instance_name() != '':
                self.hmip_id += '_' + self.get_instance_name()
            # create HomeMaticIP object
            try:
github home-assistant / home-assistant / homeassistant / components / homematic / __init__.py View on Github external
"verify_ssl": rconfig.get(CONF_VERIFY_SSL),
            "connect": True,
        }

    for sname, sconfig in conf[CONF_HOSTS].items():
        remotes[sname] = {
            "ip": sconfig.get(CONF_HOST),
            "port": DEFAULT_PORT,
            "username": sconfig.get(CONF_USERNAME),
            "password": sconfig.get(CONF_PASSWORD),
            "connect": False,
        }

    # Create server thread
    bound_system_callback = partial(_system_callback_handler, hass, config)
    hass.data[DATA_HOMEMATIC] = homematic = HMConnection(
        local=config[DOMAIN].get(CONF_LOCAL_IP),
        localport=config[DOMAIN].get(CONF_LOCAL_PORT, DEFAULT_LOCAL_PORT),
        remotes=remotes,
        systemcallback=bound_system_callback,
        interface_id="homeassistant",
    )

    # Start server thread, connect to hosts, initialize to receive events
    homematic.start()

    # Stops server when HASS is shutting down
    hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, hass.data[DATA_HOMEMATIC].stop)

    # Init homematic hubs
    entity_hubs = []
    for hub_name in conf[CONF_HOSTS].keys():