lib
/
flickerstrip-py
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added downloading of pattern

This commit is contained in:
Evan Fiordeliso 2023-01-29 20:59:50 -05:00
parent 0c86f25c9a
commit 93949ec0e9
7 changed files with 158 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.vscode/settings.json vendored
View File

@ -7,4 +7,6 @@
"python.linting.flake8Enabled": true,
"python.linting.enabled": true,
"python.pythonPath": ".venv/bin/python",
"python.analysis.typeCheckingMode": "basic",
"python.languageServer": "Pylance",
}

3
flickerstrip_py/discovery.py
View File

@ -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()

181
flickerstrip_py/flickerstrip.py
View File

@ -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)

26
flickerstrip_py/pattern.py
View File

@ -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

9
flickerstrip_py/status.py
View File

@ -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,

4
pyproject.toml
View File

@ -1,11 +1,11 @@
[tool.poetry]
name = "flickerstrip-py"
version = "0.2.1"
version = "0.2.2"
description = ""
authors = ["Evan Fiordeliso <evan.fiordeliso@gmail.com>"]
[tool.poetry.dependencies]
python = ">=3.10 <3.11"
python = "~3.10"
aiohttp = "3.8.1"
async-upnp-client = "^0.33.0"

2
setup.py
View File

@ -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 <evan.fiordeliso@gmail.com>',
license='MIT',