Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
+ cmd
)
logger.info(f"ZwavePlugin: Getting {url} ")
try:
resp = requests.get(url, auth=(self.zwave_user, self.zwave_pass))
except Exception as e:
logger.error(f"ZwavePlugin: {e}")
return False
if resp.status_code == 200:
if resp.text == response_ok:
self.state = cmd
return True
logger.error(f"ZwavePlugin: {resp.status_code} {resp.text} ")
return False
url = (
"http://"
+ self.zwave_host
+ ":"
+ str(self.zwave_port)
+ "/JS/Run/controller.devices.get('"
+ self.zwave_device
+ "').get('metrics:level')"
)
logger.info(f"ZwavePlugin: Getting {url} ")
try:
resp = requests.get(url, auth=(self.zwave_user, self.zwave_pass))
except Exception as e:
logger.error(f"ZwavePlugin: {e}")
return "unknown"
if resp.status_code == 200:
if resp.text == '"off"' or resp.text == '"on"':
return resp.text.strip('"')
logger.error(f"ZwavePlugin: {resp.status_code} {resp.text} ")
return "unknown"
k: v
for k, v in config["PLUGINS"][plugin].items()
if k not in {"DEVICES", "path"}
}
logger.debug(f"plugin_vars: {repr(plugin_vars)}")
for device in config["PLUGINS"][plugin]["DEVICES"]:
logger.debug(f"device config: {repr(device)}")
# Ensure port is `int`, set it if not given (`None`) or 0
device["port"] = int(device.get("port", 0)) or get_unused_port()
try:
plugin = PluginClass(**plugin_vars, **device)
except TypeError:
logger.error(f"Error in plugin {repr(PluginClass)}")
raise
fauxmo = partial(Fauxmo, name=plugin.name, plugin=plugin)
coro = loop.create_server(fauxmo, host=fauxmo_ip, port=plugin.port)
server = loop.run_until_complete(coro)
server.fauxmoplugin = plugin # type: ignore
servers.append(server)
ssdp_server.add_device(plugin.name, fauxmo_ip, plugin.port)
logger.debug(f"Started fauxmo device: {repr(fauxmo.keywords)}")
logger.info("Starting UDP server")
listen = loop.create_datagram_endpoint(
lambda: ssdp_server, sock=make_udp_sock()
url = (
"http://"
+ self.zwave_host
+ ":"
+ str(self.zwave_port)
+ "/ZAutomation/api/v1/devices/"
+ self.zwave_device
+ "/command/"
+ cmd
)
logger.info(f"ZwavePlugin: Getting {url} ")
try:
resp = requests.get(url, auth=(self.zwave_user, self.zwave_pass))
except Exception as e:
logger.error(f"ZwavePlugin: {e}")
return False
if resp.status_code == 200:
if resp.text == response_ok:
self.state = cmd
return True
logger.error(f"ZwavePlugin: {resp.status_code} {resp.text} ")
return False
+ self.zwave_device
+ "').get('metrics:level')"
)
logger.info(f"ZwavePlugin: Getting {url} ")
try:
resp = requests.get(url, auth=(self.zwave_user, self.zwave_pass))
except Exception as e:
logger.error(f"ZwavePlugin: {e}")
return "unknown"
if resp.status_code == 200:
if resp.text == '"off"' or resp.text == '"on"':
return resp.text.strip('"')
logger.error(f"ZwavePlugin: {resp.status_code} {resp.text} ")
return "unknown"
logger.info(f"Fauxmo {__version__}")
logger.debug(sys.version)
if config_path_str:
config_path = pathlib.Path(config_path_str)
else:
for config_dir in (".", "~/.fauxmo", "/etc/fauxmo"):
config_path = pathlib.Path(config_dir).expanduser() / "config.json"
if config_path.is_file():
logger.info(f"Using config: {config_path}")
break
try:
config = json.loads(config_path.read_text())
except FileNotFoundError:
logger.error(
"Could not find config file in default search path. Try "
"specifying your file with `-c`.\n"
)
raise
# Every config should include a FAUXMO section
fauxmo_config = config.get("FAUXMO")
fauxmo_ip = get_local_ip(fauxmo_config.get("ip_address"))
ssdp_server = SSDPServer()
servers = []
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
if verbosity < 20: