Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)
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()
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)
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
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:
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:
"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():