Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_local_ip(ip_address: str = None) -> str:
"""Attempt to get the local network-connected IP address.
Args:
ip_address: Either desired ip address or string or "auto"
Returns:
Current IP address as string
"""
if ip_address is None or ip_address.lower() == "auto":
logger.debug("Attempting to get IP address automatically")
hostname = socket.gethostname()
try:
ip_address = socket.gethostbyname(hostname)
except socket.gaierror:
ip_address = "unknown"
# Workaround for Linux returning localhost
# See: SO question #166506 by @UnkwnTech
if ip_address in ["127.0.1.1", "127.0.0.1", "localhost", "unknown"]:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.connect(("8.8.8.8", 80))
ip_address = sock.getsockname()[0]
logger.debug(f"Using IP address: {ip_address}")
return ip_address
def get_state(self) -> str:
"""Get device state.
Returns:
"on", "off", or "unknown"
If fake_state is set to true, it does not exec a query for status,
it returns the previous status stored.
"""
if self.fake_state:
logger.info(f"ZwavePlugin: return fake {self.state} ")
return self.state
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:
def data_received(self, data: bytes) -> None:
"""Decode incoming data.
Args:
data: Incoming message, either setup request or action request
"""
msg = data.decode()
logger.debug(f"Received message:\n{msg}")
if msg.startswith("GET /setup.xml HTTP/1.1"):
logger.info("setup.xml requested by Echo")
self.handle_setup()
elif "/eventservice.xml" in msg:
logger.info("eventservice.xml request by Echo")
self.handle_event()
elif "/metainfoservice.xml" in msg:
logger.info("metainfoservice.xml request by Echo")
self.handle_metainfo()
elif msg.startswith("POST /upnp/control/basicevent1 HTTP/1.1"):
logger.info("request BasicEvent1")
self.handle_action(msg)
loop.add_signal_handler(getattr(signal, signame), loop.stop)
# Workaround for Windows (https://github.com/n8henrie/fauxmo/issues/21)
except NotImplementedError:
if sys.platform == "win32":
pass
else:
raise
loop.run_forever()
# Will not reach this part unless SIGINT or SIGTERM triggers `loop.stop()`
logger.debug("Shutdown starting...")
transport.close()
for idx, server in enumerate(servers):
logger.debug(f"Shutting down server {idx}...")
server.fauxmoplugin.close() # type: ignore
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()
async def _send_async_response(
self, response: bytes, addr: Tuple[str, int], mx: float = 0.0
) -> None:
logger.debug(f"Sending response to {addr} with mx {mx}:\n{response!r}")
await asyncio.sleep(random.random() * max(0, min(5, mx)))
self.transport.sendto(response, addr)
logger.debug("Attempting to get IP address automatically")
hostname = socket.gethostname()
try:
ip_address = socket.gethostbyname(hostname)
except socket.gaierror:
ip_address = "unknown"
# Workaround for Linux returning localhost
# See: SO question #166506 by @UnkwnTech
if ip_address in ["127.0.1.1", "127.0.0.1", "localhost", "unknown"]:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.connect(("8.8.8.8", 80))
ip_address = sock.getsockname()[0]
logger.debug(f"Using IP address: {ip_address}")
return ip_address