Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ addopts = "--cov=vesta --cov-report=term-missing --doctest-modules"
src = ["src", "tests"]

[tool.ruff.lint]
select = ["E", "F", "I", "RUF", "W"]
select = ["E", "F", "I", "RUF", "UP", "W"]

[tool.ruff.lint.isort]
force-single-line = true
Expand Down
18 changes: 6 additions & 12 deletions src/vesta/chars.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,13 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from __future__ import annotations

import enum
import math
import sys
from typing import Container
from collections.abc import Container
from typing import Final
from typing import List
from typing import Literal
from typing import Optional
from typing import TextIO
from typing import Tuple
from typing import Union
from typing import cast

PRINTABLE = " ABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890!@#$() - +&=;: '\"%,. /? °"
Expand All @@ -43,10 +37,10 @@
ROWS: Final[int] = 6

#: A row of character codes.
Row = List[int]
Row = list[int]

#: A list of rows, forming a character grid.
Rows = List[Row]
Rows = list[Row]


def validate_rows(rows: Rows) -> None:
Expand Down Expand Up @@ -159,7 +153,7 @@ def encode_text(
s: str,
*,
align: Literal["left", "center", "right"] = "left",
valign: Optional[Literal["top", "middle", "bottom"]] = "top",
valign: Literal["top", "middle", "bottom"] | None = "top",
max_rows: int = ROWS,
margin: int = 0,
fill: int = Color.BLANK,
Expand Down Expand Up @@ -211,7 +205,7 @@ def encode_text(
# Find a place to break the line at a `breaks` character that results in
# the longest row possible (up to `max_cols`). The result is a pair of
# positions: the end of the resulting row and the start of the remainder.
def find_break(line: Row) -> Tuple[int, int]:
def find_break(line: Row) -> tuple[int, int]:
end = min(len(line), max_cols)
for pos in range(end, 0, -1):
if line[pos] in breaks:
Expand Down Expand Up @@ -266,7 +260,7 @@ def _format_row(row: Row, align: str, margin: int, fill: int) -> Row:


def pprint(
data: Union[Row, Rows],
data: Row | Rows,
stream: TextIO = sys.stdout,
*,
sep: str = "|",
Expand Down
68 changes: 31 additions & 37 deletions src/vesta/clients.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,17 +18,11 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from __future__ import annotations

import json
import warnings
from collections.abc import Mapping
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
from typing import cast

from . import http
Expand Down Expand Up @@ -65,7 +59,7 @@ def __init__(
api_secret: str,
*,
base_url: str = "https://platform.vestaboard.com",
headers: Optional[Mapping[str, str]] = None,
headers: Mapping[str, str] | None = None,
):
warnings.warn(
"Vestaboard has deprecated the Platform API (Client). "
Expand All @@ -85,21 +79,21 @@ def __init__(
def __repr__(self):
return f'{type(self).__name__}(base_url="{self.http.base_url!s}")'

def get_subscriptions(self) -> List[Dict[str, Any]]:
def get_subscriptions(self) -> list[dict[str, Any]]:
"""Lists all subscriptions to which the viewer has access."""
resp = self.http.request("GET", "/subscriptions")
return resp.json({}).get("subscriptions", [])

def get_viewer(self) -> Dict[str, Any]:
def get_viewer(self) -> dict[str, Any]:
"""Describes the currently authenticated viewer."""
resp = self.http.request("GET", "/viewer")
return resp.json()

def post_message(
self,
subscription_id: str,
message: Union[str, Rows],
) -> Dict[str, Any]:
message: str | Rows,
) -> dict[str, Any]:
"""Post a new message to a subscription.

The authenticated viewer must have access to the subscription.
Expand All @@ -115,7 +109,7 @@ def post_message(

:raises ValueError: if ``message`` is a list with unsupported dimensions
"""
data: Dict[str, Union[str, Rows]]
data: dict[str, str | Rows]
if isinstance(message, str):
data = {"text": message}
elif isinstance(message, list):
Expand Down Expand Up @@ -150,7 +144,7 @@ class LocalClient:

def __init__(
self,
local_api_key: Optional[str] = None,
local_api_key: str | None = None,
*,
base_url: str = "http://vestaboard.local:7000",
):
Expand All @@ -162,10 +156,10 @@ def __repr__(self):
return f'{type(self).__name__}(base_url="{self.http.base_url!s}")'

@property
def api_key(self) -> Optional[str]:
def api_key(self) -> str | None:
"""The client's Local API key."""
return cast(
Optional[str],
str | None,
self.http.headers.get("X-Vestaboard-Local-Api-Key"),
)

Expand All @@ -179,7 +173,7 @@ def enabled(self) -> bool:
support has been enabled."""
return self.api_key is not None

def enable(self, enablement_token) -> Optional[str]:
def enable(self, enablement_token) -> str | None:
"""Enable the Vestaboard's Local API using a Local API Enablement Token.

If successful, the Vestaboard's Local API key will be returned and the
Expand All @@ -197,7 +191,7 @@ def enable(self, enablement_token) -> Optional[str]:

return local_api_key

def read_message(self) -> Optional[Rows]:
def read_message(self) -> Rows | None:
"""Read the Vestaboard's current message."""
if not self.enabled:
raise RuntimeError("Local API has not been enabled")
Expand Down Expand Up @@ -239,7 +233,7 @@ def __init__(
read_write_key: str,
*,
base_url: str = "https://rw.vestaboard.com/",
headers: Optional[Mapping[str, str]] = None,
headers: Mapping[str, str] | None = None,
):
warnings.warn(
"Vestaboard has renamed the Read / Write API to the Cloud API. "
Expand All @@ -256,15 +250,15 @@ def __init__(
def __repr__(self):
return f'{type(self).__name__}(base_url="{self.http.base_url!s}")'

def read_message(self) -> Optional[Rows]:
def read_message(self) -> Rows | None:
"""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]) -> bool:
def write_message(self, message: str | Rows) -> bool:
"""Write a message to the Vestaboard.

`message` can be either a string of text or a two-dimensional (6, 22)
Expand All @@ -278,7 +272,7 @@ def write_message(self, message: Union[str, Rows]) -> bool:

:raises ValueError: if ``message`` is a list with unsupported dimensions
"""
data: Union[Dict[str, str], Rows]
data: dict[str, str] | Rows
if isinstance(message, str):
data = {"text": message}
elif isinstance(message, list):
Expand Down Expand Up @@ -309,7 +303,7 @@ def __init__(
api_secret: str,
*,
base_url: str = "https://subscriptions.vestaboard.com",
headers: Optional[Mapping[str, str]] = None,
headers: Mapping[str, str] | None = None,
):
all_headers = {
"x-vestaboard-api-key": api_key,
Expand All @@ -323,16 +317,16 @@ def __init__(
def __repr__(self):
return f'{type(self).__name__}(base_url="{self.http.base_url!s}")'

def get_subscriptions(self) -> List[Dict[str, Any]]:
def get_subscriptions(self) -> list[dict[str, Any]]:
"""Lists all subscriptions to which the viewer has access."""
resp = self.http.request("GET", "/subscriptions")
return resp.json()

def send_message(
self,
subscription_id: str,
message: Union[str, Rows],
) -> Dict[str, Any]:
message: str | Rows,
) -> dict[str, Any]:
"""Send a new message to a subscription.

The authenticated viewer must have access to the subscription.
Expand All @@ -348,7 +342,7 @@ def send_message(

:raises ValueError: if ``message`` is a list with unsupported dimensions
"""
data: Dict[str, Union[str, Rows]]
data: dict[str, str | Rows]
if isinstance(message, str):
data = {"text": message}
elif isinstance(message, list):
Expand Down Expand Up @@ -387,7 +381,7 @@ def __init__(
api_token: str,
*,
base_url: str = "https://cloud.vestaboard.com",
headers: Optional[Mapping[str, str]] = None,
headers: Mapping[str, str] | None = None,
):
all_headers = {"X-Vestaboard-Token": api_token}
if headers:
Expand All @@ -398,7 +392,7 @@ def __init__(
def __repr__(self):
return f'{type(self).__name__}(base_url="{self.http.base_url!s}")'

def read_message(self) -> Optional[Rows]:
def read_message(self) -> Rows | None:
"""Read the Vestaboard's current message."""
resp = self.http.request("GET", "")
layout = resp.json({}).get("currentMessage", {}).get("layout")
Expand All @@ -408,7 +402,7 @@ def read_message(self) -> Optional[Rows]:

def write_message(
self,
message: Union[str, Rows],
message: str | Rows,
*,
forced: bool = False,
) -> bool:
Expand All @@ -428,7 +422,7 @@ def write_message(

:raises ValueError: if ``message`` is a list with unsupported dimensions
"""
data: Dict[str, Any]
data: dict[str, Any]
if isinstance(message, str):
data = {"text": message}
elif isinstance(message, list):
Expand All @@ -443,7 +437,7 @@ def write_message(
resp = self.http.request("POST", "", json=data)
return 200 <= resp.status < 300

def get_transition(self) -> Dict[str, Any]:
def get_transition(self) -> dict[str, Any]:
"""Get the current transition settings.

:returns: A dict with ``transition`` and ``transitionSpeed`` keys.
Expand All @@ -455,7 +449,7 @@ def set_transition(
self,
transition: Literal["classic", "wave", "drift", "curtain"],
speed: Literal["gentle", "fast"],
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Update the transition settings.

``transition`` selects the animation style: ``"classic"``, ``"wave"``,
Expand All @@ -481,7 +475,7 @@ def __init__(
self,
*,
base_url: str = "https://vbml.vestaboard.com/",
headers: Optional[Mapping[str, str]] = None,
headers: Mapping[str, str] | None = None,
):
self.http = http.Client(base_url=base_url, headers=headers)

Expand All @@ -490,8 +484,8 @@ def __repr__(self):

def compose(
self,
components: List[Component],
props: Optional[Props] = None,
components: list[Component],
props: Props | None = None,
) -> Rows:
"""Composes one or more styled components into rows of character codes.

Expand All @@ -505,7 +499,7 @@ def compose(
if not components:
raise ValueError("expected at least one component")

data: Dict[str, Union[Props, List[Dict]]] = {
data: dict[str, Props | list[dict]] = {
"components": [component.asdict() for component in components],
}
if props:
Expand Down
16 changes: 6 additions & 10 deletions src/vesta/http.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,15 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from __future__ import annotations

import urllib.error
import urllib.request
from collections.abc import Mapping
from json import JSONDecodeError
from json import dumps as json_dumps
from json import loads as json_loads
from typing import Any
from typing import Dict
from typing import Literal
from typing import Mapping
from typing import NamedTuple
from typing import Optional
from urllib.parse import urljoin
from urllib.parse import urlsplit

Expand All @@ -43,7 +39,7 @@ class Response(NamedTuple):

status: int
body: bytes
headers: Dict[str, str]
headers: dict[str, str]

def json(self, default: Any = None):
try:
Expand All @@ -61,19 +57,19 @@ class Client:
def __init__(
self,
base_url: str = "",
headers: Optional[Mapping[str, str]] = None,
headers: Mapping[str, str] | None = None,
):
self.base_url = base_url
self.headers: Dict[str, str] = dict(headers) if headers else {}
self.headers: dict[str, str] = dict(headers) if headers else {}
self._opener = urllib.request.build_opener()

def request(
self,
method: Literal["GET", "POST", "PUT"],
path: str,
*,
json: Optional[Any] = None,
headers: Optional[Mapping[str, str]] = None,
json: Any | None = None,
headers: Mapping[str, str] | None = None,
) -> Response:
"""Execute HTTP request and return Response.

Expand Down
Loading
Loading