From 25bdf56aac46fc86c0771748680bad4f4ee8eca9 Mon Sep 17 00:00:00 2001 From: Jon Parise Date: Sun, 26 Apr 2026 20:18:16 -0400 Subject: [PATCH] Add CloudClient and deprecate ReadWriteClient Vestaboard renamed the Read / Write API to the Cloud API and added a Transitions API for controlling how messages animate onto the board. CloudClient supports reading and writing messages plus get_transition / set_transition. --- CHANGELOG.md | 12 +++++ docs/index.rst | 36 +++++++++---- src/vesta/__init__.py | 4 +- src/vesta/clients.py | 123 +++++++++++++++++++++++++++++++++++++++++- src/vesta/http.py | 2 +- tests/test_clients.py | 88 ++++++++++++++++++++++++++++++ 6 files changed, 252 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b694763..5762c48 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## Unreleased +### Added +- `CloudClient` provides a client interface to Vestaboard's Cloud API, which + supersedes the Read / Write API. It supports reading and writing messages + and the [Transitions API](https://docs.vestaboard.com/blog/transitions-api/) + for controlling how messages animate onto the board. + +### Changed +- Vestaboard has renamed the Read / Write API to the Cloud API, so our + `ReadWriteClient` interface is now considered deprecated. Switch to + `CloudClient`, which offers the same functionality plus transitions. + ## 0.13.0 - 2025-12-29 ### Added - Added support for Python 3.13 and 3.14. diff --git a/docs/index.rst b/docs/index.rst index 296ff6f..ab4128f 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -24,21 +24,22 @@ Vesta has no runtime dependencies and uses only the Python standard library. API Clients =========== -Read / Write API ----------------- +Cloud API +--------- -:py:class:`vesta.ReadWriteClient` provides a client interface for interacting -with a Vestaboard using the `Read / Write API -`_. +:py:class:`vesta.CloudClient` provides a client interface for interacting with +a Vestaboard using the `Cloud API +`_, which supersedes +the Read / Write API (:py:class:`vesta.ReadWriteClient`). It supports reading +and writing messages and configuring `message transitions +`_. .. important:: - A Read / Write API key is required to read or write messages. This key is - obtained by enabling the Vestaboard's Read / Write API via the *Settings* - section of the mobile app or from the `Developer section of the web app - `_. + A Vestaboard API token is required. This token can be obtained from the + `Developer section of the web app `_. -.. autoclass:: vesta.ReadWriteClient +.. autoclass:: vesta.CloudClient :members: Subscription API @@ -82,6 +83,21 @@ API. .. autoclass:: vesta.VBMLClient :members: +Read / Write API +---------------- + +:py:class:`vesta.ReadWriteClient` provides a client interface for interacting +with a Vestaboard using the **deprecated** Read / Write API. + +.. warning:: + + Vestaboard has renamed the Read / Write API to the Cloud API. This client + is **deprecated**; switch to :py:class:`vesta.CloudClient`, which offers + the same functionality plus message transitions. + +.. autoclass:: vesta.ReadWriteClient + :members: + Platform API ------------ diff --git a/src/vesta/__init__.py b/src/vesta/__init__.py index 1e0b91b..6e672e1 100644 --- a/src/vesta/__init__.py +++ b/src/vesta/__init__.py @@ -4,6 +4,7 @@ from .chars import encode_text from .chars import pprint from .clients import Client +from .clients import CloudClient from .clients import LocalClient from .clients import ReadWriteClient from .clients import SubscriptionClient @@ -12,6 +13,7 @@ __all__ = ( "Client", + "CloudClient", "Color", "HTTPError", "LocalClient", @@ -24,4 +26,4 @@ "pprint", ) -__version__ = "0.13.0" +__version__ = "0.14.0-dev" diff --git a/src/vesta/clients.py b/src/vesta/clients.py index 290bc9f..effb27b 100644 --- a/src/vesta/clients.py +++ b/src/vesta/clients.py @@ -25,6 +25,7 @@ from typing import Any from typing import Dict from typing import List +from typing import Literal from typing import Mapping from typing import Optional from typing import Union @@ -36,7 +37,14 @@ from .vbml import Component from .vbml import Props -__all__ = ["Client", "LocalClient", "ReadWriteClient", "VBMLClient"] +__all__ = [ + "Client", + "CloudClient", + "LocalClient", + "ReadWriteClient", + "SubscriptionClient", + "VBMLClient", +] class Client: @@ -223,6 +231,7 @@ class ReadWriteClient: (such as a custom `User-Agent` header). .. versionadded:: 0.8.0 + .. deprecated:: 0.14 """ def __init__( @@ -232,6 +241,12 @@ def __init__( base_url: str = "https://rw.vestaboard.com/", headers: Optional[Mapping[str, str]] = None, ): + warnings.warn( + "Vestaboard has renamed the Read / Write API to the Cloud API. " + "Consider using CloudClient instead.", + DeprecationWarning, + ) + all_headers = {"X-Vestaboard-Read-Write-Key": read_write_key} if headers: all_headers.update(headers) @@ -350,6 +365,112 @@ def send_message( return resp.json() +class CloudClient: + """Provides a Vestaboard Cloud API client interface. + + A Vestaboard API token is required. This token can be obtained from the + Developer section of the web app. + + The Cloud API supersedes the Read / Write API and additionally supports + the `Transitions API `_, + which controls how messages animate onto the board. + + Optionally, an alternate ``base_url`` can be specified, as well as any + additional HTTP ``headers`` that should be sent with every request + (such as a custom `User-Agent` header). + + .. versionadded:: 0.14.0 + """ + + def __init__( + self, + api_token: str, + *, + base_url: str = "https://cloud.vestaboard.com", + headers: Optional[Mapping[str, str]] = None, + ): + all_headers = {"X-Vestaboard-Token": api_token} + if headers: + all_headers.update(headers) + + self.http = http.Client(base_url=base_url, headers=all_headers) + + def __repr__(self): + return f'{type(self).__name__}(base_url="{self.http.base_url!s}")' + + def read_message(self) -> Optional[Rows]: + """Read the Vestaboard's current message.""" + resp = self.http.request("GET", "") + layout = resp.json({}).get("currentMessage", {}).get("layout") + if layout: + return json.loads(layout) + return None + + def write_message( + self, + message: Union[str, Rows], + *, + forced: bool = False, + ) -> bool: + """Write a message to the Vestaboard. + + `message` can be either a string of text or a two-dimensional (6, 22) + array of character codes representing the exact positions of characters + on the board. + + If text is specified, the lines will be centered horizontally and + vertically. Character codes will be inferred for alphanumeric and + punctuation characters, or they can be explicitly specified using curly + braces containing the character code (such as ``{5}`` or ``{65}``). + + Set ``forced`` to ``True`` to bypass the rate limit and force the + message to display immediately. + + :raises ValueError: if ``message`` is a list with unsupported dimensions + """ + data: Dict[str, Any] + if isinstance(message, str): + data = {"text": message} + elif isinstance(message, list): + validate_rows(message) + data = {"characters": message} + else: + raise TypeError(f"unsupported message type: {type(message)}") + + if forced: + data["forced"] = True + + resp = self.http.request("POST", "", json=data) + return 200 <= resp.status < 300 + + def get_transition(self) -> Dict[str, Any]: + """Get the current transition settings. + + :returns: A dict with ``transition`` and ``transitionSpeed`` keys. + """ + resp = self.http.request("GET", "/transition") + return resp.json({}) + + def set_transition( + self, + transition: Literal["classic", "wave", "drift", "curtain"], + speed: Literal["gentle", "fast"], + ) -> Dict[str, Any]: + """Update the transition settings. + + ``transition`` selects the animation style: ``"classic"``, ``"wave"``, + ``"drift"``, or ``"curtain"``. ``speed`` selects the animation speed: + ``"gentle"`` or ``"fast"``. + + Settings persist across subsequent messages until changed again. + + :returns: The updated settings as a dict. + """ + data = {"transition": transition, "transitionSpeed": speed} + resp = self.http.request("PUT", "/transition", json=data) + return resp.json({}) + + class VBMLClient: """Provides a VBML (Vestaboard Markup Language) API client interface. diff --git a/src/vesta/http.py b/src/vesta/http.py index 5bb43d2..8cb9ee8 100644 --- a/src/vesta/http.py +++ b/src/vesta/http.py @@ -69,7 +69,7 @@ def __init__( def request( self, - method: Literal["GET", "POST"], + method: Literal["GET", "POST", "PUT"], path: str, *, json: Optional[Any] = None, diff --git a/tests/test_clients.py b/tests/test_clients.py index 14f6b10..d5b0b0f 100644 --- a/tests/test_clients.py +++ b/tests/test_clients.py @@ -7,6 +7,7 @@ from vesta.chars import COLS from vesta.chars import ROWS from vesta.clients import Client +from vesta.clients import CloudClient from vesta.clients import LocalClient from vesta.clients import ReadWriteClient from vesta.clients import SubscriptionClient @@ -59,6 +60,11 @@ def subscription_client(): return SubscriptionClient("key", "secret") +@pytest.fixture +def cloud_client(): + return CloudClient("token") + + @pytest.fixture def vbml_client(): return VBMLClient() @@ -190,6 +196,7 @@ def test_write_message_dimensions(self, local_client: LocalClient): local_client.write_message([]) +@pytest.mark.filterwarnings("ignore:Vestaboard has renamed the Read / Write API") class TestReadWriteClient: def test_base_url(self): base_url = "http://example.local" @@ -291,6 +298,87 @@ def test_send_message_type(self, subscription_client: SubscriptionClient): subscription_client.send_message("sub_id", True) # type: ignore +class TestCloudClient: + def test_base_url(self): + base_url = "https://www.example.com" + client = CloudClient("token", base_url=base_url) + assert client.http.base_url == base_url + assert base_url in repr(client) + + def test_headers(self): + client = CloudClient("token", headers={"User-Agent": "Vesta"}) + assert client.http.headers["X-Vestaboard-Token"] == "token" + assert client.http.headers["User-Agent"] == "Vesta" + + def test_read_message(self, cloud_client: CloudClient, mock_response): + chars = [[0] * COLS] * ROWS + mock_response( + cloud_client, json={"currentMessage": {"layout": json_dumps(chars)}} + ) + message = cloud_client.read_message() + assert message == chars + + def test_read_message_empty(self, cloud_client: CloudClient, mock_response): + mock_response(cloud_client) + message = cloud_client.read_message() + assert message is None + + def test_read_message_empty_layout(self, cloud_client: CloudClient, mock_response): + mock_response(cloud_client, json={"currentMessage": {"layout": ""}}) + message = cloud_client.read_message() + assert message is None + + def test_write_message_text(self, cloud_client: CloudClient, mock_response): + text = "abc" + mock = mock_response(cloud_client) + assert cloud_client.write_message(text) + + assert mock["url"] == "" + assert mock["json"] == {"text": text} + + def test_write_message_list(self, cloud_client: CloudClient, mock_response): + chars = [[0] * COLS] * ROWS + mock = mock_response(cloud_client) + assert cloud_client.write_message(chars) + + assert mock["url"] == "" + assert mock["json"] == {"characters": chars} + + def test_write_message_forced(self, cloud_client: CloudClient, mock_response): + text = "abc" + mock = mock_response(cloud_client) + assert cloud_client.write_message(text, forced=True) + + assert mock["json"] == {"text": text, "forced": True} + + def test_write_message_list_dimensions(self, cloud_client: CloudClient): + with pytest.raises(ValueError, match=rf"expected a \({COLS}, {ROWS}\) array"): + cloud_client.write_message([]) + + def test_write_message_type(self, cloud_client: CloudClient): + with pytest.raises(TypeError, match=r"unsupported message type"): + cloud_client.write_message(True) # type: ignore + + def test_get_transition(self, cloud_client: CloudClient, mock_response): + settings = {"transition": "wave", "transitionSpeed": "gentle"} + mock = mock_response(cloud_client, json=settings) + assert cloud_client.get_transition() == settings + assert mock["method"] == "GET" + assert mock["url"] == "/transition" + + def test_get_transition_empty(self, cloud_client: CloudClient, mock_response): + mock_response(cloud_client) + assert cloud_client.get_transition() == {} + + def test_set_transition(self, cloud_client: CloudClient, mock_response): + settings = {"transition": "curtain", "transitionSpeed": "fast"} + mock = mock_response(cloud_client, json=settings) + assert cloud_client.set_transition("curtain", "fast") == settings + assert mock["method"] == "PUT" + assert mock["url"] == "/transition" + assert mock["json"] == settings + + class TestVBMLClient: def test_base_url(self): base_url = "http://example.local"