将电脑接入米家,远程、语音开关机,推送消息 

1.前言
1.1 要求:电脑主板支持网络唤醒,并且需要另一台设备运行服务,理论上可以跑python、24小时运行、能上网在同一局域网内的都行,比如路由器、nas、不用的手机等(我是运行在华硕路由器上)
1.2 门槛不低,并非拿来就能用,需要有点基础,因为要亿点点设置,所以应该也算不上是教程
1.3 搜索了各种实现方法,参考内容写在最后
1.4 并不是python高手,AI指导,欢迎给出各种意见,讨论学习

2.功能
2.1 通过小爱音箱、天猫精灵等语音开关电脑,解放双手¿
2.2 通过巴法云的app或微信小程序远程开关电脑
2.3 电脑开机或关机,会推送消息

3.实现
3.1 电脑设置:开启网络唤醒,可能需要从bios里设置,记录网卡mac地址
3.2 巴法云中新建TCP创客云虚拟设备,主题名最后以001结尾,昵称可以设置为电脑,记录下私钥和主题名
3.3 米家为例,添加第三方平台设备,选择巴法,同步设备
3.4 设备python环境安装模块

pip3 install wakeonlan pythonping pywinrm

3.5 代码中有4个方法可以自由选择是否启用
3.5.1 ping检测更新电脑状态:用于手动开关电脑后,同步更新巴法云虚拟设备的状态,如果开启消息推送,还会告诉你电脑开机/关机了
3.5.2 关机指令:如果想要语音或远程关机,需要电脑配置好winRM,并填写好参数
3.5.3 日志记录:指定目录内生成日志文件记录
3.5.4 消息推送:用的方糖推送,也就是Server酱(ServerChan),需要填写参数
3.6 填写必要参数,运行python代码
3.7 可以百度最后的参考内容获得更详细的教程

4.代码

# -*- coding: utf-8
import re
import socket
import threading
import time
from datetime import datetime
import winrm
import requests
from pythonping import ping
from wakeonlan import send_magic_packet
 
class PCpower():
    '''
    PCpower类,已测试:python3.8.18
    :param uid: 巴法云用户私钥,必填
    :param topic: 巴法云设备主题,必填
    :param pc_mac: 电脑mac地址,必填(格式:xx:xx:xx……或xx-xx-xx……)
    :param local_ip: 本机ip地址,非电脑ip地址,必填(运行python的设备局域网ip地址,或填写'auto'尝试自动获取)
    :param use_ping: 是否开启ping检测更新电脑状态,True为开启,False为关闭
    :param use_shutdown: 是否需要关机指令,True为开启,False为关闭
    :param pc_ip: 电脑局域网ip地址,非必填,如开启ping或关机指令,则必填
    :param pc_account: 电脑登录账户,非必填,如开启关机指令,则必填
    :param pc_password: 电脑登录密码,非必填,如开启关机指令,则必填
    :param shutdown_time: 延迟关机时间,非必填,如开启关机指令,则必填(单位:秒,立即关机填0)
    :param use_write_log: 是否开启日志记录,True为开启,False为关闭
    :param log_path: 日志文件路径,非必填,如开启日志记录,则必填
    :param use_send_message: 是否开启消息推送[方糖推送],True为开启,False为关闭
    :param url_send_message: [方糖推送]的个人接口,非必填,如开启消息推送,则必填
    :param channel_send_message: [方糖推送]的频道,非必填,如开启消息推送,则必填
    '''
    def __init__(self, uid:str, topic:str, pc_mac:str, local_ip:str, use_ping:bool=False, use_shutdown:bool=False, pc_ip:str ='', pc_account:str='', pc_password:str='', shutdown_time:int=0, use_write_log:bool=False, log_path:str='', use_send_message:bool=False, url_send_message:str='', channel_send_message:str=''):
        self.__uid = uid
        self.__topic = topic
        self.__pc_mac = pc_mac
        self.__local_ip = local_ip if not local_ip == 'auto' else self.get_ip_address()
        self.__use_ping = use_ping
        self.__use_shutdown = use_shutdown
        self.__pc_ip = pc_ip
        self.__pc_account = pc_account
        self.__pc_password = pc_password
        self.__shutdown_time = shutdown_time
        self.__use_write_log = use_write_log
        self.__log_path = log_path
        self.__use_send_message = use_send_message
        self.__url_send_message = url_send_message
        self.__channel_send_message = channel_send_message
         
        self.__pc_state = None
        self.__t_check = None
         
        self.__check_correct(self.__uid, 'uid')
        self.__check_correct(self.__topic, 'topic')
        self.__check_correct(self.__pc_mac, 'pc_mac', is_mac=True)
        self.__check_correct(self.__local_ip, 'local_ip', is_ip=True)
        self.__check_correct(self.__use_ping, 'use_ping', is_bool=True)
        self.__check_correct(self.__use_shutdown, 'use_shutdown', is_bool=True)
        self.__check_correct(self.__use_write_log, 'use_write_log', is_bool=True)
        self.__check_correct(self.__use_send_message, 'use_send_message', is_bool=True)
 
        if self.__use_ping or self.__use_shutdown:
            self.__check_correct(self.__pc_ip, 'pc_ip', is_ip=True)
        if self.__use_shutdown:
            self.__check_correct(self.__pc_account, 'pc_account')
            self.__check_correct(self.__pc_password, 'pc_password')
            self.__check_correct(self.__shutdown_time, 'shutdown_time', is_int=True)
        if self.__use_write_log:
            self.__check_correct(self.__log_path, 'log_path')
        if self.__use_send_message:
            self.__check_correct(self.__url_send_message, 'url_send_message')
            self.__check_correct(self.__channel_send_message, 'channel_send_message')
             
        print("初始化成功")
        self.__write_log("初始化成功")
        if local_ip == 'auto':
            print(f"自动获取本机局域网ip地址:{self.__local_ip}")
            self.__write_log(f"自动获取本机局域网ip地址:{self.__local_ip}")
     
    def __check_correct(self, self_var, var_name, is_mac=False, is_ip=False, is_bool=False, is_int=False):
        if is_mac:  
            if not isinstance(self_var, str) or not self.check_mac_address(self_var):  
                raise ValueError(f"{var_name}必须是有效的MAC地址,当前为:{self_var}")  
        elif is_ip:  
            if not isinstance(self_var, str) or not self.check_ip_address(self_var):  
                raise ValueError(f"{var_name}必须是有效的IP地址,当前为:{self_var}")  
        elif is_bool:  
            if not isinstance(self_var, bool):  
                raise TypeError(f"{var_name}必须是True或False,当前为:{self_var},类型:{type(self_var)}")  
        elif is_int:  
            if not isinstance(self_var, int) or self_var < 0:  
                raise ValueError(f"{var_name}必须是大于等于0的整数,当前为:{self_var},类型:{type(self_var)}")  
        else:  
            if not isinstance(self_var, str) or not self_var.strip():  
                raise ValueError(f"{var_name}必须是非空字符串")  
         
    @staticmethod
    def check_mac_address(mac):
        # 定义正则表达式模式
        pattern = r'^(([0-9A-Fa-f]{2}:){5}([0-9A-Fa-f]{2})|([0-9A-Fa-f]{2}-){5}([0-9A-Fa-f]{2}))$'
        # 利用re模块进行匹配
        return bool(re.match(pattern, mac))
     
    @staticmethod
    def check_ip_address(ip):
        pattern = r'^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
        return bool(re.match(pattern, ip))
         
    @classmethod
    def get_time(cls):
        # 获取当前时间
        time_now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        return time_now
     
    @classmethod
    def get_ip_address(cls):
        try:
            # 创建一个socket对象
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            # 连接到目标网址
            s.connect(("192.168.255.255", 80))
            # 获取本地IP地址
            ip_address = s.getsockname()[0]
            # 关闭socket连接
            s.close()
            return ip_address
        except socket.error:
            return "无法获取IP地址"
 
    #订阅
    def __connTCP(self):
        # 创建socket
        self.__tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # IP和端口
        server_ip = 'bemfa.com'
        server_port = 8344
        try:
            # 连接服务器
            self.__tcp_client_socket.connect((server_ip, server_port))
            # 发送订阅指令
            substr = f'cmd=1&uid={self.__uid}&topic={self.__topic}\r\n'
            self.__tcp_client_socket.send(substr.encode('utf-8'))
            # 写入日志
            self.__write_log("订阅成功")
        except:
            self.__write_log("订阅失败,正在重试")
            time.sleep(2)
            self.__connTCP()
 
    #心跳
    def __Ping(self):
        # 发送心跳
        try:
            keeplive = 'ping\r\n'
            self.__tcp_client_socket.send(keeplive.encode('utf-8'))
            self.__write_log("已发送心跳")
        except:
            time.sleep(2)
            self.__write_log("发送心跳出现问题,重新订阅")
            self.__connTCP()
        #开启定时,30秒发送一次心跳
        t = threading.Timer(30, self.__Ping)
        t.start()
 
    #定时ping电脑检测开关机状态
    def __check_pc_state(self, is_power_off):
        if self.__use_ping:
            ping_result = str(ping(self.__pc_ip, timeout = 1, count = 1))  #timeout:超时时间,count:包的个数
            self.__write_log('\n' + ping_result + '\n' + '---------------------', '_ping_result')
            if 'Reply' in ping_result:
                is_power_off = False
                new_state = "电脑已开机"
                message = "开机"
                update_state = "on"
            elif 'timed out' in ping_result:
                if not is_power_off:
                    self.__check_pc_state(True)
                    return
                new_state = "电脑已关机"
                message = "关机"
                update_state = "off"
            if self.__pc_state != new_state:
                self.__pc_state = new_state
                substr = f'cmd=2&uid={self.__uid}&topic={self.__topic}/up&msg={update_state}\r\n'
                self.__tcp_client_socket.send(substr.encode("utf-8"))
                self.__write_log(f"电脑状态更新为:{message}")
                self.__send_message(f"电脑状态更新为:{message}")
            #开启定时,120秒ping一次pc
            self.__t_check = threading.Timer(120, self.__check_pc_state, args=(is_power_off,))
            self.__t_check.start()
 
    #重置ping定时
    def __restart_pc_check(self, value):
        if self.__use_ping:
            try:
                self.__t_check.cancel()
                self.__t_check = threading.Timer(120, self.__check_pc_state, args=(value,))
                self.__t_check.start()
            except Exception as e:
                self.__write_log(" error: " + str(e))
 
    #推送消息
    def __send_message(self, content):
        if self.__use_send_message:
            data = {
                    'title': content,
                    'desp': self.get_time(),
                    'channel': self.__channel_send_message
                }
            _ = requests.post(self.__url_send_message, data=data)
 
    #写入日志
    def __write_log(self, content, nameadd = ''):
        if self.__use_write_log:
            # 打开日志文件,"a"追加写入
            time_day = datetime.now().strftime('%Y-%m-%d')
            try:
                with open(rf'{self.__log_path}/{time_day}{nameadd}.log', 'a') as file:
                    file.write(self.get_time() + ' ' + content + '\n')
            except Exception as e:
                print(self.get_time() + " 写入日志出错了:\n" + str(e))
     
    #网络唤醒
    def wake_on_lan(self):
        # subprocess.Popen([f'ether-wake', '-i', 'br0', '-b', '{self.__pc_mac}']) 
        send_magic_packet(self.__pc_mac,  interface = self.__local_ip)
        self.__pc_state = "电脑已开机"
        self.__restart_pc_check(False)
        self.__write_log(self.__pc_state)
        self.__send_message(self.__pc_state)
 
    #收到消息后执行开关机
    def pc_power_control(self, state):
        if state == 'on' and (self.__use_ping and self.__pc_state != "电脑已开机" or not self.__use_ping):
            self.wake_on_lan()
        elif state == 'off' and (self.__use_ping and self.__pc_state != "电脑已关机" or not self.__use_ping):
            self.__restart_pc_check(True)
            if self.__use_shutdown:
                try:
                    session = winrm.Session(self.__pc_ip, auth=(self.__pc_account, self.__pc_password))
                    _ = session.run_cmd(f'shutdown -s -t {self.__shutdown_time}')
                except Exception as e:
                    print(self.get_time() + " 电脑可能已经关机,error: " + str(e))
                    self.__write_log(str(e) + "电脑可能已经关机")
            self.__pc_state = "电脑已关机"
            self.__write_log(self.__pc_state)
            self.__send_message(self.__pc_state)
 
    #启动
    def start(self):
        self.__connTCP()
        self.__Ping()
        self.__check_pc_state(False)
        while True:
            # 接收服务器发送过来的数据
            __recv_Data = self.__tcp_client_socket.recv(1024)
            __recv_str = str(__recv_Data.decode('utf-8'))
            self.__write_log(__recv_str)
             
            if not __recv_Data:
                self.__write_log("服务器未返回任何数据,重新订阅")
                self.__connTCP()
             
            elif f'uid={self.__uid}' in __recv_str:
                #收到开机消息
                if 'msg=on' in __recv_str:
                    self.pc_power_control('on')
                #收到关机消息
                elif 'msg=off' in __recv_str:
                    self.pc_power_control('off')
 
 
if __name__ == '__main__':
    c = PCpower(
        uid = ''                        #巴法云用户私钥,必填
        ,topic = ''                     #巴法云设备主题,必填
        ,pc_mac = ''                    #电脑mac地址,必填(格式:xx:xx:xx……或xx-xx-xx……)
        ,local_ip = ''                  #本机ip地址,非电脑ip地址,必填(运行python的设备局域网ip地址,或填写'auto'尝试自动获取)
        ,use_ping = False               #是否开启ping检测更新电脑状态,True为开启,False为关闭
        ,use_shutdown = False           #是否需要关机指令,True为开启,False为关闭
        ,pc_ip = ''                     #电脑局域网ip地址,非必填,如开启ping或关机指令,则必填
        ,pc_account = ''                #电脑登录账户,非必填,如开启关机指令,则必填
        ,pc_password = ''               #电脑登录密码,非必填,如开启关机指令,则必填
        ,shutdown_time = 60             #延迟关机时间,非必填,如开启关机指令,则必填(单位:秒,立即关机填0)
        ,use_write_log = False          #是否开启日志记录,True为开启,False为关闭
        ,log_path = ''                  #日志文件路径,非必填,如开启日志记录,则必填
        ,use_send_message = False       #是否开启消息推送[方糖推送],True为开启,False为关闭
        ,url_send_message = ''          #[方糖推送]的个人接口,非必填,如开启消息推送,则必填
        ,channel_send_message = ''      #[方糖推送]的频道,非必填,如开启消息推送,则必填
    )
    c.start()

5.参考内容
5.1 《用小爱同学控制台式机睡眠和唤醒的思路》,来自知乎,作者:Comzyh
5.2 《电脑接入米家,控制电脑开关机,并且无需购买米家外设》,来自cnblogs,作者:hackyo
5.3 《python 接入,mqtt和tcp》,来自巴法开放论坛
5.4 巴法文档中心

给TA打赏
共{{data.count}}人
人已打赏
未分类

小米汽车SU7宣传视频

2024-3-30 16:54:49

未分类

YouTube转载:M3Max MacBook开箱评测

2024-1-3 9:33:52

0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
今日签到
有新私信 私信列表
搜索