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

Finish discovery method and rename all methods

This commit is contained in:
Evan Fiordeliso 2021-12-18 20:52:51 -05:00
parent fa438c8266
commit 5665d57191
6 changed files with 104 additions and 135 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.vscode/settings.json vendored
View File

@ -4,7 +4,6 @@
], ],
"python.testing.unittestEnabled": false, "python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true, "python.testing.pytestEnabled": true,
"python.pythonPath": "/home/evanf/.cache/pypoetry/virtualenvs/flickerstrip-py-zxDi65bL-py3.9/bin/python",
"python.linting.flake8Enabled": true, "python.linting.flake8Enabled": true,
"python.linting.enabled": true "python.linting.enabled": true
} }

23
flickerstrip_py/discovery.py
View File

@ -1,18 +1,31 @@
from ssdpy import SSDPClient from ssdpy import SSDPClient
import re
import json
from flickerstrip import Flickerstrip
class FlickerstripDiscoveryClient: class FlickerstripDiscoveryClient:
def __init__(self): def __init__(self):
self.client = SSDPClient() self.client = SSDPClient(timeout=10)
def __discovered(self, device):
loc = device["location"]
result = re.search("http://(.*):80/description.xml", loc)
ip_address = result.group(1)
return Flickerstrip(ip_address)
def discover(self): def discover(self):
print("Discovering devices...") print("Discovering devices...")
devices = self.client.m_search("ssdp:all") devices = self.client.m_search("ssdp:all")
print(f"Discovered {len(devices)} devices.") print(f"Discovered {len(devices)} device(s).")
for device in devices: filtered = [d for d in devices if "Flickerstrip" in d["server"]]
print(device) print(f"Discovered {len(filtered)} flickerstrip(s).")
mapped = [self.__discovered(d) for d in filtered]
return mapped
if __name__ == "__main__": if __name__ == "__main__":
client = FlickerstripDiscoveryClient() client = FlickerstripDiscoveryClient()
client.discover() flickerstrips = client.discover()
flickerstrips[0].force_update()
print(json.dumps(flickerstrips[0].status.to_json()))

96
flickerstrip_py/flickerstrip.py
View File

@ -1,6 +1,6 @@
import requests import requests
from pattern import PatternMeta
from status import Status from status import Status
from pattern import PatternMeta
class Flickerstrip: class Flickerstrip:
@ -13,7 +13,7 @@ class Flickerstrip:
self.ip_address = ip_address self.ip_address = ip_address
self.status = None self.status = None
def __checkResponse(self, response): def __check_response(self, response):
"""Check if the request was succesful. """Check if the request was succesful.
Args: Args:
@ -22,22 +22,22 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was successful boolean: If the request was successful
""" """
json = response.json() data = response.json()
respType = json["type"] respType = data["type"]
if respType == "error": if respType == "error":
message = json["message"] message = data["message"]
print(f"ERROR: Request failed with message \"{message}\".") print(f"ERROR: Request failed with message \"{message}\".")
return False return False
elif respType == "OK": elif respType == "OK":
return True return True
elif respType == "status": elif respType == "status":
self.status = Status.fromJSON(json) self.status = Status.from_json(data)
return True return True
else: else:
print(f"ERROR: Unrecognized response type: {respType}.") print(f"ERROR: Unrecognized response type: {respType}.")
return False return False
def __getRequest(self, path, params=None): def __get_request(self, path, params=None):
"""Send a GET request to the strip. """Send a GET request to the strip.
Args: Args:
@ -49,9 +49,9 @@ class Flickerstrip:
boolean: If the request was successful boolean: If the request was successful
""" """
resp = requests.get(f"http://{self.ip_address}/{path}", params=params) resp = requests.get(f"http://{self.ip_address}/{path}", params=params)
return self.__checkResponse(resp) return self.__check_response(resp)
def __postRequest(self, path, json=None, data=None): def __post_request(self, path, json=None, data=None):
"""Send a POST request to the strip. """Send a POST request to the strip.
Args: Args:
@ -65,49 +65,49 @@ class Flickerstrip:
""" """
resp = requests.post(f"http://{self.ip_address}/{path}", resp = requests.post(f"http://{self.ip_address}/{path}",
json=json, data=data) json=json, data=data)
return self.__checkResponse(resp) return self.__check_response(resp)
def forceUpdate(self): def force_update(self):
"""Force updates the status of the strip. """Force updates the status of the strip.
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("status") return self.__get_request("status")
def powerOn(self): def power_on(self):
"""Trigger the strip to turn the lights on. """Trigger the strip to turn the lights on.
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("power/on") return self.__get_request("power/on")
def powerOff(self): def power_off(self):
"""Trigger the strip to turn the lights off. """Trigger the strip to turn the lights off.
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("power/off") return self.__get_request("power/off")
def powerToggle(self): def power_toggle(self):
"""Trigger the strip to toggle the power status. """Trigger the strip to toggle the power status.
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("power/toggle") return self.__get_request("power/toggle")
def nextPattern(self): def next_pattern(self):
"""Trigger the strip to switch to the next pattern in memory. """Trigger the strip to switch to the next pattern in memory.
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("pattern/next") return self.__get_request("pattern/next")
def freezeFrame(self, frame): def freeze_frame(self, frame):
"""Freeze the animation at the specified frame. """Freeze the animation at the specified frame.
Args: Args:
@ -116,9 +116,9 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("pattern/frame", {"value": frame}) return self.__get_request("pattern/frame", {"value": frame})
def setPattern(self, index): def select_pattern(self, index):
"""Change the current pattern to the pattern at the specified index. """Change the current pattern to the pattern at the specified index.
Args: Args:
@ -127,23 +127,23 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("pattern/select", {"index": index}) return self.__get_request("pattern/select", {"index": index})
def deletePattern(self, pattern=None, index=None, id=None): def delete_pattern(self, pattern=None, index=None, id=None):
if pattern is not None: if pattern is not None:
if not isinstance(pattern, PatternMeta): if not isinstance(pattern, PatternMeta):
raise TypeError( raise TypeError(
"Expected a PatternMeta object for the pattern arg.") "Expected a PatternMeta object for the pattern arg.")
return self.deletePattern(id=pattern.id) return self.delete_pattern(id=pattern.id)
elif index is not None: elif index is not None:
return self.__getRequest("pattern/forget", {"index": index}) return self.__get_request("pattern/forget", {"index": index})
elif id is not None: elif id is not None:
return self.__getRequest("pattern/forget", {"id": id}) return self.__get_request("pattern/forget", {"id": id})
else: else:
raise TypeError( raise TypeError(
"Deleting a pattern requires one of the three args.") "Deleting a pattern requires one of the three args.")
def setBrightness(self, value): def set_brightness(self, value):
"""Set the brightness of the flickerstrip. """Set the brightness of the flickerstrip.
Args: Args:
@ -152,9 +152,9 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("brightness", {"value": value}) return self.__get_request("brightness", {"value": value})
def setName(self, value): def set_name(self, value):
"""Set the name of the flickerstrip. """Set the name of the flickerstrip.
Args: Args:
@ -163,9 +163,9 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__postRequest("config/name", {"name": value}) return self.__post_request("config/name", {"name": value})
def setGroup(self, value): def set_group(self, value):
"""Set the group of the flickerstrip. """Set the group of the flickerstrip.
Args: Args:
@ -174,9 +174,9 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__postRequest("config/group", {"name": value}) return self.__post_request("config/group", {"name": value})
def setCycle(self, value): def set_cycle(self, value):
"""Set the cycle timer of the flickerstrip. """Set the cycle timer of the flickerstrip.
Args: Args:
@ -186,9 +186,9 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("config/cycle", {"value": value}) return self.__get_request("config/cycle", {"value": value})
def setFadeDuration(self, value): def set_fade_duration(self, value):
"""Set the fade timer of the flickerstrip. """Set the fade timer of the flickerstrip.
Args: Args:
@ -198,20 +198,20 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("config/fade", {"value": value}) return self.__get_request("config/fade", {"value": value})
def setStripLength(self, value): def set_strip_length(self, value):
"""Set the length of the flickerstrip. """Set the length of the flickerstrip.
Args: Args:
value (int): The length of the strip. value (int): The length of the strip in pixels.
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("config/length", {"value": value}) return self.__get_request("config/length", {"value": value})
def setStripStart(self, value): def set_strip_start(self, value):
"""Set the start pixel of the strip. """Set the start pixel of the strip.
This will make the flickerstrip ignore all pixels before This will make the flickerstrip ignore all pixels before
@ -223,9 +223,9 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was successful. boolean: If the request was successful.
""" """
return self.__getRequest("config/start", {"value": value}) return self.__get_request("config/start", {"value": value})
def setStripEnd(self, value): def set_strip_end(self, value):
"""Set the end pixel of the strip. """Set the end pixel of the strip.
This will make the flickerstrip ignore all pixels after This will make the flickerstrip ignore all pixels after
@ -237,9 +237,9 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was successful. boolean: If the request was successful.
""" """
return self.__getRequest("config/end", {"value": value}) return self.__get_request("config/end", {"value": value})
def setReversed(self, value): def set_reversed(self, value):
"""Set the reversed state of the flickerstrip. """Set the reversed state of the flickerstrip.
Args: Args:
@ -249,5 +249,5 @@ class Flickerstrip:
Returns: Returns:
boolean: If the request was succesful boolean: If the request was succesful
""" """
return self.__getRequest("config/reversed", return self.__get_request("config/reversed",
{"value": 1 if value else 0}) {"value": 1 if value else 0})

14
flickerstrip_py/pattern.py
View File

@ -7,11 +7,15 @@ class PatternMeta:
self.flags = flags self.flags = flags
self.fps = fps self.fps = fps
def fromJSON(json): def to_json(self):
return PatternMeta( return {
"id": self.id, "name": self.name, "frames": self.frames,
"pixels": self.pixels, "flags": self.flags, "fps": self.fps
}
@classmethod
def from_json(cls, json):
return cls(
json["id"], json["name"], json["frames"], json["id"], json["name"], json["frames"],
json["pixels"], json["flags"], json["fps"] json["pixels"], json["flags"], json["fps"]
) )
def fromJSONArray(array):
return map(PatternMeta.fromJSON, array)

66
flickerstrip_py/ssdp.py
View File

@ -1,66 +0,0 @@
# Copyright 2014 Dan Krause, Python 3 hack 2016 Adam Baxter
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import socket
import http.client
import io
class SSDPResponse(object):
class _FakeSocket(io.BytesIO):
def makefile(self, *args, **kw):
return self
def __init__(self, response):
r = http.client.HTTPResponse(self._FakeSocket(response))
r.begin()
self.location = r.getheader("location")
self.usn = r.getheader("usn")
self.st = r.getheader("st")
self.cache = r.getheader("cache-control").split("=")[1]
def __repr__(self):
return "<SSDPResponse({location}, {st}, {usn})>"\
.format(**self.__dict__)
def discover(service, timeout=5, retries=1, mx=3):
group = ("239.255.255.250", 1900)
message = "\r\n".join([
'M-SEARCH * HTTP/1.1',
'HOST: {0}:{1}',
'MAN: "ssdp:discover"',
'ST: {st}', 'MX: {mx}', '', ''])
socket.setdefaulttimeout(timeout)
responses = {}
for _ in range(retries):
sock = socket.socket(
socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
message_bytes = message.format(
*group, st=service, mx=mx).encode('utf-8')
sock.sendto(message_bytes, group)
while True:
try:
response = SSDPResponse(sock.recv(1024))
responses[response.location] = response
except socket.timeout:
break
return list(responses.values())
# Example:
# import ssdp
# ssdp.discover("roku:ecp")

39
flickerstrip_py/status.py
View File

@ -3,12 +3,18 @@ from pattern import PatternMeta
class MemoryUsage: class MemoryUsage:
def __init__(self, used, free, total): def __init__(self, used, free, total):
self.used self.used = used
self.free self.free = free
self.total self.total = total
def fromJSON(json): def to_json(self):
return MemoryUsage(json["used"], json["free"], json["total"]) return {
"used": self.used, "free": self.free, "total": self.total
}
@classmethod
def from_json(cls, json):
return cls(json["used"], json["free"], json["total"])
class Status: class Status:
@ -35,12 +41,25 @@ class Status:
self.memory = memory self.memory = memory
self.patterns = patterns self.patterns = patterns
def fromJSON(json): def to_json(self):
return Status( return {
"ap": self.ap, "name": self.name, "group": self.group,
"firmware": self.firmware, "power": 1 if self.power else 0,
"mac": self.mac, "selectedPattern": self.selectedPattern,
"brightness": self.brightness, "length": self.length,
"start": self.start, "end": self.end, "fade": self.fade,
"reversed": 1 if self.isReversed else 0, "cycle": self.cycle,
"uptime": self.uptime, "memory": self.memory.to_json(),
"patterns": [p.to_json() for p in self.patterns]
}
@classmethod
def from_json(cls, json):
return cls(
json["ap"], json["name"], json["group"], json["firmware"], json["ap"], json["name"], json["group"], json["firmware"],
json["power"] == 1, json["mac"], json["selectedPattern"], json["power"] == 1, json["mac"], json["selectedPattern"],
json["brightness"], json["length"], json["start"], json["end"], json["brightness"], json["length"], json["start"], json["end"],
json["fade"], json["reversed"] == 1, json["cycle"] == 1, json["fade"], json["reversed"] == 1, json["cycle"],
json["uptime"], MemoryUsage.fromJSON(json["memory"]), json["uptime"], MemoryUsage.from_json(json["memory"]),
PatternMeta.fromJSONArray(json["patterns"]) [PatternMeta.from_json(p) for p in json["patterns"]]
) )