tweak type hints
This commit is contained in:
binaryflesh
parent
c0827fd8f2
commit
a181c0cb87
9 changed files with 177 additions and 99 deletions
aioupnp
|
|
@ -44,11 +44,13 @@ ST
|
|||
characters in the domain name must be replaced with hyphens in accordance with RFC 2141.
|
||||
"""
|
||||
|
||||
import typing
|
||||
import collections
|
||||
from collections import OrderedDict
|
||||
|
||||
from typing import List, Any, Collection
|
||||
|
||||
from aioupnp.constants import SSDP_DISCOVER, SSDP_HOST
|
||||
|
||||
SEARCH_TARGETS: typing.List[typing.AnyStr] = [
|
||||
SEARCH_TARGETS: Collection = [
|
||||
'upnp:rootdevice',
|
||||
'urn:schemas-upnp-org:device:InternetGatewayDevice:1',
|
||||
'urn:schemas-wifialliance-org:device:WFADevice:1',
|
||||
|
|
@ -59,17 +61,27 @@ SEARCH_TARGETS: typing.List[typing.AnyStr] = [
|
|||
]
|
||||
|
||||
|
||||
def format_packet_args(order: typing.List, **kwargs) -> collections.OrderedDict:
|
||||
args: typing.List = []
|
||||
def format_packet_args(order: List, **kwargs) -> OrderedDict:
|
||||
"""Format packet arguments.
|
||||
|
||||
:param list order:
|
||||
:param dict kwargs:
|
||||
:return dict:
|
||||
"""
|
||||
args: List = []
|
||||
for o in order:
|
||||
for k, v in kwargs.items():
|
||||
if k.lower() == o.lower():
|
||||
args.append((k, v))
|
||||
break
|
||||
return collections.OrderedDict(args)
|
||||
return OrderedDict(args)
|
||||
|
||||
|
||||
def packet_generator() -> typing.Any[collections.OrderedDict]:
|
||||
def packet_generator() -> Any[OrderedDict]:
|
||||
"""Generate Packets.
|
||||
|
||||
:return dict:
|
||||
"""
|
||||
for st in SEARCH_TARGETS:
|
||||
order = ["HOST", "MAN", "MX", "ST"]
|
||||
yield format_packet_args(order, HOST=SSDP_HOST, MAN='"%s"' % SSDP_DISCOVER, MX=1, ST=st)
|
||||
|
|
|
|||
|
|
@ -4,25 +4,51 @@ import struct
|
|||
|
||||
|
||||
class MulticastProtocol(asyncio.DatagramProtocol):
|
||||
"""Multicast Protocol."""
|
||||
|
||||
def __init__(self, multicast_address, bind_address) -> None:
|
||||
"""Multicast Protocol.
|
||||
|
||||
:param str or bytes multicast_address:
|
||||
:param str or bytes bind_address:
|
||||
:return None:
|
||||
"""
|
||||
self.multicast_address = multicast_address
|
||||
self.bind_address = bind_address
|
||||
self.transport = None
|
||||
|
||||
@property
|
||||
def sock(self):
|
||||
"""Multicast Socket.
|
||||
|
||||
:return socket.socket:
|
||||
"""
|
||||
return self.transport.get_extra_info(name='socket')
|
||||
|
||||
def get_ttl(self):
|
||||
"""Get Time-To-Live for the address.
|
||||
|
||||
:return int ttl: Amount of time(seconds) an address will be static.
|
||||
"""
|
||||
_socket = self.sock()
|
||||
return _socket.getsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL)
|
||||
|
||||
def set_ttl(self, ttl):
|
||||
"""Set Time-To-Live for the address.
|
||||
|
||||
:param int ttl: Time-To-Live(seconds)
|
||||
:return None:
|
||||
"""
|
||||
_socket = self.sock()
|
||||
_ttl = struct.pack('b', ttl)
|
||||
_socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, _ttl)
|
||||
|
||||
def join_group(self, multicast_address, bind_address):
|
||||
"""Join Multicast address with bound socket address.
|
||||
|
||||
:param str multicast_address: Multicast Address
|
||||
:param str bind_address: Bound Address
|
||||
"""
|
||||
_socket = self.sock()
|
||||
p_mcast_addr = socket.inet_aton(multicast_address)
|
||||
p_bind_addr = socket.inet_aton(bind_address)
|
||||
|
|
@ -30,6 +56,11 @@ class MulticastProtocol(asyncio.DatagramProtocol):
|
|||
_socket.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, packed_ip_addr)
|
||||
|
||||
def leave_group(self, multicast_address, bind_address):
|
||||
"""Leave Multicast group.
|
||||
|
||||
:param str multicast_address: Multicast Address.
|
||||
:param str or bytes bind_address: Bind address.
|
||||
"""
|
||||
_socket = self.sock()
|
||||
p_mcast_addr = socket.inet_aton(multicast_address)
|
||||
p_bind_addr = socket.inet_aton(bind_address)
|
||||
|
|
@ -37,10 +68,19 @@ class MulticastProtocol(asyncio.DatagramProtocol):
|
|||
_socket.setsockopt(socket.IPPROTO_IP, socket.IP_DROP_MEMBERSHIP, packed_ip_addr)
|
||||
|
||||
def connection_made(self, transport):
|
||||
"""When connection is established, assign DatagramTransport.
|
||||
|
||||
:param asyncio.Transport transport:
|
||||
"""
|
||||
self.transport = transport
|
||||
|
||||
@classmethod
|
||||
def create_multicast_socket(cls, bind_address):
|
||||
"""Creates a multicast socket.
|
||||
|
||||
:param str or bytes bind_address:
|
||||
:return socket
|
||||
"""
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
sock.bind((bind_address, 0))
|
||||
|
|
|
|||
|
|
@ -1,32 +1,31 @@
|
|||
import typing
|
||||
import socket
|
||||
import asyncio
|
||||
from asyncio import DatagramTransport, DatagramProtocol
|
||||
from socket import SocketType
|
||||
|
||||
@typing.runtime
|
||||
class MulticastProtocol(asyncio.DatagramProtocol):
|
||||
def __init__(self, multicast_address: str, bind_address: str) -> typing.NoReturn:
|
||||
from typing import runtime, Any, NoReturn
|
||||
|
||||
|
||||
@runtime
|
||||
class MulticastProtocol(DatagramProtocol):
|
||||
def __init__(self, multicast_address: str, bind_address: str) -> None:
|
||||
self.multicast_address: str = multicast_address
|
||||
self.bind_address: str = bind_address
|
||||
self.transport: asyncio.DatagramTransport = None
|
||||
self.transport: Any[DatagramTransport, None] = None
|
||||
|
||||
def sock(self) -> socket.SocketType:
|
||||
def sock(self) -> SocketType:
|
||||
...
|
||||
|
||||
def get_ttl(self) -> int:
|
||||
...
|
||||
|
||||
def set_ttl(self, ttl: typing.Optional[int]) -> typing.NoReturn:
|
||||
def set_ttl(self, ttl: int = 1) -> NoReturn:
|
||||
...
|
||||
|
||||
def join_group(self, multicast_address: str, bind_address: str) -> typing.NoReturn:
|
||||
def join_group(self, multicast_address: str, bind_address: str) -> NoReturn:
|
||||
...
|
||||
|
||||
def leave_group(self, multicast_address: str, bind_address: str) -> typing.NoReturn:
|
||||
...
|
||||
|
||||
def connection_made(self, transport: asyncio.Transport) -> typing.NoReturn:
|
||||
def leave_group(self, multicast_address: str, bind_address: str) -> NoReturn:
|
||||
...
|
||||
|
||||
@classmethod
|
||||
def create_multicast_socket(cls, bind_address: str) -> socket.SocketType:
|
||||
def create_multicast_socket(cls, bind_address: str) -> SocketType:
|
||||
...
|
||||
|
|
|
|||
|
|
@ -40,6 +40,13 @@ class SCPDHTTPClientProtocol(Protocol):
|
|||
"""
|
||||
|
||||
def __init__(self, message, finished, soap_method, soap_service_id):
|
||||
"""SCPDHTTPClientProtocol.
|
||||
|
||||
:param message:
|
||||
:param finished:
|
||||
:param soap_method:
|
||||
:param soap_service_id:
|
||||
"""
|
||||
self.message = message
|
||||
self.response_buff = b''
|
||||
self.finished = finished
|
||||
|
|
@ -54,9 +61,18 @@ class SCPDHTTPClientProtocol(Protocol):
|
|||
self._body = b''
|
||||
|
||||
def connection_made(self, transport):
|
||||
transport.write(self.message)
|
||||
"""Called when connection is established.
|
||||
|
||||
:param DatagramTransport transport: Transport object.
|
||||
"""
|
||||
addr = transport.get_extra_info("peername")
|
||||
transport.sendto(self.message, addr)
|
||||
|
||||
def data_received(self, data):
|
||||
"""Called when data has been received.
|
||||
|
||||
:param bytes or str data: Data
|
||||
"""
|
||||
self.response_buff += data # TODO: make response_buff bytearray?
|
||||
for i, line in enumerate(self.response_buff.split(b'\r\n')):
|
||||
if not line: # we hit the blank line between the headers and the body
|
||||
|
|
@ -88,7 +104,13 @@ class SCPDHTTPClientProtocol(Protocol):
|
|||
|
||||
|
||||
async def scpd_get(control_url, address, port, loop):
|
||||
loop |= asyncio.get_event_loop_policy().get_event_loop()
|
||||
"""SCPD GET request.
|
||||
:param control_url:
|
||||
:param address:
|
||||
:param port:
|
||||
:param loop:
|
||||
"""
|
||||
loop = loop or asyncio.get_event_loop_policy().get_event_loop()
|
||||
finished = asyncio.Future()
|
||||
packet = serialize_scpd_get(control_url, address)
|
||||
transport, protocol = await loop.create_connection(
|
||||
|
|
|
|||
|
|
@ -1,35 +1,41 @@
|
|||
import collections
|
||||
import typing
|
||||
import asyncio
|
||||
from asyncio import DatagramProtocol, Future, transports, protocols, DatagramTransport, AbstractEventLoop
|
||||
from collections import OrderedDict
|
||||
|
||||
HTTPD_CODE_REGEX: typing.Pattern[bytes]
|
||||
from typing import Pattern, AnyStr, Any, Mapping, Union, List, Tuple, runtime, Optional, NoReturn, \
|
||||
SupportsBytes, Sized, SupportsInt, Generic, Awaitable, IO, BinaryIO, Text
|
||||
|
||||
def parse_headers(response: bytes) -> typing.Tuple[collections.OrderedDict, int, bytes]:
|
||||
HTTPD_CODE_REGEX: Union[Pattern, AnyStr, SupportsBytes]
|
||||
|
||||
|
||||
def parse_headers(response: Union[AnyStr, SupportsBytes]) -> Tuple[OrderedDict, Union[Sized, SupportsInt], Union[AnyStr, SupportsBytes]]:
|
||||
...
|
||||
|
||||
class SCPDHTTPClientProtocol(typing.runtime(asyncio.DatagramProtocol)):
|
||||
def __init__(self, message: bytes, finished: asyncio.Future, soap_method: typing.Optional[str] = None, soap_service_id: typing.Optional[str] = None) -> typing.NoReturn:
|
||||
self.message: bytes = message
|
||||
|
||||
@runtime
|
||||
class SCPDHTTPClientProtocol(DatagramProtocol):
|
||||
def __init__(self, message: str, finished: Future, soap_method: Any[Optional[bytes], None] = None, soap_service_id: Any[Optional[bytes], None] = None) -> None:
|
||||
self.message: str = message
|
||||
self.response_buff: bytes = b''
|
||||
self.finished: asyncio.Future = finished
|
||||
self.soap_method: typing.Any[str, None] = soap_method
|
||||
self.soap_service_id: typing.Any[str, None] = soap_service_id
|
||||
self.finished: Future = finished
|
||||
self.soap_method: Any[Optional[bytes], None] = soap_method
|
||||
self.soap_service_id: Any[Optional[bytes], None] = soap_service_id
|
||||
self._response_code: int = 0
|
||||
self._response_msg: bytes = b''
|
||||
self._content_length: int = 0
|
||||
self._got_headers: bool = False
|
||||
self._headers: typing.Dict = {}
|
||||
self._headers: Any[Mapping[None], Mapping[str, str]] = {}
|
||||
self._body: bytes = b''
|
||||
|
||||
|
||||
def connection_made(self, transport: asyncio.Transport) -> typing.NoReturn:
|
||||
def connection_made(self, transport: transports.DatagramTransport = DatagramTransport) -> NoReturn:
|
||||
...
|
||||
|
||||
def data_received(self, data: bytes) -> typing.NoReturn:
|
||||
def data_received(self, data: Union[bytes, Text]) -> NoReturn:
|
||||
...
|
||||
|
||||
async def scpd_get(control_url: str, address: str, port: int, loop: asyncio.AbstractEventLoop = None) -> typing.Tuple[typing.Dict, bytes, typing.Optional[Exception]]:
|
||||
|
||||
async def scpd_get(control_url: str, address: str, port: int, loop: Any[Optional[AbstractEventLoop], None] = None) -> Tuple[Mapping[str, str], str, Optional[Exception]]:
|
||||
...
|
||||
|
||||
async def scpd_post(control_url: str, address: str, port: int, method: str, param_names: typing.List[str], service_id: bytes, loop: asyncio.AbstractEventLoop = None, **kwargs) -> typing.Tuple[typing.Dict, bytes, typing.Optional[Exception]]:
|
||||
|
||||
async def scpd_post(control_url: str, address: str, port: int, method: str, param_names: List[str], service_id: bytes, loop: Any[Optional[AbstractEventLoop], None] = None, **kwargs) -> Tuple[Mapping[str, str], str, Optional[Exception]]:
|
||||
...
|
||||
|
|
|
|||
|
|
@ -1,45 +1,47 @@
|
|||
import asyncio
|
||||
import typing
|
||||
import collections
|
||||
from . import multicast
|
||||
from asyncio import Future, Handle, AbstractEventLoop, DatagramTransport
|
||||
from collections import OrderedDict
|
||||
|
||||
from typing import runtime, Union, Pattern, Set, Optional, AnyStr, Any, Tuple, NoReturn, List, SupportsFloat, SupportsBytes, SupportsInt, Generic, Text
|
||||
|
||||
from aioupnp.serialization.ssdp import SSDPDatagram
|
||||
from .multicast import MulticastProtocol
|
||||
|
||||
ADDRESS_REGEX = typing.Pattern[str]
|
||||
ADDRESS_REGEX: Union[Pattern, AnyStr]
|
||||
|
||||
@typing.runtime
|
||||
class SSDPProtocol(multicast.MulticastProtocol):
|
||||
def __init__(self, multicast_address: str, lan_address: str, ignored: typing.Set[str] = None, unicast: typing.Optional[bool] = False) -> typing.NoReturn:
|
||||
@runtime
|
||||
class SSDPProtocol(MulticastProtocol):
|
||||
def __init__(self, multicast_address: str, lan_address: str, ignored: Set[str] = None, unicast: bool = False) -> NoReturn:
|
||||
super().__init__(multicast_address, lan_address)
|
||||
self._unicast: bool = unicast
|
||||
self._ignored: typing.Set[str] = ignored or set() # ignored locations
|
||||
self._pending_searches: typing.List[typing.Tuple[str, str, asyncio.Future, asyncio.Handle]] = []
|
||||
self.notifications: typing.List = []
|
||||
self._ignored: Set[str] = ignored or set() # ignored locations
|
||||
self._pending_searches: List[Tuple[str, str, Future, Handle]] = []
|
||||
self.notifications: Any[List[None], List[str]] = []
|
||||
|
||||
def disconnect(self) -> typing.NoReturn:
|
||||
def disconnect(self) -> NoReturn:
|
||||
...
|
||||
|
||||
def _callback_m_search_ok(self, address: str, packet: SSDPDatagram) -> typing.NoReturn:
|
||||
def _callback_m_search_ok(self, address: str, packet: SSDPDatagram) -> NoReturn:
|
||||
...
|
||||
|
||||
def send_many_m_searches(self, address: str, packets: typing.List[SSDPDatagram]) -> typing.NoReturn:
|
||||
def send_many_m_searches(self, address: str, packets: List[SSDPDatagram]) -> NoReturn:
|
||||
...
|
||||
|
||||
async def m_search(self, address: str, timeout: float, datagrams: typing.List[collections.OrderedDict]) -> SSDPDatagram:
|
||||
fut: asyncio.Future = asyncio.Future()
|
||||
packets: typing.List[SSDPDatagram] = []
|
||||
async def m_search(self, address: str, timeout: Any[float, int], datagrams: List[OrderedDict[bytes]]) -> SSDPDatagram:
|
||||
fut: Future = Future()
|
||||
packets: List[SSDPDatagram] = []
|
||||
...
|
||||
|
||||
def datagram_received(self, data: bytes, addr: str) -> typing.NoReturn:
|
||||
def datagram_received(self, data: Union[Text, bytes], addr: str) -> NoReturn:
|
||||
...
|
||||
|
||||
async def listen_ssdp(lan_address: str, gateway_address: str, loop: asyncio.AbstractEventLoop = None, ignored: typing.Set[str] = None, unicast: bool = False) -> typing.Tuple[asyncio.DatagramTransport, SSDPProtocol, str, str]:
|
||||
async def listen_ssdp(lan_address: str, gateway_address: str, loop: Any[AbstractEventLoop, None] = None, ignored: Any[Optional[Set[str]], None] = None, unicast: bool = False) -> Tuple[DatagramTransport, SSDPProtocol, str, str]:
|
||||
...
|
||||
|
||||
async def m_search(lan_address: str, gateway_address: str, datagram_args: collections.OrderedDict, timeout: int, loop: asyncio.AbstractEventLoop, ignored: typing.Set[str], unicast: bool = False) -> SSDPDatagram:
|
||||
async def m_search(lan_address: str, gateway_address: str, datagram_args: OrderedDict[bytes], timeout: Any[float, int], loop: Any[Optional[AbstractEventLoop], None], ignored: Set[str], unicast: bool = False) -> SSDPDatagram:
|
||||
...
|
||||
|
||||
async def _fuzzy_m_search(lan_address: str, gateway_address: str, timeout: int = 30, loop: asyncio.AbstractEventLoop = None, ignored: typing.Set[str] = None, unicast: bool = False) -> typing.List[collections.OrderedDict]:
|
||||
async def _fuzzy_m_search(lan_address: str, gateway_address: str, timeout: Any[float, int] = 30, loop: Any[Optional[AbstractEventLoop], None] = None, ignored: Any[Optional[Set[bytes]], None] = None, unicast: bool = False) -> List[OrderedDict]:
|
||||
...
|
||||
|
||||
async def fuzzy_m_search(lan_address: str, gateway_address: str, timeout: int = 30, loop: asyncio.AbstractEventLoop = None, ignored: typing.Set[str] = None, unicast: bool = False) -> typing.Tuple[collections.OrderedDict, SSDPDatagram]:
|
||||
async def fuzzy_m_search(lan_address: str, gateway_address: str, timeout: Any[float, int] = 30, loop: Any[Optional[AbstractEventLoop], None] = None, ignored: Any[Optional[Set[bytes]], None] = None, unicast: bool = False) -> Tuple[OrderedDict, SSDPDatagram]:
|
||||
...
|
||||
|
|
|
|||
|
|
@ -1,21 +1,21 @@
|
|||
import re
|
||||
from xml.etree import ElementTree
|
||||
|
||||
from typing import Union, Pattern, AnyStr, Any, Dict, List
|
||||
from typing import Union, Pattern, Any, Mapping, List
|
||||
|
||||
from aioupnp.constants import XML_VERSION
|
||||
from aioupnp.util import etree_to_dict, flatten_keys
|
||||
|
||||
CONTENT_PATTERN: Union[Pattern, AnyStr] = re.compile("(\<\?xml version=\"1\.0\"\?\>(\s*.)*|\>)".encode())
|
||||
CONTENT_PATTERN: Union[Pattern, bytes] = re.compile("(\<\?xml version=\"1\.0\"\?\>(\s*.)*|\>)".encode())
|
||||
|
||||
XML_ROOT_SANITY_PATTERN: Union[Pattern, AnyStr] = re.compile(
|
||||
XML_ROOT_SANITY_PATTERN: Union[Pattern, str] = re.compile(
|
||||
r'(?i)(\{|(urn:schemas-[\w|\d]*-(com|org|net))[:|-](device|service)[:|-]([\w|\d|\:|\-|\_]*)|\}([\w|\d|\:|\-|\_]*))'
|
||||
)
|
||||
|
||||
XML_OTHER_KEYS: Union[Pattern, AnyStr] = re.compile(r'{[\w|\:\/\.]*}|(\w*)')
|
||||
XML_OTHER_KEYS: Union[Pattern, str] = re.compile(r'{[\w|\:\/\.]*}|(\w*)')
|
||||
|
||||
|
||||
def serialize_scpd_get(path: AnyStr, address: AnyStr) -> Any[AnyStr]:
|
||||
def serialize_scpd_get(path: str, address: str) -> bytes:
|
||||
"""Serialize SCPD GET request.
|
||||
|
||||
:param str or bytes path:
|
||||
|
|
@ -38,7 +38,7 @@ def serialize_scpd_get(path: AnyStr, address: AnyStr) -> Any[AnyStr]:
|
|||
\r\n""".encode()
|
||||
|
||||
|
||||
def deserialize_scpd_get_response(content: AnyStr) -> Any[Dict[AnyStr], Dict[None]]:
|
||||
def deserialize_scpd_get_response(content: bytes) -> Any[Mapping[bytes, str], Mapping[None]]:
|
||||
"""deserialize SCPD GET response.
|
||||
|
||||
:param str or bytes content:
|
||||
|
|
@ -52,13 +52,13 @@ def deserialize_scpd_get_response(content: AnyStr) -> Any[Dict[AnyStr], Dict[Non
|
|||
return {}
|
||||
|
||||
|
||||
def parse_device_dict(xml_dict: Dict[AnyStr]) -> Any[Dict[AnyStr], Dict[None]]:
|
||||
def parse_device_dict(xml_dict: Mapping[str, str]) -> Any[Mapping[str, str], Mapping[None]]:
|
||||
"""Parse device dictionary.
|
||||
|
||||
:param dict xml_dict:
|
||||
:return dict result:
|
||||
"""
|
||||
keys: List[AnyStr] = [xml_dict.keys()]
|
||||
keys: List[str] = [xml_dict.keys()]
|
||||
for k in keys:
|
||||
m = XML_ROOT_SANITY_PATTERN.findall(k)
|
||||
if len(m) == 3 and m[1][0] and m[2][5]:
|
||||
|
|
|
|||
|
|
@ -1,20 +1,19 @@
|
|||
import re
|
||||
from xml.etree import ElementTree
|
||||
|
||||
from typing import Union, Pattern, AnyStr, Any, List, Dict
|
||||
from typing import Union, Pattern, List, Mapping
|
||||
|
||||
from aioupnp.constants import XML_VERSION, ENVELOPE, BODY
|
||||
from aioupnp.fault import handle_fault, UPnPError
|
||||
from aioupnp.util import etree_to_dict, flatten_keys
|
||||
|
||||
CONTENT_NO_XML_VERSION_PATTERN: Union[Pattern, AnyStr] = re.compile(
|
||||
CONTENT_NO_XML_VERSION_PATTERN: Union[Pattern, str] = re.compile(
|
||||
"(\<s\:Envelope xmlns\:s=\"http\:\/\/schemas\.xmlsoap\.org\/soap\/envelope\/\"(\s*.)*\>)".encode()
|
||||
)
|
||||
|
||||
|
||||
def serialize_soap_post(method: AnyStr, param_names: List[AnyStr],
|
||||
service_id: AnyStr, gateway_address: AnyStr,
|
||||
control_url: AnyStr, **kwargs) -> AnyStr:
|
||||
def serialize_soap_post(method: str, param_names: List[str], service_id: bytes, gateway_address: bytes,
|
||||
control_url: bytes, **kwargs) -> bytes:
|
||||
"""Serialize SOAP post data.
|
||||
|
||||
:param str method:
|
||||
|
|
@ -59,8 +58,7 @@ def serialize_soap_post(method: AnyStr, param_names: List[AnyStr],
|
|||
).encode()
|
||||
|
||||
|
||||
def deserialize_soap_post_response(response: AnyStr, method: AnyStr,
|
||||
service_id: AnyStr) -> Any[Dict[AnyStr], UPnPError]:
|
||||
def deserialize_soap_post_response(response: bytes, method: bytes, service_id: bytes) -> Mapping[bytes, str]:
|
||||
"""Deserialize SOAP post.
|
||||
|
||||
:param bytes response:
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import logging
|
|||
import re
|
||||
from collections import OrderedDict
|
||||
|
||||
from typing import Dict, NamedTuple, Pattern, Any, AnyStr, List, Union
|
||||
from typing import NamedTuple, Pattern, Any, List, Union, Mapping, Optional
|
||||
|
||||
from aioupnp.constants import line_separator
|
||||
from aioupnp.fault import UPnPError
|
||||
|
|
@ -13,7 +13,7 @@ log = logging.getLogger(__name__)
|
|||
|
||||
_template: Union[Pattern, str] = r'(?i)^(%s):[ ]*(.*)$'
|
||||
|
||||
ssdp_datagram_patterns: Dict[NamedTuple[Union[Pattern, str], Any[str, int]]] = {
|
||||
ssdp_datagram_patterns: Mapping[NamedTuple[Union[Pattern, str], Any[str, int]]] = {
|
||||
"host": (re.compile(r'(?i)^(host):(.*)$'), str),
|
||||
"st": (re.compile(_template % 'st'), str),
|
||||
"man": (re.compile(_template % 'man'), str),
|
||||
|
|
@ -36,9 +36,9 @@ class SSDPDatagram(object):
|
|||
_NOTIFY: str = "NOTIFY"
|
||||
_OK: str = "OK"
|
||||
|
||||
_start_lines = {_M_SEARCH: "M-SEARCH * HTTP/1.1", _NOTIFY: "NOTIFY * HTTP/1.1", _OK: "HTTP/1.1 200 OK"}
|
||||
_friendly_names = {_M_SEARCH: "m-search", _NOTIFY: "notify", _OK: "m-search response"}
|
||||
_vendor_field_pattern = vendor_pattern
|
||||
_start_lines: Mapping[str, str] = {_M_SEARCH: "M-SEARCH * HTTP/1.1", _NOTIFY: "NOTIFY * HTTP/1.1", _OK: "HTTP/1.1 200 OK"}
|
||||
_friendly_names: Mapping[str, str] = {_M_SEARCH: "m-search", _NOTIFY: "notify", _OK: "m-search response"}
|
||||
_vendor_field_pattern: Union[Pattern, str] = vendor_pattern
|
||||
_patterns = ssdp_datagram_patterns
|
||||
|
||||
_required_fields = {
|
||||
|
|
@ -47,7 +47,7 @@ class SSDPDatagram(object):
|
|||
_OK: ["cache_control", "location", "server", "st", "usn"] # "date", "ext"
|
||||
}
|
||||
|
||||
def __init__(self, packet_type: AnyStr, kwargs: OrderedDict = None) -> None:
|
||||
def __init__(self, packet_type: str, kwargs: Mapping[str, str]) -> None:
|
||||
"""SSDP Datagram.
|
||||
|
||||
:param str or bytes packet_type:
|
||||
|
|
@ -56,7 +56,6 @@ class SSDPDatagram(object):
|
|||
if packet_type not in [self._M_SEARCH, self._NOTIFY, self._OK]:
|
||||
raise UPnPError("Unknown packet type: {}.".format(packet_type))
|
||||
self._packet_type = packet_type
|
||||
kwargs = kwargs or OrderedDict()
|
||||
self._field_order: list = [
|
||||
k.lower().replace("-", "_") for k in kwargs.keys()
|
||||
]
|
||||
|
|
@ -81,7 +80,7 @@ class SSDPDatagram(object):
|
|||
if getattr(self, k) is None:
|
||||
raise UPnPError("Missing required field %s." % k)
|
||||
|
||||
def get_cli_igd_kwargs(self) -> AnyStr:
|
||||
def get_cli_igd_kwargs(self) -> str:
|
||||
"""Get CLI IGD kwargs.
|
||||
|
||||
:return str kwargs:
|
||||
|
|
@ -92,14 +91,14 @@ class SSDPDatagram(object):
|
|||
fields.append("--%s=%s" % (self._case_mappings.get(field, field), v))
|
||||
return " ".join(fields)
|
||||
|
||||
def __repr__(self) -> AnyStr:
|
||||
def __repr__(self) -> str:
|
||||
"""repr(SSDP).
|
||||
|
||||
:return str:
|
||||
"""
|
||||
return self.as_json()
|
||||
|
||||
def __getitem__(self, item) -> Any[AnyStr, KeyError]:
|
||||
def __getitem__(self, item: str) -> Any[str, Optional[KeyError]]:
|
||||
"""SSDP.get().
|
||||
|
||||
:return str or KeyError:
|
||||
|
|
@ -109,14 +108,14 @@ class SSDPDatagram(object):
|
|||
return getattr(self, i)
|
||||
raise KeyError(item)
|
||||
|
||||
def get_friendly_name(self) -> AnyStr:
|
||||
def get_friendly_name(self) -> str:
|
||||
"""Return human friendly field name.
|
||||
|
||||
:return str:
|
||||
"""
|
||||
return self._friendly_names[self._packet_type]
|
||||
|
||||
def encode(self, trailing_newlines: Any[int] = 2) -> AnyStr:
|
||||
def encode(self, trailing_newlines: int = 2) -> str:
|
||||
"""Encode SSDP Datagram.
|
||||
|
||||
:param int trailing_newlines:
|
||||
|
|
@ -139,14 +138,14 @@ class SSDPDatagram(object):
|
|||
serialized += line_separator
|
||||
return serialized
|
||||
|
||||
def as_dict(self) -> OrderedDict:
|
||||
def as_dict(self) -> Mapping[str, str]:
|
||||
"""Dictionary representation.
|
||||
|
||||
:return dict:
|
||||
"""
|
||||
return self._lines_to_content_dict(self.encode().split(line_separator))
|
||||
|
||||
def as_json(self) -> AnyStr:
|
||||
def as_json(self) -> str:
|
||||
"""JSON representation.
|
||||
|
||||
:return bytes:
|
||||
|
|
@ -154,10 +153,10 @@ class SSDPDatagram(object):
|
|||
return json.dumps(self.as_dict(), indent=2)
|
||||
|
||||
@classmethod
|
||||
def decode(cls, datagram: AnyStr) -> Any[AnyStr, UPnPError]:
|
||||
def decode(cls, datagram: bytes) -> Any[str, Optional[UPnPError]]:
|
||||
"""Decode SSDP Datagram.
|
||||
|
||||
:param str or bytes datagram:
|
||||
:param bytes datagram:
|
||||
:return str or UPnPError:
|
||||
"""
|
||||
packet = cls._from_string(datagram.decode())
|
||||
|
|
@ -176,7 +175,7 @@ class SSDPDatagram(object):
|
|||
return packet
|
||||
|
||||
@classmethod
|
||||
def _lines_to_content_dict(cls, lines: List) -> Any[OrderedDict]:
|
||||
def _lines_to_content_dict(cls, lines: List[str]) -> Mapping[str, str]:
|
||||
"""Filter lines and return in dictionary.
|
||||
|
||||
:param list lines:
|
||||
|
|
@ -205,7 +204,7 @@ class SSDPDatagram(object):
|
|||
return result
|
||||
|
||||
@classmethod
|
||||
def _from_string(cls, datagram: AnyStr) -> Any[None, Dict[AnyStr], Dict[None]]:
|
||||
def _from_string(cls, datagram: str) -> Any[None, Mapping[str, str], Mapping[None]]:
|
||||
"""Convert str repr.
|
||||
|
||||
:param str datagram:
|
||||
|
|
@ -222,7 +221,7 @@ class SSDPDatagram(object):
|
|||
return cls._from_response(lines[1:])
|
||||
|
||||
@classmethod
|
||||
def _from_response(cls, lines: List) -> Any[Dict[AnyStr], Dict[None]]:
|
||||
def _from_response(cls, lines: List[str]) -> __class__(_OK, Mapping[str, str]):
|
||||
"""SSDP response repr.
|
||||
|
||||
:param list lines:
|
||||
|
|
@ -231,7 +230,7 @@ class SSDPDatagram(object):
|
|||
return cls(cls._OK, cls._lines_to_content_dict(lines))
|
||||
|
||||
@classmethod
|
||||
def _from_notify(cls, lines: List) -> Any[Dict[AnyStr], Dict[None]]:
|
||||
def _from_notify(cls, lines: List[str]) -> __class__(_NOTIFY, Mapping[str, str]):
|
||||
"""SSDP Notify repr.
|
||||
|
||||
:param list lines:
|
||||
|
|
@ -240,7 +239,7 @@ class SSDPDatagram(object):
|
|||
return cls(cls._NOTIFY, cls._lines_to_content_dict(lines))
|
||||
|
||||
@classmethod
|
||||
def _from_request(cls, lines: List) -> Any[Dict[AnyStr], Dict[None]]:
|
||||
def _from_request(cls, lines: List[str]) -> __class__(_M_SEARCH, Mapping[str, str]):
|
||||
"""SSDP Request repr.
|
||||
|
||||
:param list lines:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue