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

viklion

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环境安装模块

[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.代码

[Python] 纯文本查看 复制代码

# -*- 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:
            try:
                # 接收服务器发送过来的数据
                __recv_Data = self.__tcp_client_socket.recv(1024)
                __recv_str = str(__recv_Data.decode('utf-8'))
                self.write_log(__recv_str)
            except Exception as e:
                self.write_log("接收消息出错，可能网络断开，error: " + str(e))
                time.sleep(5)
                continue            
            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 = 'abcdefghijk'                        #巴法云用户私钥，必填
        ,topic = 'mypc001'                     #巴法云设备主题，必填
        ,pc_mac = '11:22:33:44:55:66'                    #电脑mac地址，必填（格式：xx:xx:xx……或xx-xx-xx……）
        ,local_ip = 'auto'                  #本机ip地址，非电脑ip地址，必填（运行python的设备局域网ip地址，或填写'auto'尝试自动获取）
        ,use_ping = False               #是否开启ping检测更新电脑状态，True为开启，False为关闭
        ,use_shutdown = False           #是否需要关机指令，True为开启，False为关闭
        ,pc_ip = '192.168.1.23'                     #电脑局域网ip地址，非必填，如开启ping或关机指令，则必填
        ,pc_account = 'admin'                #电脑登录账户，非必填，如开启关机指令，则必填
        ,pc_password = '123456'               #电脑登录密码，非必填，如开启关机指令，则必填
        ,shutdown_time = 60             #延迟关机时间，非必填，如开启关机指令，则必填（单位：秒，立即关机填0）
        ,use_write_log = False          #是否开启日志记录，True为开启，False为关闭
        ,log_path = r'/opt/log'                  #日志文件路径，非必填，如开启日志记录，则必填
        ,use_send_message = False       #是否开启消息推送[方糖推送]，True为开启，False为关闭
        ,url_send_message = r'https://*******'          #[方糖推送]的个人接口，非必填，如开启消息推送，则必填
        ,channel_send_message = '9'      #[方糖推送]的频道，非必填，如开启消息推送，则必填
    )
    c.start()

5.参考内容
5.1 《用小爱同学控制台式机睡眠和唤醒的思路》，来自知乎，作者：Comzyh
https://zhuanlan.zhihu.com/p/412298207
5.2 《电脑接入米家，控制电脑开关机，并且无需购买米家外设》，来自cnblogs，作者：hackyo
https://www.cnblogs.com/hackyo/p/18000627
5.3 《python 接入，mqtt和tcp》，来自巴法开放论坛
https://bbs.bemfa.com/81
5.4 巴法文档中心
https://cloud.bemfa.com/docs/src/
5.5 Server酱 Key&API
https://sct.ftqq.com/sendkey

Nemo92

如果单纯只是语音开电脑的话，直接买个接入米家或者天猫精灵的智能插座（便宜的第三方十几块钱），然后电脑主板打开通电开机就行了（有些旧电脑不支持），非局域网也能控制，开机之后的功能用远程控制软件

viklion

flamestsui 发表于 2024-5-8 18:01
感谢楼主分享，试了下，思路很不错。
在巴法上能正常显示电脑了。但是无法关机
但是暂时没第二台电脑来测 ...

可以在电脑上调试看看
新建一个py文件和保存好的py放在同一文件夹
粘贴下面代码，运行只会执行关机指令
可以调试winrm是否有效

[Python] 纯文本查看 复制代码

from 之前保存的py文件名 import PCpower

c = PCpower(
    uid='test'  #不用改
    ,topic='test'   #不用改
    ,pc_mac='00:00:00:00:00:00' #不用改
    ,local_ip='auto'    #不用改
    ,use_shutdown=True  #不用改
    ,pc_ip='192.168.x.x'    #改成电脑ip
    ,pc_account='电脑账户名'    #改可以winrm的账户
    ,pc_password='电脑账户密码' #账户密码
    ,shutdown_time=120
)
c.pc_power_control('off')
'''
如果成功，cmd中执行'shutdown -a'取消关机
'''

ZhjhJZ

牛，家庭设备智能化了

redapple2015

这个功能强大，现在全设备向智能化发展。

心伤的天堂

这个不错

dingqh

666666   将电脑接入米家  这你是怎么想到的，太神奇了...

aigo891

只能开关机，能像控制智能音箱那样就好了，想放音乐就放音乐，想查什么就查什么

squallzcy

确实神奇，但是我一开始理解错了，哈哈哈哈，感谢大佬无私分享

淼先森

aigo891 发表于 2024-5-8 17:03
只能开关机，能像控制智能音箱那样就好了，想放音乐就放音乐，想查什么就查什么

可以搭配神秘鸭试下，我现在是这样弄的

海是倒过来的天

手残党在网上买了个几十的开机卡解决的

fast001

先收藏后续有用