LBRYCommunity/aioupnp
1
0
Fork 0
Code Issues 7 Pull requests 1 Packages Projects Releases 1 Wiki Activity

detect working m search keyword args by default and include them in the result

Browse source
This commit is contained in:
Jack Robison 2018-10-12 13:22:08 -04:00
parent ad63f3e853
commit d37992e292
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
4 changed files with 36 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
aioupnp/__main__.py
View file

@ -47,11 +47,6 @@ def main():
'gateway_address': '',
'lan_address': '',
'timeout': 30,
'HOST': SSDP_HOST,
'ST': UPNP_ORG_IGD,
'MAN': SSDP_DISCOVER,
'MX': 1,
}
options = OrderedDict()

16
aioupnp/gateway.py
View file

@ -6,7 +6,7 @@ from aioupnp.util import get_dict_val_case_insensitive, BASE_PORT_REGEX, BASE_AD
from aioupnp.constants import SPEC_VERSION, SERVICE
from aioupnp.commands import SOAPCommands
from aioupnp.device import Device, Service
from aioupnp.protocols.ssdp import fuzzy_m_search
from aioupnp.protocols.ssdp import fuzzy_m_search, m_search
from aioupnp.protocols.scpd import scpd_get
from aioupnp.protocols.soap import SOAPCommand
from aioupnp.serialization.ssdp import SSDPDatagram
@ -100,9 +100,10 @@ class Gateway:
@property
def manufacturer_string(self) -> str:
if not self._device:
if not self.devices:
raise NotImplementedError()
return "%s %s" % (self._device.manufacturer, self._device.modelName)
device = list(self.devices.values())[0]
return "%s %s" % (device.manufacturer, device.modelName)
@property
def services(self) -> Dict:
@ -143,8 +144,13 @@ class Gateway:
@classmethod
async def discover_gateway(cls, lan_address: str, gateway_address: str, timeout: int = 30,
ssdp_socket: socket.socket = None, soap_socket: socket.socket = None):
m_search_args, datagram = await fuzzy_m_search(lan_address, gateway_address, timeout, ssdp_socket)
igd_args: OrderedDict = None, ssdp_socket: socket.socket = None,
soap_socket: socket.socket = None):
if not igd_args:
m_search_args, datagram = await fuzzy_m_search(lan_address, gateway_address, timeout, ssdp_socket)
else:
m_search_args = OrderedDict(igd_args)
datagram = await m_search(lan_address, gateway_address, igd_args, timeout, ssdp_socket)
gateway = cls(datagram, m_search_args, lan_address, gateway_address)
await gateway.discover_commands(soap_socket)
return gateway

9
aioupnp/serialization/ssdp.py
View file

@ -101,10 +101,15 @@ class SSDPDatagram(object):
setattr(self, normalized, v)
self._case_mappings: dict = {k.lower(): k for k in kwargs.keys()}
def get_cli_igd_kwargs(self) -> str:
fields = []
for field in self._field_order:
v = getattr(self, field)
fields.append("--%s=%s" % (self._case_mappings.get(field, field), v))
return " ".join(fields)
def __repr__(self) -> str:
return self.as_json()
# return ("SSDPDatagram(packet_type=%s, " % self._packet_type) + \
# ", ".join("%s=%s" % (n, v) for n, v in self.as_dict().items()) + ")"
def __getitem__(self, item):
for i in self._required_fields[self._packet_type]:

26
aioupnp/upnp.py
View file

@ -8,8 +8,9 @@ from typing import Tuple, Dict, List, Union
from aioupnp.fault import UPnPError
from aioupnp.gateway import Gateway
from aioupnp.util import get_gateway_and_lan_addresses
from aioupnp.protocols.ssdp import m_search
from aioupnp.protocols.ssdp import m_search, fuzzy_m_search
from aioupnp.protocols.soap import SOAPCommand
from aioupnp.serialization.ssdp import SSDPDatagram
log = logging.getLogger(__name__)
@ -44,27 +45,35 @@ class UPnP:
@classmethod
async def discover(cls, lan_address: str = '', gateway_address: str = '', timeout: int = 30,
interface_name: str = 'default',
igd_args: OrderedDict = None, interface_name: str = 'default',
ssdp_socket: socket.socket = None, soap_socket: socket.socket = None):
try:
lan_address, gateway_address = cls.get_lan_and_gateway(lan_address, gateway_address, interface_name)
except Exception as err:
raise UPnPError("failed to get lan and gateway addresses: %s" % str(err))
gateway = await Gateway.discover_gateway(
lan_address, gateway_address, timeout, ssdp_socket, soap_socket
lan_address, gateway_address, timeout, igd_args, ssdp_socket, soap_socket
)
return cls(lan_address, gateway_address, gateway)
@classmethod
@cli
async def m_search(cls, lan_address: str = '', gateway_address: str = '', timeout: int = 1,
args: OrderedDict = None, interface_name: str = 'default') -> Dict:
args = args or OrderedDict()
lan_address, gateway_address = cls.get_lan_and_gateway(lan_address, gateway_address, interface_name)
datagram = await m_search(lan_address, gateway_address, args, timeout)
igd_args: OrderedDict = None, interface_name: str = 'default',
ssdp_socket: socket.socket = None) -> Dict:
try:
lan_address, gateway_address = cls.get_lan_and_gateway(lan_address, gateway_address, interface_name)
except Exception as err:
raise UPnPError("failed to get lan and gateway addresses: %s" % str(err))
if not igd_args:
igd_args, datagram = await fuzzy_m_search(lan_address, gateway_address, timeout, ssdp_socket)
else:
igd_args = OrderedDict(igd_args)
datagram = await m_search(lan_address, gateway_address, igd_args, timeout, ssdp_socket)
return {
'lan_address': lan_address,
'gateway_address': gateway_address,
'm_search_kwargs': SSDPDatagram("M-SEARCH", igd_args).get_cli_igd_kwargs(),
'discover_reply': datagram.as_dict()
}
@ -234,7 +243,7 @@ class UPnP:
else:
try:
u = await cls.discover(
lan_address, gateway_address, timeout, interface_name
lan_address, gateway_address, timeout, igd_args, interface_name
)
except UPnPError as err:
fut.set_exception(err)
@ -249,6 +258,7 @@ class UPnP:
fut.set_result(result)
except UPnPError as err:
fut.set_exception(err)
except Exception as err:
log.exception("uncaught error")
fut.set_exception(UPnPError("uncaught error: %s" % str(err)))