适用于Home Assistant的Python脚本

用于在Home Assistant中轻松运行Python脚本的自定义组件。这是默认python_script组件的改进版本。

安装

HACS > 集成 > 3个点 > 自定义存储库 > URL: AlexxIT/PythonScriptsPro, 类别: 集成 > 添加 > 等待 > PythonScriptsPro > 安装

或者手动将最新版本中的python_script文件夹复制到custom_components文件夹。

配置

重要： 该组件会替换标准的python_script组件！

在configuration.yaml中添加：

python_script:  # 结尾没有S！

如果需要使用其他Python库，该组件可以安装它们：

python_script:
  requirements:
  - paramiko>=2.7.1

使用python_script.exec服务

该组件创建了python_script.exec服务。
您可以运行位于任何文件夹中的外部Python脚本（使用file参数）。
或者您可以直接将Python源代码粘贴到YAML中（使用source参数）。
您可以在Python脚本中导入并使用任何库。标准的python_script组件不允许这样做。
您可以向脚本传递任何变量，就像在标准python_script中一样。
该组件会编译并缓存可执行代码，以加快下次启动速度。如果您想在不重新加载HA的情况下更改Python文件，可以使用cache: false参数禁用缓存。

从Home Assistant 2023.7版本开始，该服务将返回所有具有简单类型的脚本本地变量作为响应。

脚本中可用的变量有：

hass - Home Assistant API
data - 传递给Python脚本服务调用的数据
logger - 允许您记录消息的日志记录器

从Python文件运行脚本

在通知中显示Home Assistant的启动时间。使用我的另一个组件StartTime。将变量传递给脚本。

script:
  test_file:
    sequence:
    - service: python_script.exec
      data_template:  # 如果参数中有Jinja2模板，请使用`data_template`
        file: path_to/test_file.py  # 相对于配置文件夹的路径
        cache: false  # 如果您想在不重新加载HA的情况下更改Python文件，请禁用缓存
        title: Python从文件测试
        time_val: "{{ states('sensor.start_time')|round }}"

test_file.py

logger.debug(data)
hass.services.call('persistent_notification', 'create', {
  'title': data['title'],
  'message': f"Home Assistant启动用时 { data['time_val'] } 秒"
})
out1 = 123  # 服务响应的一些变量

从内联源代码运行脚本

在通知中显示您的IP地址。使用requests库。它默认与Home Assistant一起安装。

script:
  test_source:
    sequence:
    - service: python_script.exec
      data:
        title: Python内联测试
        source: |
          import requests
          r = requests.get('https://api.ipify.org?format=json')
          resp = r.json()
          logger.debug(resp)
          hass.services.call('persistent_notification', 'create', {
            'title': data['title'],
            'message': f"我的IP: { resp['ip'] }"
          })

远程SSH命令运行示例

这个示例完全重复了我的另一个组件 - SSHCommand的逻辑。

默认情况下没有paramiko库，但是组件可以安装它。这在Hass.io或Docker中可以正常工作。

python_script:
  requirements:
  - paramiko>=2.7.1

script:
  ssh_command:
    sequence:
    - service: python_script.exec
      data:
        file: path_to/ssh_command.py
        host: 192.168.1.123  # 可选
        user: myusername  # 可选
        pass: mypassword  # 可选
        command: ls -la

ssh_command.py

from paramiko import SSHClient, AutoAddPolicy

host = data.get('host', '172.17.0.1')
port = data.get('port', 22)
username = data.get('user', 'pi')
password = data.get('pass', 'raspberry')
command = data.get('command')

client = SSHClient()
client.set_missing_host_key_policy(AutoAddPolicy())
client.connect(host, port, username, password)
stdin, stdout, stderr = client.exec_command(command)
resp = stdout.read()
stderr.read()
client.close()

logger.info(f"SSH响应:\n{ resp.decode() }")

使用hass API的示例

在Python脚本中读取状态和属性、调用服务和触发事件的示例。

state1 = hass.states.get('sensor.start_time').state
name1 = hass.states.get('sensor.start_time').attributes['friendly_name']

if float(state1) < 30:
    hass.services.call('persistent_notification', 'create', {
        'title': "我的Python脚本",
        'message': "Home Assistant启动非常快"
    })

    hass.states.set('sensor.start_time', state1, {
        'friendly_name': f"快速 {name1}"
    })

else:
    hass.services.call('persistent_notification', 'create', {
        'title': "我的Python脚本",
        'message': "Home Assistant运行时间非常长"
    })

    hass.states.set('sensor.start_time', state1, {
        'friendly_name': f"缓慢 {name1}"
    })

hass.bus.fire('my_event_name', {
    'param1': 'value1'
})

使用python_script传感器

该组件允许您创建传感器。

配置：

您可以使用内联source或从file加载Python代码（相对于配置文件夹的路径）。
您可以为传感器设置name、icon、unit_of_measurement和scan_interval。

脚本中可用的变量有：

self.hass - Home Assistant API
self.state - 更改它以更新传感器值
self.attributes - 更改它以更新传感器属性
logger - 允许您记录消息的日志记录器

Python源代码在加载时被编译和缓存。如果Python源文件有更改，您需要重启Home Assistant。

sensor:
- platform: python_script
  name: 我的IP地址
  scan_interval: '00:05:00'  # 可选，默认：30秒
  source: |
    import requests
    r = requests.get('https://api.ipify.org?format=json')
    self.state = r.json()['ip']

- platform: python_script
  name: 我的数据库大小
  icon: mdi:database
  unit_of_measurement: MB
  scan_interval: '01:00:00'  # 可选
  source: |
    import os
    logger.debug("更新数据库大小")
    filename = self.hass.config.path('home-assistant_v2.db')
    self.state = round(os.stat(filename).st_size / 1_000_000, 1)

- platform: python_script
  name: 实例外部URL  # 更多信息 https://developers.home-assistant.io/docs/instance_url/
  scan_interval: '01:00:00'  # 可选
  source: |
    from homeassistant.helpers import network
    try:
      self.state = network.get_url(
           self.hass,
           allow_internal=False,
      )
    except network.NoURLAvailableError:
      raise MyInvalidValueError("无法为我的集成找到合适的URL")