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

add mx argument

Browse source
This commit is contained in:
Jack Robison 2018-10-10 15:18:11 -04:00
parent 28c01b5b31
commit b0d1f7a193
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
5 changed files with 22 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
aioupnp/__main__.py
View file

@ -47,6 +47,7 @@ def main():
'timeout': 1,
'service': '', # if not provided try all of them
'man': '',
'mx': 1,
'return_as_json': True
}
@ -80,8 +81,8 @@ def main():
UPnP.run_cli(
command.replace('-', '_'), options.pop('lan_address'), options.pop('gateway_address'),
options.pop('timeout'), options.pop('service'), options.pop('man'), options.pop('interface'),
kwargs
options.pop('timeout'), options.pop('service'), options.pop('man'), options.pop('mx'),
options.pop('interface'), kwargs
)

2
aioupnp/commands.py
View file

@ -3,7 +3,7 @@ from typing import Tuple, Union
none_or_str = Union[None, str]
class SCPDCommands:
class SOAPCommands:
def debug_commands(self) -> dict:
raise NotImplementedError()

8
aioupnp/gateway.py
View file

@ -3,7 +3,7 @@ import socket
from typing import Dict, List, Union, Type
from aioupnp.util import get_dict_val_case_insensitive, BASE_PORT_REGEX, BASE_ADDRESS_REGEX
from aioupnp.constants import SPEC_VERSION, UPNP_ORG_IGD, SERVICE
from aioupnp.commands import SCPDCommands
from aioupnp.commands import SOAPCommands
from aioupnp.device import Device, Service
from aioupnp.protocols.ssdp import m_search
from aioupnp.protocols.scpd import scpd_get
@ -90,7 +90,7 @@ class Gateway:
self._unsupported_actions = {}
self._registered_commands = {}
self.commands = SCPDCommands()
self.commands = SOAPCommands()
def gateway_descriptor(self) -> dict:
r = {
@ -129,9 +129,9 @@ class Gateway:
@classmethod
async def discover_gateway(cls, lan_address: str, gateway_address: str, timeout: int = 1,
service: str = UPNP_ORG_IGD, man: str = '',
service: str = UPNP_ORG_IGD, man: str = '', mx: int = 1,
ssdp_socket: socket.socket = None, soap_socket: socket.socket = None):
datagram = await m_search(lan_address, gateway_address, timeout, service, man, ssdp_socket)
datagram = await m_search(lan_address, gateway_address, timeout, service, man, mx, ssdp_socket)
gateway = cls(**datagram.as_dict())
await gateway.discover_commands(soap_socket)
return gateway

14
aioupnp/protocols/ssdp.py
View file

@ -25,15 +25,15 @@ class SSDPProtocol(MulticastProtocol):
self.notifications: List = []
self.replies: List = []
def send_m_search_packet(self, service, address, man):
def send_m_search_packet(self, service, address, man, mx):
packet = SSDPDatagram(
SSDPDatagram._M_SEARCH, host="{}:{}".format(SSDP_IP_ADDRESS, SSDP_PORT), st=service,
man=man, mx=1
man=man, mx=mx
)
log.debug("sending packet to %s:%i: %s", address, SSDP_PORT, packet)
self.transport.sendto(packet.encode().encode(), (address, SSDP_PORT))
async def m_search(self, address, timeout: int = 1, service='', man='') -> SSDPDatagram:
async def m_search(self, address: str, timeout: int = 1, service: str = '', man: str = '', mx: int = 1) -> SSDPDatagram:
if (address, service) in self.discover_callbacks:
return self.discover_callbacks[(address, service)]
man = man or SSDP_DISCOVER
@ -49,10 +49,10 @@ class SSDPProtocol(MulticastProtocol):
# D-Link works with both
# Cisco only works with quotes
self.send_m_search_packet(service, address, '\"%s\"' % man)
self.send_m_search_packet(service, address, '\"%s\"' % man, mx)
# DD-WRT only works without quotes
self.send_m_search_packet(service, address, man)
self.send_m_search_packet(service, address, man, mx)
f: Future = Future()
f.add_done_callback(lambda _f: outer_fut.set_result(_f.result()))
@ -125,12 +125,12 @@ async def listen_ssdp(lan_address: str, gateway_address: str,
async def m_search(lan_address: str, gateway_address: str, timeout: int = 1,
service: str = '', man: str = '', ssdp_socket: socket.socket = None) -> SSDPDatagram:
service: str = '', man: str = '', mx: int = 1, ssdp_socket: socket.socket = None) -> SSDPDatagram:
transport, protocol, gateway_address, lan_address = await listen_ssdp(
lan_address, gateway_address, ssdp_socket
)
try:
return await protocol.m_search(address=gateway_address, timeout=timeout, service=service, man=man)
return await protocol.m_search(address=gateway_address, timeout=timeout, service=service, man=man, mx=mx)
except asyncio.TimeoutError:
raise UPnPError("M-SEARCH for {}:{} timed out".format(gateway_address, SSDP_PORT))
finally:

14
aioupnp/upnp.py
View file

@ -42,23 +42,23 @@ class UPnP:
@classmethod
async def discover(cls, lan_address: str = '', gateway_address: str = '', timeout: int = 1,
service: str = '', man: str = '', interface_name: str = 'default',
service: str = '', man: str = '', mx: int = 1, 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, service, man, ssdp_socket, soap_socket
lan_address, gateway_address, timeout, service, man, mx, 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,
service: str = '', man: str = '', interface_name: str = 'default') -> Dict:
service: str = '', man: str = '', mx: int = 1, interface_name: str = 'default') -> Dict:
lan_address, gateway_address = cls.get_lan_and_gateway(lan_address, gateway_address, interface_name)
datagram = await m_search(lan_address, gateway_address, timeout, service, man)
datagram = await m_search(lan_address, gateway_address, timeout, service, man, mx)
return {
'lan_address': lan_address,
'gateway_address': gateway_address,
@ -216,7 +216,7 @@ class UPnP:
@classmethod
def run_cli(cls, method, lan_address: str = '', gateway_address: str = '', timeout: int = 60,
service: str = '', man: str = '', interface_name: str = 'default',
service: str = '', man: str = '', mx: int = 1, interface_name: str = 'default',
kwargs: dict = None) -> None:
kwargs = kwargs or {}
timeout = int(timeout)
@ -231,12 +231,12 @@ class UPnP:
async def wrapper():
if method == 'm_search':
fn = lambda *_a, **_kw: cls.m_search(
lan_address, gateway_address, timeout, service, man, interface_name
lan_address, gateway_address, timeout, service, man, mx, interface_name
)
else:
try:
u = await cls.discover(
lan_address, gateway_address, timeout, service, man, interface_name
lan_address, gateway_address, timeout, service, man, mx, interface_name
)
except UPnPError as err:
fut.set_exception(err)