Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2024-12-22 19:23:34 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Update get_ip function

Browse Source
This commit is contained in:
Steven B 2024-12-19 13:44:23 +00:00
parent 6b411700b7
commit d201e59096
No known key found for this signature in database
GPG Key ID: 6D5B46B3679F2A43
1 changed files with 21 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
kasa/smartcam/modules/listen.py
View File

@ -6,11 +6,9 @@ import asyncio
import logging
import os
import socket
import sys
import uuid
from collections.abc import Callable, Iterable
from datetime import timedelta
from subprocess import check_output
import onvif # type: ignore[import-untyped]
from aiohttp import web
@ -18,6 +16,7 @@ from onvif.managers import NotificationManager # type: ignore[import-untyped]
from ...credentials import Credentials
from ...eventtype import EventType
from ...exceptions import KasaException
from ..smartcammodule import SmartCamModule
_LOGGER = logging.getLogger(__name__)
@ -111,16 +110,20 @@ class Listen(SmartCamModule):
await self.runner.shutdown()
async def _get_host_ip(self) -> str:
def _get_host() -> str:
if sys.platform == "win32":
return socket.gethostbyname(socket.gethostname())
else:
res = check_output(["hostname", "-I"]) # noqa: S603, S607
listen_ip, _, _ = res.decode().partition(" ")
return listen_ip
def get_ip() -> str:
# From https://stackoverflow.com/a/28950776
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0)
try:
# doesn't even have to be reachable
s.connect(("10.254.254.254", 1))
ip = s.getsockname()[0]
finally:
s.close()
return ip
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, _get_host)
return await loop.run_in_executor(None, get_ip)
async def _start_server(self, listen_ip: str | None, listen_port: int) -> str:
app = web.Application()
@ -132,7 +135,14 @@ class Listen(SmartCamModule):
await self.runner.setup()
if not listen_ip:
listen_ip = await self._get_host_ip()
try:
listen_ip = await self._get_host_ip()
except Exception as ex:
raise KasaException(
"Unable to determine listen ip starting "
f"listener for {self._device.host}",
ex,
) from ex
self.site = web.TCPSite(self.runner, listen_ip, listen_port)
try: