SSDP协议的Python示例「建议收藏」

SSDP协议的Python示例「建议收藏」废话少说,直接上代码。服务端ssdp_server.py

大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE稳定放心使用

废话少说,直接上代码。服务端ssdp_server.py:

# -*- coding: utf-8 -*-

import socket

SSDP_ADDR = '239.255.255.250'
SSDP_PORT = 1900
SERVICE_NAME = 'my_service'


class Connection():
    def __init__(self, s, data, addr):
        self.__s = s
        self.__data = data
        self.__addr = addr
        self.is_find_service = False

    def handle_request(self):
        if self.__data.startswith('M-SEARCH * HTTP/1.1\r\n'):
            self.__handle_search()
        elif self.__data.startswith('HTTP/1.1 200 OK\r\n'):
            self.__handle_ok()

    def __handle_search(self):
        props = self.__parse_props(['HOST', 'MAN', 'ST', 'MX'])
        if not props:
            return

        if props['HOST'] != '%s:%d' % (SSDP_ADDR, SSDP_PORT) \
                or props['MAN'] != '"ssdp:discover"' \
                or props['ST'] != 'ssdp:all':
            return

        print 'RECV: %s' % str(self.__data)
        print 'ADDR: %s' % str(self.__addr)

        response = 'HTTP/1.1 200 OK\r\nST: %s\r\n\r\n' % SERVICE_NAME
        self.__s.sendto(response, self.__addr)

    def __handle_ok(self):
        props = self.__parse_props(['ST'])
        if not props:
            return

        if props['ST'] != SERVICE_NAME:
            return

        print 'RECV: %s' % str(self.__data)
        print 'ADDR: %s' % str(self.__addr)
        print 'Find service!!!!'

        self.is_find_service = True

    def __parse_props(self, target_keys):
        lines = self.__data.split('\r\n')

        props = {}
        for i in range(1, len(lines)):
            if not lines[i]:
                continue

            index = lines[i].find(':')
            if index == -1:
                return None

            props[lines[i][:index]] = lines[i][index + 1:].strip()

        if not set(target_keys).issubset(set(props.keys())):
            return None

        return props

class SSDPServer():
    def __init__(self):
        self.__s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.__s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        local_ip = socket.gethostbyname(socket.gethostname())
        any_ip = '0.0.0.0'

        # 绑定到任意地址和SSDP组播端口上
        self.__s.bind((any_ip, SSDP_PORT))

        # INFO: 使用默认值
        # self.__s.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_TTL, 20)
        # self.__s.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_LOOP, 1)
        # self.__s.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_IF,
        #                     socket.inet_aton(intf) + socket.inet_aton('0.0.0.0'))
        # INFO: 添加到多播组
        self.__s.setsockopt(socket.SOL_IP, socket.IP_ADD_MEMBERSHIP,
                            socket.inet_aton(SSDP_ADDR) + socket.inet_aton(local_ip))
        self.local_ip = local_ip

    def start(self):
        while True:
            data, addr = self.__s.recvfrom(2048)
            conn = Connection(self.__s, data, addr)
            conn.handle_request()
        self.__s.setsockopt(socket.SOL_IP, socket.IP_DROP_MEMBERSHIP,
                            socket.inet_aton(SSDP_ADDR) + socket.inet_aton(self.local_ip))
        self.__s.close()

if __name__ == '__main__':
    port = SSDPServer()
    port.start()

客户端ssdp_client.py:

# -*- coding: utf-8 -*-

import socket
import time
import select
import ssdp_server

SSDP_ADDR = '239.255.255.250'
SSDP_PORT = 1900

MS = 'M-SEARCH * HTTP/1.1\r\nHOST: %s:%d\r\nMAN: "ssdp:discover"\r\nMX: 2\r\nST: ssdp:all\r\n\r\n' \
     % (SSDP_ADDR, SSDP_PORT)


class SSDPClient():
    def __init__(self):
        self.__s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

        # INFO: 若绑定,服务端收到的是固定的地址和端口号
        self.__s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        local_ip = socket.gethostbyname(socket.gethostname())
        self.__s.bind((local_ip, 50000))

    def start(self):
        self.__send_search()
        while True:
            reads, _, _ = select.select([self.__s], [], [], 5)
            if reads:
                data, addr = self.__s.recvfrom(2048)
                conn = ssdp_server.Connection(self.__s, data, addr)
                conn.handle_request()
                if conn.is_find_service:
                    break
            else:  # timeout
                self.__send_search()
        self.__s.close()

    def __send_search(self):
        print "Sending M-SEARCH..."
        # INFO: 发送到SSDP组播地址上
        self.__s.sendto(MS, (SSDP_ADDR, SSDP_PORT))

if __name__ == '__main__':
    port = SSDPClient()
    port.start()

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/183233.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • CANoe/CANalyzer诊断功能的深入理解以及CAPL诊断编程实现[通俗易懂]

    CANoe/CANalyzer诊断功能的深入理解以及CAPL诊断编程实现[通俗易懂]之前和大家分享了CANoe的基础使用(分析、仿真、测试、诊断),这篇文章将继续深入探讨如何使用CANoe/CANalyzer中的诊断功能。诊断用于在将ECU安装到系统之前或之后配置,维护,支持,控制和扩展ECU,例如,一辆车。诊断通常在请求-响应方案中执行:测试仪(客户端)向…

  • C语言程序设计第五版 谭浩强 第五版课后答案

    C语言程序设计第五版 谭浩强 第五版课后答案谭浩强C语言程序设计第五版第4章课后答案3.求两个正整数m和n,求其最大公约数和最小公倍数。#include<stdio.h>voidmain(){ intm,n,t,i,a=1; scanf(“%d%d”,&m,&n); if(m<n) { t=m; m=n; n=t; } for(i…

  • 硬币问题 动态规划_动态规划

    硬币问题 动态规划_动态规划动态规划-硬币问题分析

  • 基于麦克风阵列的声源定位_python播放声音模块

    基于麦克风阵列的声源定位_python播放声音模块上一篇文章说到odas_web界面非常难安装,并且运行也很卡。所以我自己用python写了一个界面程序,用来接收odas处理完的结果。这个界面程序与odas之间是通过socket连接的,界面作为服务器,odas作为客户端,由于有两路数据,所以各有两个服务器和客户端。但是实际绘制在界面上的是SSL的结果,不是SST的结果。其实我也试过SST的结果,从直观的感受而言,效果会比SSL差一些,实时性不是很高,我的理解SST的好处是可以跟踪音源是否有活动。

  • MyBatisPlus–逻辑删除「建议收藏」

    MyBatisPlus–逻辑删除「建议收藏」逻辑删除开发系统时,有时候在实现功能时,删除操作需要实现逻辑删除,所谓欧吉删除就是将数据标记为删除,而并非真正的物理删除(非DELETE操作),查询时需要携带状态条件,确保被标记的数据不被查询,这样做的目的就是避免数据被真正的删除。配置application.properties#删除状态值为1mybatis-plus.global-config.db-config.logic-del…

  • 测试工程师常见面试题及答案_测试工程师数据库面试题

    测试工程师常见面试题及答案_测试工程师数据库面试题1、测试人员在软件开发过程中的任务是什么?(初级)(5分)答:1、寻找Bug;2、避免软件开发过程中的缺陷;3、衡量软件的品质;4、关注用户的需求。总的目标是:确保软件的质量。2、在您以往的工作中,一条软件缺陷(或者叫Bug)记录都包含了哪些内容?如何提交高质量的软件缺陷(Bug)记录?(初级)(6分)答:一条Bug记录最基本应包含:编号、Bug所属模块、Bug描述、Bug级别、…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号