简单介绍
用pyscript可以在hass中写python脚本，实现一些自动化功能，相比于homeassistant的自动化，pyscript更灵活，可以实现一些更复杂的自动化功能，pyscript执行速度相较于homeassistant的自动化更快。
更重要的是。。。可以调试 知道问题出在哪
相比NR采用轮询的方式获取状态，pyscript采用事件驱动的方式获取状态，可以减少资源消耗，能更及时对实体状态改变做出响应。
前置技能
- 有一丢丢的python基础，至少能看懂代码、会debug
安装
- hacs中找pyscript，集成中添加pyscript，安装完成后在/config/pyscript里新建xxx.py就可以
秃头开发了
- hassio加载项中安装vscode(可选，方便调试)
功能介绍
实体操作
传感器、选择器、输入框
- 获取实体状态：
#举个栗子：获取小爱收到的消息
state = state.get("sensor.xiaomi_lx06_e7c0_conversation")# 只有收到的信息：打开窗帘
state = state.getattr('sensor.xiaomi_lx06_e7c0_conversation')#得到一个python字典，包括收到的信息、TTS信息、发送的时间等
- 修改实体状态：
#举个栗子：设置温度传感器值，修改为摄氏度
state.set("sensor.ikuai_cpu_temperature", 25.5, attributes={"unit_of_measurement": "°C"})
state.setattr("sensor.ikuai_cpu_temperature", friendly_name="CPU温度传感器", unit_of_measurement="°C")
温度传感器的温度值只能使用state.get获取，state.getattr获取到的结果是{'querytime': '2024-03-19 13:38:52', 'unit_of_measurement': '°C', 'device_class': 'temperature', 'icon': 'mdi:thermometer', 'friendly_name': 'iKuai CPU_temperature'}并没有温度值
当然，也无法用state.setattr去设置温度值，因为state.setattr只能修改实体状态的属性，而温度值是实体的值，无法修改。
开关、按钮、灯、空调等
开关、按钮、灯、空调等实体，都可以通过state.get和state.getattr获取状态，但是无法通过set来设置状态。只能使用服务去开启。
服务
- 调用服务：
#举个栗子：开灯
light.turn_on(entity_id="light.mijia_group3_0112_light", brightness=255, kelvin=4000) #开灯，最高亮度，色温4000k
service.call("light", "turn_on", entity_id="light.mijia_group3_0112_light", kelvin=4000) #同上
- 注册服务：
通过python新建一个服务，执行自己想要的结果，在这个服务中可以使用state.get、state.set等操作；
然后通过@service函数注解注册服务到homeassistant中，这样就可以通过页面、HA自动化甚至service.call来调用这个服务了。
举个栗子：猫眼视频保存，参考我的另一个帖子： 米家猫眼视频本地存储
import re, requests, functools, urllib.parse
@service #注册一个猫眼视频保存的服务
def video_doorbell():
entity = camera.loock_v06_d9c1_video_doorbell
stream_address = urllib.parse.quote(getattr(entity, 'stream_address'))
motion_video_time = re.sub(r'[^0-9]', '', getattr(entity, 'motion_video_time'))
save_video(stream_address, motion_video_time)
async def save_video(stream_address, motion_video_time):
post_request = functools.partial(requests.post, 'http://192.168.0.250:5005/save_video', data={"stream_address": stream_address, "motion_video_time": motion_video_time})
response = await hass.async_add_executor_job(post_request)
return response
监听
-
实体状态变化：
自动化开始的条件之一，某个传感器、开关、插座等状态发生变化时触发自动化；
#举个栗子：监听小爱收消息
@state_trigger("sensor.xiaomi_lx06_e7c0_conversation.timestamp") #最后一次收到消息时间
#@state_trigger("sensor.xiaomi_lx06_e7c0_conversation == '关闭窗帘'") 只能使用简单等、大于、小于、不等于这几个简单判断，复杂点的参考下面的
async def get_conversation():
xiaoai_state_dict = state.getattr(sensor.xiaomi_lx06_e7c0_conversation) #获取小爱属性数据
if "content" not in xiaoai_state_dict:
log.error("获取内容失败")
return
xiaoai_stt = xiaoai_state_dict["content"] #获取小爱语音内容
if xiaoai_stt is not None and "本地播放" in xiaoai_stt:
music_msg = xiaoai_stt.replace("本地播放","").replace("的"," ").replace("歌曲","").replace("歌","")
log.error(xiaoai_state_dict["content"]) #语音内容
-
MQTT消息：
自动化开始条件之一，MQTT消息到达时触发自动化；
@mqtt_trigger(topic="home/bedroom/light/set", qos=1)
#@mqtt_trigger(str_expr="home/{room}/light/set") str_expr和topic二选一，用来不同房间发来的消息
def do_everyone(msg):
room = msg.data["room"]
light_state = msg.payload
pass
可选参数：
- topic：MQTT主题；
- qos：MQTT消息质量；
- retain：MQTT消息是否保留；
- payload：MQTT消息内容；
- str_expr：MQTT主题正则表达式；
条件判断
#举个栗子：监听小爱收消息
@state_trigger("sensor.xiaomi_lx06_e7c0_conversation.timestamp") #最后一次收到消息时间
@time_active("10:00", "12:00") #10点到12点之间
async def get_conversation():
xiaoai_state_dict = state.getattr(sensor.xiaomi_lx06_e7c0_conversation) #获取小爱属性数据
if "content" not in xiaoai_state_dict:
log.error("获取内容失败")
return
log.error(xiaoai_state_dict["content"]) #语音内容
#举个栗子：监听小爱收消息
@state_trigger("sensor.xiaomi_lx06_e7c0_conversation.timestamp") #最后一次收到消息时间
@state_active("sensor.xiaomi_lx06_e7c0_conversation == '关闭窗帘'") #只有是 关闭窗帘 时触发
async def get_conversation():
xiaoai_state_dict = state.getattr(sensor.xiaomi_lx06_e7c0_conversation) #获取小爱属性数据
if "content" not in xiaoai_state_dict:
log.error("获取内容失败")
return
log.error(xiaoai_state_dict["content"]) #语音内容
over
HACS-pyscript常用和特有的功能就这么多了，需要实现的功能就靠自己写if else了。
写完代码保存后新版HAOS会自动重载pyscript，然后就能在开发工具-服务中找到对应的服务了(自动化找不到)，低版本需要手动重启HA。
避坑
异步
在代码中，如果有涉及读写文件、请求网站等操作，需要使用async def来定义函数，否则会报错。
样例参考:
import re, requests, functools, urllib.parse
@service #注册一个猫眼视频保存的服务
def video_doorbell():
entity = camera.loock_v06_d9c1_video_doorbell
stream_address = urllib.parse.quote(getattr(entity, 'stream_address'))
motion_video_time = re.sub(r'[^0-9]', '', getattr(entity, 'motion_video_time'))
save_video(stream_address, motion_video_time)
async def save_video(stream_address, motion_video_time):
post_request = functools.partial(requests.post, 'http://192.168.0.250:5005/save_video', data={"stream_address": stream_address, "motion_video_time": motion_video_time})
response = await hass.async_add_executor_job(post_request)
return response
服务中使用多个实体
同时暂停多个实体：
media_player.media_pause(entity_id=["media_player.xiaomi_lx06_e7c0_play_control", "media_player.yun_yin_le_xiao_ai_yin_xiang_5860"])
实践
米家猫眼视频本地存储
小爱音乐本地播放(需要Alist开启匿名访问，不想开匿名也可以加上认证)
[复制链接]
发表于 2024-3-19 16:27:45
楼主