海波图书馆

Hyperion-NG 的Python 库 。有关此库的输入和输出的更多详细信息，请参阅JSON API。

安装

$ pip3 install hyperion-py

用法

数据模型哲学

虽然并非普遍适用，但该库试图精确表示Hyperion JSON 文档中定义的数据模型、API 和参数。因此，没有尝试（故意）以比模型已经支持的更精细的粒度级别呈现方便的访问器/调用。这是为了确保客户端有足够的机会保持功能，而不管服务器的底层数据模型如何更改，并且将更改与服务器的数据模型（例如新的 Hyperion 服务器功能）相匹配的责任属于调用者。

构造函数参数

以下参数可以传递给HyperionClient构造函数：

连接、断开和客户端控制调用

• async_client_connect(): 连接客户端。
• async_client_disconnect(): 断开客户端。
• async_client_login()：登录已连接的客户端。async_client_connect()除非raw_connection构造函数参数为 True ，否则自动调用 。
• async_client_switch_instance()：切换到 Hyperion 服务器上配置的实例。async_client_connect()除非raw_connection 构造函数参数为 True ，否则自动调用。

原生 API 调用

所有 API 调用都可以在 client.py中找到。所有异步调用都以async_.

请注意，上述链接文档中显示的command和subcommand键将自动包含在客户端发送的调用中，无需指定。

客户输入/输出

API 参数和输出都在JSON API 文档中定义。

示例用法：

#!/usr/bin/env python
"""Simple Hyperion client read demonstration."""

import asyncio

from hyperion import client, const

HOST = "hyperion"


async def print_brightness() -> None:
    """Print Hyperion brightness."""

    async with client.HyperionClient(HOST) as hyperion_client:
        assert hyperion_client

        adjustment = hyperion_client.adjustment
        assert adjustment

        print("Brightness: %i%%" % adjustment[0][const.KEY_BRIGHTNESS])


if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(print_brightness())

在后台运行

后台asyncio task运行以处理所有连接后的入站数据（例如，请求响应，或来自服务器端状态更改的订阅更新）。此后台任务必须在连接后启动，或者启动（它自己会建立连接）。

可选地，此后台任务可以将回调回调给用户。

等待回复

如果用户进行了名称中没有的调用_send_（见上表），函数调用将等待响应并将其返回给调用者。这种请求和响应的匹配是通过tan参数完成的。如果未指定，客户端将自动附加一个tan整数，这将在返回的输出数据中可见。这种匹配对于区分由于请求的响应和来自订阅更新的“自发数据”是必要的。

示例：等待响应

#!/usr/bin/env python
"""Simple Hyperion client request demonstration."""

import asyncio

from hyperion import client

HOST = "hyperion"


async def print_if_auth_required() -> None:
    """Print whether auth is required."""

    hc = client.HyperionClient(HOST)
    await hc.async_client_connect()

    result = await hc.async_is_auth_required()
    print("Result: %s" % result)

    await hc.async_client_disconnect()


asyncio.get_event_loop().run_until_complete(print_if_auth_required())

输出：

Result: {'command': 'authorize-tokenRequired', 'info': {'required': False}, 'success': True, 'tan': 1}

示例：发送命令

发送命令的稍微复杂一点的示例（以给定的优先级清除 Hyperion 源选择，然后以相同的优先级设置颜色）。

#!/usr/bin/env python
"""Simple Hyperion client request demonstration."""

import asyncio
import logging
import sys

from hyperion import client

HOST = "hyperion"
PRIORITY = 20


async def set_color() -> None:
    """Set red color on Hyperion."""

    async with client.HyperionClient(HOST) as hc:
        assert hc

        if not await hc.async_client_connect():
            logging.error("Could not connect to: %s", HOST)
            return

        if not client.ResponseOK(
            await hc.async_clear(priority=PRIORITY)
        ) or not client.ResponseOK(
            await hc.async_set_color(
                color=[255, 0, 0], priority=PRIORITY, origin=sys.argv[0]
            )
        ):
            logging.error("Could not clear/set_color on: %s", HOST)
            return


logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
asyncio.get_event_loop().run_until_complete(set_color())

示例：启动和切换实例

下面的示例将启动一个停止的实例，等待它准备好，然后切换到它。使用回调，下面讨论。

#!/usr/bin/env python
"""Simple Hyperion client request demonstration."""

from __future__ import annotations

import asyncio
import logging
import sys
from typing import Any

from hyperion import client

HOST = "hyperion"
PRIORITY = 20


async def instance_start_and_switch() -> None:
    """Wait for an instance to start."""

    instance_ready = asyncio.Event()

    def instance_update(json: dict[str, Any]) -> None:
        for data in json["data"]:
            if data["instance"] == 1 and data["running"]:
                instance_ready.set()

    async with client.HyperionClient(
        HOST, callbacks={"instance-update": instance_update}
    ) as hc:
        assert hc

        if not client.ResponseOK(await hc.async_start_instance(instance=1)):
            logging.error("Could not start instance on: %s", HOST)
            return

        # Blocks waiting for the instance to start.
        await instance_ready.wait()

        if not client.ResponseOK(await hc.async_switch_instance(instance=1)):
            logging.error("Could not switch instance on: %s", HOST)
            return


logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
asyncio.get_event_loop().run_until_complete(instance_start_and_switch())

回调

客户端可以配置为在 Hyperion 服务器报告新值时进行回调。支持两类回调：

• default_callback：当未指定更具体的回调时，将调用此回调。
• 回调：以 Hyperion 订阅“命令”为键的回调字典（请参阅JSON API 文档）

回调可以在HyperionClient构造函数（default_callback=或callbacks=参数）中指定，也可以在构造之后通过 set_callbacks()andset_default_callback()方法指定。

如上所述，callbacksdict 以相关的 Hyperion 订阅为键 command（例如components-update，priorities-update）。client-update客户端还使用以下形式的命令提供自定义回调：

{"command": "client-update",
 "connected": True,
 "logged-in": True,
 "instance": 0,
 "loaded-state": True}

这可用于在客户端与服务器连接或断开连接时采取特殊操作。

示例：回调

#!/usr/bin/env python
"""Simple Hyperion client callback demonstration."""

from __future__ import annotations

import asyncio
from typing import Any

from hyperion import client

HOST = "hyperion"


def callback(json: dict[str, Any]) -> None:
    """Sample callback function."""

    print("Received Hyperion callback: %s" % json)


async def show_callback() -> None:
    """Show a default callback is called."""

    async with client.HyperionClient(HOST, default_callback=callback):
        pass


if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(show_callback())

输出，显示连接阶段的进展：

Received Hyperion callback: {'connected': True, 'logged-in': False, 'instance': None, 'loaded-state': False, 'command': 'client-update'}
Received Hyperion callback: {'connected': True, 'logged-in': True, 'instance': None, 'loaded-state': False, 'command': 'client-update'}
Received Hyperion callback: {'connected': True, 'logged-in': True, 'instance': 0, 'loaded-state': False, 'command': 'client-update'}
Received Hyperion callback: {'command': 'serverinfo', ... }
Received Hyperion callback: {'connected': True, 'logged-in': True, 'instance': 0, 'loaded-state': True, 'command': 'client-update'}

ThreadedHyperionClient

AThreadedHyperionClient也作为非异步代码的便利包装器提供。用ThreadedHyperionClient非异步版本包装异步调用（方法命名如上所示，除了不以 开头 async_）。

等待线程初始化客户端

线程必须有机会在与其交互之前初始化客户端。此方法调用将阻塞调用者，直到客户端被初始化。

• wait_for_client_init()

线程客户端的示例使用

#!/usr/bin/env python
"""Simple Threaded Hyperion client demonstration."""

from hyperion import client, const

HOST = "hyperion"

if __name__ == "__main__":
    hyperion_client = client.ThreadedHyperionClient(HOST)

    # Start the asyncio loop in a new thread.
    hyperion_client.start()

    # Wait for the client to initialize in the new thread.
    hyperion_client.wait_for_client_init()

    # Connect the client.
    hyperion_client.client_connect()

    print("Brightness: %i%%" % hyperion_client.adjustment[0][const.KEY_BRIGHTNESS])

    # Disconnect the client.
    hyperion_client.client_disconnect()

    # Stop the loop (will stop the thread).
    hyperion_client.stop()

    # Join the created thread.
    hyperion_client.join()

输出：

Brightness: 59%

异常/错误

哲学

无论网络环境如何，HyperionClient 都力求不抛出异常，重新连接将在后台自动发生。仅在可能出现程序员错误的情况下（有意）引发异常。

超离子错误

不直接引发，但其他异常继承自此。

HyperionClientTanNotAvailable

tan如果向 API 调用提供了参数，但该 tan参数已被另一个正在进行的调用使用，则会引发异常。用户要么根本不指定tan（客户端库将以增量方式自动管理它），要么如果手动指定，则调用者有责任确保没有两个同时调用共享 a tan（否则客户端将无法将调用与响应相匹配，并且此异常将在调用之前自动引发）。

“任务已销毁，但未决！”

如果HyperionClient对象已连接但在断开连接之前已销毁，则可能会打印一条警告消息（“任务已销毁，但未决！”）。为避免这种情况，请确保始终async_client_disconnect在销毁连接的客户端之前调用。或者使用异步上下文管理器：

async with client.HyperionClient(TEST_HOST, TEST_PORT) as hc:
    if not hc:
        return
    ...

超时

客户端自由使用超时，可以在多个级别指定：

• 在客户端构造函数参数timeout_secs中，用于连接和请求。
• 在每个请求中使用timeout_secs单个调用的参数

超时值：

• None：如果None用作超时，客户端将永远等待。
• 0：如果0用作超时，将使用客户端默认值（在构造函数中指定）。
• >0.0：将使用此秒数（或部分秒数）。

默认情况下，所有请求都将遵守timeout_secs客户端构造函数中指定的请求，除非显式覆盖并默认为 5 秒（请参阅const.py）。一个例外是async_send_request_token它具有更大的默认值（180 秒，请参阅const.py），因为此请求涉及用户需要在调用能够返回之前与 Hyperion UI 进行交互。

帮手

响应OK

提供了一些方便的可调用类来确定服务器响应是否成功。

• ResponseOK：是否有任何 Hyperion 命令响应成功（一般）。
• ServerInfoResponseOK: aasync_get_serverinfo是否成功。
• LoginResponseOK: 是否async_login成功。
• SwitchInstanceResponseOK:async_switch_instance命令是否成功。

示例用法

if not client.ResponseOK(await hc.async_clear(priority=PRIORITY))

身份验证 ID

请求身份验证令牌时，可以指定 5 个字符的 ID 以确保管理员用户正在授权来自正确来源的正确请求。默认情况下async_request_token会随机生成一个 ID，但如果需要允许用户确认匹配，则可以显式提供。在这种情况下，此辅助方法可用。

• generate_random_auth_id：生成一个随机的 5 个字符的身份验证 ID，用于外部显示并包含在对async_request_token.

示例用法

auth_id  = hc.generate_random_auth_id()
hc.async_send_login(comment="Trustworthy actor", id=auth_id)
# Show auth_id to the user to allow them to verify the origin of the request,
# then have them visit the Hyperion UI.