Home Assistant を使っていたら、UPnP の機能でルータのパケット情報を取得しているらしい。ではどうやって取得するのか？
Gemini に「UPnPの機能で Buffalo のルータの通信速度などの情報を調べるには？」と聞いたら、「SSDP（Simple Service Discovery Protocol）を使用してネットワーク上のUPnP対応デバイスを検索します。これは、マルチキャストアドレス239.255.255.250のUDPポート1900にM-SEARCHリクエストを送信することで行います。」とのアドバイスがもらえた。
「取得するプログラムを提案して」と聞いたら、以下の Python コードを教えてくれた。
import socket
import time
# SSDPマルチキャストアドレスとポート
SSDP_ADDR = '239.255.255.250'
SSDP_PORT = 1900
# M-SEARCHリクエストの作成
# ST: Service Type、ここではルーター（WANIPConnection）を指定
# MX: Maximum wait time in seconds
msearch_request = '\r\n'.join([
'M-SEARCH * HTTP/1.1',
'HOST: {}:{}'.format(SSDP_ADDR, SSDP_PORT),
'MAN: "ssdp:discover"',
'ST: urn:schemas-upnp-org:service:WANIPConnection:1',
'MX: 3',
'',
''
])
# ソケットの作成
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(5) # 応答を待機する時間（秒）
print("M-SEARCHリクエストを送信中...")
try:
# リクエストの送信
sock.sendto(msearch_request.encode('utf-8'), (SSDP_ADDR, SSDP_PORT))
while True:
try:
# 応答の受信
data, addr = sock.recvfrom(1024)
print("-" * 20)
print("デバイスを発見しました！")
print("IPアドレス:", addr[0])
print("ポート:", addr[1])
print("\n応答内容:")
print(data.decode('utf-8'))
except socket.timeout:
print("-" * 20)
print("タイムアウトしました。すべての応答を受信したか、デバイスが見つかりませんでした。")
break
except Exception as e:
print("エラーが発生しました:", e)
finally:
sock.close()
実行したら、以下の情報が得られた。これより、/getedesc.xml などで 49355 ポートに問い合わせすればいいみたい。
M-SEARCHリクエストを送信中... -------------------- デバイスを発見しました！ IPアドレス: 192.168.xx.1 ポート: 1900 応答内容: HTTP/1.1 200 OK CACHE-CONTROL: max-age=1800 DATE: Tue, 09 Sep 2025 13:57:21 GMT EXT: LOCATION: http://192.168.xx.1:49355/gatedesc.xml SERVER: Linux/3.14.77 UPnP/1.0 Intel_SDK_for_UPnP_devices/1.2 ST: urn:schemas-upnp-org:service:WANIPConnection:1 USN: (略)
- http://192.168.xx.1:49355/gatedesc.xml
- http://192.168.xx.1:49355/gateinfoSCPD.xml – 情報量少ない
- http://192.168.xx.1:49355/gateicfgSCPD.xml
- http://192.168.xx.1:49355/gateconnSCPD.xml
いろいろと試す中、送受信パケットを取得する lua スクリプトを Gemini に作らせたら、下記のようになった。
-- socket.http ライブラリをロードします。
-- 'luasocket'パッケージが必要です。
local http = require("socket.http")
local ltn12 = require("ltn12")
-- UPnPサービス情報
local url = "http://192.168.xx.1:49355/upnp/control/WANCommonIFC1"
-- リクエスト情報を格納するテーブル
local requests = {
{
action = "GetTotalBytesReceived",
tag = "<NewTotalBytesReceived>",
label = "受信"
},
{
action = "GetTotalBytesSent",
tag = "<NewTotalBytesSent>",
label = "送信"
}
}
-- データを取得する関数
local function get_upnp_data(req_info)
local soap_data = string.format([[<?xml version="1.0"?>
<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">
<s:Body>
<u:%s xmlns:u="urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1">
</u:%s>
</s:Body>
</s:Envelope>]], req_info.action, req_info.action)
local headers = {
["Content-Type"] = "text/xml; charset=\"utf-8\"",
["SOAPAction"] = '"urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1#' .. req_info.action .. '"'
}
print(req_info.label .. "データを取得中...")
local response_body = {}
local res, status_code = http.request{
url = url,
method = "POST",
headers = headers,
source = ltn12.source.string(soap_data),
sink = ltn12.sink.table(response_body),
timeout = 10
}
if status_code and (status_code == 200 or status_code == "200 OK") then
local response_text = table.concat(response_body)
local start_tag = req_info.tag
local end_tag = string.gsub(start_tag, "<", "</")
local _, _, bytes_string = string.find(response_text, start_tag .. "(.*)" .. end_tag)
if bytes_string then
local total_bytes = tonumber(bytes_string)
local megabytes = total_bytes / 1024 / 1024
print(string.format("合計%sバイト数: %d", req_info.label, total_bytes))
print(string.format("合計%sメガバイト数: %.2f MB", req_info.label, megabytes))
else
print(req_info.label .. "データが見つかりませんでした。")
print("応答内容:\n" .. response_text)
end
else
print(req_info.label .. "データの取得に失敗しました。")
print("ステータスコード: " .. (status_code or "不明"))
print("応答内容:\n" .. table.concat(response_body))
end
print("----------------------------------------")
end
-- 各リクエストを実行
for _, req_info in ipairs(requests) do
get_upnp_data(req_info)
end