From 93949ec0e982381320437c842f7664e2a6a44fd8 Mon Sep 17 00:00:00 2001 From: Evan Fiordeliso Date: Sun, 29 Jan 2023 20:59:50 -0500 Subject: [PATCH] Added downloading of pattern --- .vscode/settings.json | 2 + flickerstrip_py/discovery.py | 3 +- flickerstrip_py/flickerstrip.py | 181 +++++++++++++++++++++----------- flickerstrip_py/pattern.py | 26 ++++- flickerstrip_py/status.py | 9 ++ pyproject.toml | 4 +- setup.py | 2 +- 7 files changed, 158 insertions(+), 69 deletions(-) diff --git a/.vscode/settings.json b/.vscode/settings.json index e265b41..f73590f 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -7,4 +7,6 @@ "python.linting.flake8Enabled": true, "python.linting.enabled": true, "python.pythonPath": ".venv/bin/python", + "python.analysis.typeCheckingMode": "basic", + "python.languageServer": "Pylance", } \ No newline at end of file diff --git a/flickerstrip_py/discovery.py b/flickerstrip_py/discovery.py index 802917b..054b1f8 100644 --- a/flickerstrip_py/discovery.py +++ b/flickerstrip_py/discovery.py @@ -43,7 +43,8 @@ async def main(): flickerstrips = await client.discover() if len(flickerstrips) > 0: await flickerstrips[0].force_update() - print(json.dumps(flickerstrips[0].status.to_json())) + if flickerstrips[0].status is not None: + print(json.dumps(flickerstrips[0].status.to_json())) if __name__ == "__main__": loop = asyncio.get_event_loop() diff --git a/flickerstrip_py/flickerstrip.py b/flickerstrip_py/flickerstrip.py index 940f284..c3e554e 100644 --- a/flickerstrip_py/flickerstrip.py +++ b/flickerstrip_py/flickerstrip.py @@ -1,11 +1,12 @@ -from typing import Any, Optional, Mapping import aiohttp +import json +from typing import Any, Optional, Mapping from .status import Status from .pattern import PatternMeta, PatternBuilder class Flickerstrip: - def __init__(self, host: str, port=80): + def __init__(self, host: str, port: int = 80): """The initializer for the Flickerstrip class. Args: @@ -13,11 +14,11 @@ class Flickerstrip: port (int, optional): The port of the flickerstrip. Defaults to port 80. """ - self.host = host - self.port = port - self.status = None + self.host: str = host + self.port: int = port + self.status: Status | None = None - async def __check_response(self, response): + async def __handle_response(self, response): """Check if the request was succesful. Args: @@ -26,7 +27,13 @@ class Flickerstrip: Returns: bool: If the request was successful """ - data = await response.json() + body = await response.text() + + if body == "Not Found": + print(f"ERROR: Request failed with message \"{body}\".") + return False + + data = json.loads(body) if "type" not in data: if "id" in data: # create pattern response. @@ -36,22 +43,24 @@ class Flickerstrip: return False respType = data["type"] - if respType == "error": - message = data["message"] - print(f"ERROR: Request failed with message \"{message}\".") - return False - elif respType == "OK": - return True - elif respType == "status": - self.status = Status.from_json(data) - return True - else: - print(f"ERROR: Unrecognized response type: {respType}.") - print(data) - return False + match respType: + case "error": + message = data["message"] + print(f"ERROR: Request failed with message \"{message}\".") + return False + case "OK": + return True + case "status": + self.status = Status.from_json(data) + return True + case _: + print(f"ERROR: Unrecognized response type: {respType}.") + print(data) + return False async def __get_request(self, path: str, - params: Optional[Mapping[str, str]] = None): + params: Optional[Mapping[str, str]] = None + ) -> aiohttp.ClientResponse: """Send a GET request to the strip. Args: @@ -60,17 +69,18 @@ class Flickerstrip: request. Defaults to None. Returns: - bool: If the request was successful + ClientResponse: the client response """ async with aiohttp.ClientSession() as session: async with session.get(f"http://{self.host}:{self.port}/{path}", params=params) as response: - return await self.__check_response(response) + return response async def __post_request(self, path, params: Optional[Mapping[str, str]] = None, json: Any = None, - data: Any = None): + data: Any = None + ) -> aiohttp.ClientResponse: """Send a POST request to the strip. Args: @@ -82,13 +92,13 @@ class Flickerstrip: data (string, optional): The raw data body. Defaults to None. Returns: - bool: If the request was successful + ClientResponse: the client response """ async with aiohttp.ClientSession() as session: async with session.post(f"http://{self.host}:{self.port}/{path}", params=params, json=json, data=data ) as response: - return await self.__check_response(response) + return response async def force_update(self): """Force updates the status of the strip. @@ -96,7 +106,8 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("status") + resp = await self.__get_request("status") + return await self.__handle_response(resp) async def power_on(self): """Trigger the strip to turn the lights on. @@ -104,7 +115,8 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("power/on") + resp = await self.__get_request("power/on") + return await self.__handle_response(resp) async def power_off(self): """Trigger the strip to turn the lights off. @@ -112,7 +124,8 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("power/off") + resp = await self.__get_request("power/off") + return await self.__handle_response(resp) async def power_toggle(self): """Trigger the strip to toggle the power status. @@ -120,7 +133,8 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("power/toggle") + resp = await self.__get_request("power/toggle") + return await self.__handle_response(resp) async def next_pattern(self): """Trigger the strip to switch to the next pattern in memory. @@ -128,7 +142,8 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("pattern/next") + resp = await self.__get_request("pattern/next") + return await self.__handle_response(resp) async def freeze_frame(self, frame: int): """Freeze the animation at the specified frame. @@ -139,8 +154,33 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("pattern/frame", - params={"value": frame}) + params = {"value": str(frame)} + resp = await self.__get_request("pattern/frame", params=params) + return await self.__handle_response(resp) + + async def download_pattern(self, id: int) -> PatternBuilder | None: + """Download a pattern by its id + + Args: + id (int): The ID of the pattern + + Returns: + PatternBuilder: a pattern builder with the data downloaded + """ + + if self.status is None: + return None + meta = self.status.get_pattern_by_id(id) + if meta is None: + return None + + params = {"id": str(id)} + resp = await self.__get_request("pattern/download", params=params) + if resp.status != 200: + return None + + data = await resp.read() + return PatternBuilder.from_binary_string(data, meta) async def select_pattern(self, index: int): """Change the current pattern to the pattern at the specified index. @@ -151,19 +191,20 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("pattern/select", - params={"index": index}) + params = {"index": str(index)} + resp = await self.__get_request("pattern/select", params=params) + return await self.__handle_response(resp) async def delete_pattern(self, pattern: Optional[PatternMeta] = None, index: Optional[int] = None, - id: Optional[str] = None): + id: Optional[int] = None): """Delete a pattern by its meta, index, or id Args: pattern (PatternMeta, optional): The pattern meta. index (int, optional): The index of the pattern. - id (string, optional): The id of the pattern. + id (int, optional): The id of the pattern. Returns: bool: If the request was succesful @@ -174,11 +215,13 @@ class Flickerstrip: "Expected a PatternMeta object for the pattern arg.") return await self.delete_pattern(id=pattern.id) elif index is not None: - return await self.__get_request("pattern/forget", - params={"index": index}) + params = {"index": str(index)} + resp = await self.__get_request("pattern/forget", params=params) + return await self.__handle_response(resp) elif id is not None: - return await self.__get_request("pattern/forget", - params={"id": id}) + params = {"id": str(id)} + resp = await self.__get_request("pattern/forget", params=params) + return await self.__handle_response(resp) else: raise TypeError( "Deleting a pattern requires one of the three args.") @@ -194,13 +237,17 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__post_request("pattern/create", params={ + params = { "name": pattern.name, - "fps": pattern.fps, - "frames": pattern.frame_count, - "pixels": pattern.pixel_count, - "preview": preview - }, data=pattern.to_binary_string()) + "fps": str(pattern.fps), + "frames": str(pattern.frame_count), + "pixels": str(pattern.pixel_count), + "preview": str(1 if preview else 0) + } + data = pattern.to_binary_string() + resp = await self.__post_request("pattern/create", + params=params, data=data) + return await self.__handle_response(resp) async def set_color(self, r: int, g: int, b: int): """Sets the strip to a solid color (given as an RGB value). @@ -226,7 +273,9 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("brightness", params={"value": value}) + params = {"value": str(value)} + resp = await self.__get_request("brightness", params=params) + return await self.__handle_response(resp) async def set_name(self, value: str): """Set the name of the flickerstrip. @@ -237,7 +286,9 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__post_request("config/name", json={"name": value}) + json = {"name": value} + resp = await self.__post_request("config/name", json=json) + return await self.__handle_response(resp) async def set_group(self, value: str): """Set the group of the flickerstrip. @@ -248,7 +299,9 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__post_request("config/group", json={"name": value}) + json = {"name": value} + resp = await self.__post_request("config/group", json=json) + return await self.__handle_response(resp) async def set_cycle(self, value): """Set the cycle timer of the flickerstrip. @@ -260,8 +313,9 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("config/cycle", - params={"value": value}) + params = {"value": value} + resp = await self.__get_request("config/cycle", params=params) + return await self.__handle_response(resp) async def set_fade_duration(self, value: int): """Set the fade timer of the flickerstrip. @@ -273,7 +327,9 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("config/fade", params={"value": value}) + params = {"value": str(value)} + resp = await self.__get_request("config/fade", params=params) + return await self.__handle_response(resp) async def set_strip_length(self, value: int): """Set the length of the flickerstrip. @@ -284,8 +340,9 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("config/length", - params={"value": value}) + params = {"value": str(value)} + resp = await self.__get_request("config/length", params=params) + return await self.__handle_response(resp) async def set_strip_start(self, value: int): """Set the start pixel of the strip. @@ -299,8 +356,9 @@ class Flickerstrip: Returns: bool: If the request was successful. """ - return await self.__get_request("config/start", - params={"value": value}) + params = {"value": str(value)} + resp = await self.__get_request("config/start", params=params) + return await self.__handle_response(resp) async def set_strip_end(self, value: int): """Set the end pixel of the strip. @@ -314,7 +372,9 @@ class Flickerstrip: Returns: bool: If the request was successful. """ - return await self.__get_request("config/end", params={"value": value}) + params = {"value": str(value)} + resp = await self.__get_request("config/end", params=params) + return await self.__handle_response(resp) async def set_reversed(self, value: bool): """Set the reversed state of the flickerstrip. @@ -326,5 +386,6 @@ class Flickerstrip: Returns: bool: If the request was succesful """ - return await self.__get_request("config/reversed", - params={"value": 1 if value else 0}) + params = {"value": str(1 if value else 0)} + resp = await self.__get_request("config/reversed", params=params) + return await self.__handle_response(resp) diff --git a/flickerstrip_py/pattern.py b/flickerstrip_py/pattern.py index 2eae6e5..c153bd3 100644 --- a/flickerstrip_py/pattern.py +++ b/flickerstrip_py/pattern.py @@ -1,4 +1,4 @@ -from typing import Any +from typing import Any, Tuple class PatternMeta: @@ -27,7 +27,7 @@ class PatternMeta: class PatternBuilder: def __init__(self, name: str, fps=1): - self.pixels: list[list[int]] = [] + self.data: list[int] = [] self.name: str = name self.fps: int = fps self.frame_count: int = 1 @@ -39,14 +39,30 @@ class PatternBuilder: self.pixel_count += 1 self.frame_pixels += 1 - self.pixels += [r, g, b] + self.data += [r, g, b] + + def get_frame(self) -> list[Tuple[int, int, int]]: + return [ + (self.data[i], self.data[i + 1], self.data[i + 2]) + for i in range(0, len(self.data), 3) + ] def next_frame(self): self.frame_count += 1 self.frame_pixels = 0 def is_valid(self): - self.frame_pixels == self.pixel_count + return self.frame_pixels == self.pixel_count def to_binary_string(self) -> str: - return ''.join([chr(item) for item in self.pixels]) + return ''.join([chr(item) for item in self.data]) + + @classmethod + def from_binary_string(cls, data: bytes, meta: PatternMeta): + builder = PatternBuilder(meta.name, meta.fps) + builder.data = [c for c in data] + builder.fps = meta.fps + builder.frame_count = meta.frames + builder.pixel_count = meta.pixels + builder.frame_pixels = int(meta.pixels / meta.frames) + return builder diff --git a/flickerstrip_py/status.py b/flickerstrip_py/status.py index 30bd45e..d64a6d1 100644 --- a/flickerstrip_py/status.py +++ b/flickerstrip_py/status.py @@ -42,6 +42,15 @@ class Status: self.memory: MemoryUsage = memory self.patterns: list[PatternMeta] = patterns + def get_current_pattern(self) -> PatternMeta: + return self.patterns[self.selectedPattern] + + def get_pattern_by_id(self, id: int) -> PatternMeta | None: + for pattern in self.patterns: + if pattern.id == id: + return pattern + return None + def to_json(self): return { "ap": self.ap, "name": self.name, "group": self.group, diff --git a/pyproject.toml b/pyproject.toml index fec73c6..6d15dd8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [tool.poetry] name = "flickerstrip-py" -version = "0.2.1" +version = "0.2.2" description = "" authors = ["Evan Fiordeliso "] [tool.poetry.dependencies] -python = ">=3.10 <3.11" +python = "~3.10" aiohttp = "3.8.1" async-upnp-client = "^0.33.0" diff --git a/setup.py b/setup.py index 535e1cf..53086a3 100644 --- a/setup.py +++ b/setup.py @@ -3,7 +3,7 @@ from setuptools import find_packages, setup setup( name='flickerstrip-py', packages=find_packages(), - version='0.1.0', + version='0.2.2', description='A python library for interracting with a flickerstrip.', author='Evan Fiordeliso ', license='MIT',