obs-websocket-py

用于与obs-websocket服务器通信的Python库。这是使用trio异步库对obs-websocket-py的分支。

根据MIT许可证授权

项目页面

GitHub项目：https://github.com/mkdryden/obs-websocket-py-trio

PyPI软件包：https://pypi.ac.cn/project/obs-websocket-py-trio/

安装

只需在您的Python venv或直接在系统上运行pip install obswebsocket-trio。

对于开发，从源目录运行poetry install将生成包含所有依赖项的venv。

对于手动安装，克隆github仓库并将目录obswebsocket_trio复制到您的Python项目根目录。

用法

请参阅samples目录中的Python脚本。

与原始obs-websocket-py相比，最大的变化是使用了trio异步库。这意味着大多数ObsWS方法必须在trio事件循环内部使用await调用。一个新的便利函数open_obs_websocket作为异步上下文管理器，提供ObsWs实例，自动连接并启动trio Nursery以管理websocket连接所需的后台任务，并处理来自OBS的事件。

打印所有场景名称的简单示例

from obswebsocket_trio import open_obs_websocket, requests
import trio

async def main(host: str = 'localhost', port: int = 4444, password: str = 'secret'):
    async with open_obs_websocket(host, port, password) as ws:
        scenes = await ws.call(requests.GetSceneList())
        for scene in scenes.getScenes():
            print(scene['name'])

trio.run(main)

或查看下面的文档

pydoc obswebsocket.core.ObsWS的输出

Help on class ObsWS in obswebsocket.core:

obswebsocket.core.ObsWS = class ObsWS(trio.abc.AsyncResource)
 |  obswebsocket.core.ObsWS(nursery: trio.Nursery, host='localhost', port=4444, password='')
 |
 |  Core class for using obs-websocket-py
 |
 |  Simple usage:
 |      >>> import obswebsocket_trio, obswebsocket_trio.requests as obsrequests
 |      >>> async with obswebsocket_trio.open_obs_websocket("localhost", 4444, "secret") as client:
 |      >>>     await client.call(obsrequests.GetVersion()).getObsWebsocketVersion()
 |      '4.1.0'
 |
 |  For advanced usage, including events callback, see the 'samples' directory.
 |
 |  Method resolution order:
 |      ObsWS
 |      trio.abc.AsyncResource
 |      builtins.object
 |
 |  Methods defined here:
 |
 |  __init__(self, nursery: trio.Nursery, host='localhost', port=4444, password='')
 |      Construct a new obsws wrapper
 |
 |      :param nursery: A trio Nursery to run background tasks
 |      :param host: Hostname to connect to
 |      :param port: TCP Port to connect to (Default is 4444)
 |      :param password: Password for the websocket server (Leave this field
 |          empty if no auth enabled on the server)
 |
 |  async aclose(self)
 |      Close this resource, possibly blocking.
 |
 |      IMPORTANT: This method may block in order to perform a "graceful"
 |      shutdown. But, if this fails, then it still *must* close any
 |      underlying resources before returning. An error from this method
 |      indicates a failure to achieve grace, *not* a failure to close the
 |      connection.
 |
 |      For example, suppose we call :meth:`aclose` on a TLS-encrypted
 |      connection. This requires sending a "goodbye" message; but if the peer
 |      has become non-responsive, then our attempt to send this message might
 |      block forever, and eventually time out and be cancelled. In this case
 |      the :meth:`aclose` method on :class:`~trio.SSLStream` will
 |      immediately close the underlying transport stream using
 |      :func:`trio.aclose_forcefully` before raising :exc:`~trio.Cancelled`.
 |
 |      If the resource is already closed, then this method should silently
 |      succeed.
 |
 |      Once this method completes, any other pending or future operations on
 |      this resource should generally raise :exc:`~trio.ClosedResourceError`,
 |      unless there's a good reason to do otherwise.
 |
 |      See also: :func:`trio.aclose_forcefully`.
 |
 |  async call(self, obj) -> obswebsocket.base_classes.Baserequests
 |      Make a call to the OBS server through the Websocket.
 |
 |      :param obj: Request (class from obswebsocket.requests module) to send
 |          to the server.
 |      :return: Request object populated with response data.
 |
 |  async connect(self, autoreconnect: bool = True)
 |      Connect to the websocket server
 |      :param autoreconnect: If True, tries to reconnect every 2 seconds if disconnected.
 |
 |      :return: Nothing
 |
 |  async disconnect(self)
 |      Disconnect from websocket server
 |
 |      :return: Nothing
 |
 |  async reconnect(self)
 |      Restart the connection to the websocket server
 |
 |      :return: Nothing
 |
 |  register(self, func, event=None)
 |      Register a new hook in the websocket client
 |
 |      :param func: Callback function pointer for the hook
 |      :param event: Event (class from obswebsocket.events module) to trigger
 |          the hook on. Default is None, which means trigger on all events.
 |      :return: Nothing
 |
 |  async send(self, data: dict) -> dict
 |      Make a raw json call to the OBS server through the Websocket.
 |
 |      :param data: Request (python dict) to send to the server. Do not
 |          include field "message-id".
 |      :return: Response (python dict) from the server.
 |
 |  unregister(self, func, event=None)
 |      Unregister a new hook in the websocket client
 |
 |      :param func: Callback function pointer for the hook
 |      :param event: Event (class from obswebsocket.events module) which
 |          triggered the hook on. Default is None, which means unregister this
 |          function for all events.
 |      :return: Nothing
 |
 |  ----------------------------------------------------------------------
 |  Static methods defined here:
 |
 |  build_event(data)
 |
 |  ----------------------------------------------------------------------
 |  Data descriptors defined here:
 |
 |  __dict__
 |      dictionary for instance variables (if defined)
 |
 |  __weakref__
 |      list of weak references to the object (if defined)
 |
 |  ----------------------------------------------------------------------
 |  Data and other attributes defined here:
 |
 |  __abstractmethods__ = frozenset()
 |
 |  ----------------------------------------------------------------------
 |  Methods inherited from trio.abc.AsyncResource:
 |
 |  async __aenter__(self)
 |
 |  async __aexit__(self, *args)