Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, ledfx):
super().__init__(ledfx, Device, self.PACKAGE_NAME)
def cleanup_effects(e):
self.clear_all_effects()
self._ledfx.events.add_listener(
cleanup_effects, Event.LEDFX_SHUTDOWN)
def __init__(self, graph_id : str, melbank : np.ndarray, frequencies : np.ndarray):
super().__init__(Event.GRAPH_UPDATE)
self.graph_id = graph_id
self.melbank = melbank.tolist()
self.frequencies = frequencies.tolist()
_LOGGER = logging.getLogger(__name__)
class Event:
"""Base for events"""
LEDFX_SHUTDOWN = 'shutdown'
DEVICE_UPDATE = 'device_update'
GRAPH_UPDATE = 'graph_update'
def __init__(self, type : str):
self.event_type = type
def to_dict(self):
return self.__dict__
class DeviceUpdateEvent(Event):
"""Event emmitted when a device's pixels are updated"""
def __init__(self, device_id : str, pixels : np.ndarray):
super().__init__(Event.DEVICE_UPDATE)
self.device_id = device_id
self.pixels = pixels.T.tolist()
class GraphUpdateEvent(Event):
"""Event emmitted when a device's pixels are updated"""
def __init__(self, graph_id : str, melbank : np.ndarray, frequencies : np.ndarray):
super().__init__(Event.GRAPH_UPDATE)
self.graph_id = graph_id
self.melbank = melbank.tolist()
self.frequencies = frequencies.tolist()
def __init__(self, type : str):
self.event_type = type
def to_dict(self):
return self.__dict__
class DeviceUpdateEvent(Event):
"""Event emmitted when a device's pixels are updated"""
def __init__(self, device_id : str, pixels : np.ndarray):
super().__init__(Event.DEVICE_UPDATE)
self.device_id = device_id
self.pixels = pixels.T.tolist()
class GraphUpdateEvent(Event):
"""Event emmitted when a device's pixels are updated"""
def __init__(self, graph_id : str, melbank : np.ndarray, frequencies : np.ndarray):
super().__init__(Event.GRAPH_UPDATE)
self.graph_id = graph_id
self.melbank = melbank.tolist()
self.frequencies = frequencies.tolist()
class LedFxShutdownEvent(Event):
"""Event emmitted when LedFx is shutting down"""
def __init__(self):
super().__init__(Event.LEDFX_SHUTDOWN)
class EventListener:
def __init__(self, callback: Callable, event_filter: dict={}):
def __init__(self, device_id : str, pixels : np.ndarray):
super().__init__(Event.DEVICE_UPDATE)
self.device_id = device_id
self.pixels = pixels.T.tolist()
async def handle(self, request):
"""Handle the websocket connection"""
socket = self._socket = web.WebSocketResponse()
await socket.prepare(request)
_LOGGER.info("Websocket connected.")
self._receiver_task = asyncio.Task.current_task(loop=self._ledfx.loop)
self._sender_task = self._ledfx.loop.create_task(self._sender())
def shutdown_handler(e):
self.close()
remove_listeners = self._ledfx.events.add_listener(
shutdown_handler, Event.LEDFX_SHUTDOWN)
try:
message = await socket.receive_json()
while message:
message = BASE_MESSAGE_SCHEMA(message)
if message['type'] in websocket_handlers:
websocket_handlers[message['type']](self, message)
else:
_LOGGER.error(('Received unknown command {}').format(message['type']))
self.send_error(message['id'], 'Unknown command type.')
message = await socket.receive_json()
except (vol.Invalid, ValueError) as e:
_LOGGER.info('Invalid message format.')
def __init__(self):
super().__init__(Event.LEDFX_SHUTDOWN)
def __init__(self, device_id : str, pixels : np.ndarray):
super().__init__(Event.DEVICE_UPDATE)
self.device_id = device_id
self.pixels = pixels.T.tolist()
class GraphUpdateEvent(Event):
"""Event emmitted when a device's pixels are updated"""
def __init__(self, graph_id : str, melbank : np.ndarray, frequencies : np.ndarray):
super().__init__(Event.GRAPH_UPDATE)
self.graph_id = graph_id
self.melbank = melbank.tolist()
self.frequencies = frequencies.tolist()
class LedFxShutdownEvent(Event):
"""Event emmitted when LedFx is shutting down"""
def __init__(self):
super().__init__(Event.LEDFX_SHUTDOWN)
class EventListener:
def __init__(self, callback: Callable, event_filter: dict={}):
self.callback = callback
self.filter = event_filter
def filter_event(self, event):
event_dict = event.to_dict()
for filter_key in self.filter:
if event_dict.get(filter_key) != self.filter[filter_key]:
return True