Python SSDP

Python asyncio库，用于简单的服务发现协议（SSDP）。

SSDP是UPnP子标准。更多信息请参见：https://en.wikipedia.org/wiki/Simple_Service_Discovery_Protocol

安装

python3 -m pip install ssdp # lightweight, without any dependencies
# or
python3 -m pip install ssdp[cli] # with cli support for testing and debugging

用法

命令行界面

$ ssdp --help
Usage: ssdp [OPTIONS] COMMAND [ARGS]...

  SSDP command line interface.

Options:
  -v, --verbose  Increase verbosity.
  --help         Show this message and exit.

Commands:
  discover  Send out an M-SEARCH request and listening for responses.

发现

在网络中查找设备并打印响应。

ssdp discover --help
Usage: ssdp discover [OPTIONS]

  Send out an M-SEARCH request and listening for responses.

Options:
  -b, --bind TEXT             Specify alternate bind address [default: all
                              interfaces]
  --search-target, --st TEXT  Search target [default: ssdp:all]
  --max-wait, --mx INTEGER    Maximum wait time in seconds [default: 5]
  --help                      Show this message and exit.

示例

$ ssdp discover
[::]:1900 - - [Sun Jun 11 12:07:09 2023] M-SEARCH * HTTP/1.1
HOST: 239.255.255.250:1900
MAN: "ssdp:discover"
MX: 5
ST: ssdp:all

[::ffff:192.168.178.1]:1900 - - [Sun Jun 11 12:07:09 2023] HTTP/1.1 200 OK
Cache-Control: max-age=1800
Location: http://192.168.178.1:49000/MediaServerDevDesc.xml
Server: FRITZ!Box 7590 UPnP/1.0 AVM FRITZ!Box 7590 154.07.50
Ext:
ST: upnp:rootdevice
USN: uuid:fa095ecc-e13e-40e7-8e6c-3ca62f98471f::upnp:rootdevice

Python API

消息

SSDP库提供了两个用于SSDP消息的类： SSDPRequest 和 SSDPResponse。这两个类都是 SSDPMessage 的子类，并提供了以下方法

• parse：从字符串解析SSDP消息。
• __bytes__：将SSDP消息转换为字节对象。
• __str__：将SSDP消息转换为字符串。

您可以使用 parse 方法从字符串中解析SSDP消息。它将根据消息类型返回一个 SSDPRequest 或 SSDPResponse 对象。

>>> import ssdp.messages
>>> ssdp.messages.SSDPRequest.parse('NOTIFY * HTTP/1.1\r\n\r\n')
<ssdp.messages.SSDPRequest object at 0x7f8b1c0b6a90>
>>> ssdp.messages.SSDPResponse.parse('HTTP/1.1 200 OK\r\n\r\n')
<ssdp.messages.SSDPResponse object at 0x7f8b1c0b6a90>

SSDPRequest

>>> from ssdp.messages import SSDPRequest
>>> SSDPRequest('NOTIFY', headers={
...     'HOST': '10.0.0.42',
...     'NT': 'upnp:rootdevice',
...     'NTS': 'ssdp:alive',
... })
<ssdp.messages.SSDPRequest object at 0x7f8b1c0b6a90>

《SSDPRequest》类提供了一个sendto方法，用于通过开放传输发送请求。

>>> from ssdp import network, messages
>>> notify = messages.SSDPRequest('NOTIFY')
>>> notify.sendto(transport, (network.MULTICAST_ADDRESS_IPV4, network.PORT))

SSDPResponse

>>> from ssdp.messages import SSDPResponse
>>> SSDPResponse(200, 'OK', headers={
...     'CACHE-CONTROL': 'max-age=1800',
...     'LOCATION': 'http://10.0.0.1:80/description.xml',
...     'SERVER': 'Linux/2.6.18 UPnP/1.0 quick_ssdp/1.0',
...     'ST': 'upnp:rootdevice',
... })
<ssdp.messages.SSDPResponse object at 0x7f8b1c0b6a90>

Asyncio SSD协议数据报端点

《aio.SimpleServiceDiscoveryProtocol》类是《asyncio.DatagramProtocol》的子类，并提供了以下附加方法

• response_received：当接收到SSDP响应时调用。
• request_received：当接收到SSDP请求时调用。

该协议可用于在asyncio事件循环中响应SSDP消息。

此示例发送SSDP NOTIFICATION消息并打印所有接收到的SSDP消息

#!/usr/bin/env python3
import asyncio
import socket

from ssdp import aio, messages, network


class MyProtocol(aio.SimpleServiceDiscoveryProtocol):

  def response_received(self, response, addr):
    print(response, addr)

  def request_received(self, request, addr):
    print(request, addr)


loop = asyncio.get_event_loop()
connect = loop.create_datagram_endpoint(MyProtocol, family=socket.AF_INET)
transport, protocol = loop.run_until_complete(connect)

notify = messages.SSDPRequest('NOTIFY')
notify.sendto(transport, (network.MULTICAST_ADDRESS_IPV4, network.PORT))

try:
  loop.run_forever()
except KeyboardInterrupt:
  pass

transport.close()
loop.close()

SSDP解析器插件Pygments

SSDP库附带了用于突出显示SSDP消息的《Pygments》解析器插件。它基于HTTP解析器并添加了SSDP特定关键字。

您可以使用以下命令安装插件

pip install ssdp[pymgments]  # included in ssdp[cli]

您可以通过名称获取解析器

>>> from pygments.lexers import get_lexer_by_name
>>> get_lexer_by_name('ssdp')
<pygments.lexers.SSDPLexer>

突出显示SSDP消息可能如下所示

#/usr/bin/env python3
from pygments import highlight
from pygments.lexers import get_lexer_by_name
from pygments.formatters import TerminalFormatter


if __name__ == '__main__':
    lexer = get_lexer_by_name('ssdp')
    formatter = TerminalFormatter()
    code = 'NOTIFY * HTTP/1.1\r\nHOST: localhost:1900'
    msg = highlight(code, lexer, formatter)
    print(msg)