Mercurial > code > home > repos > light-bridge

comparison light.py @ 11:028ed39aa78f

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
more ts types; attempted multiplayer but the change events are managed wrong
author drewp@bigasterisk.com
date Sun, 28 Jan 2024 17:31:06 -0800
parents 9f427d8073c3
children 7cc004eafb82
comparison
equal deleted inserted replaced
10:140633abfa2a 11:028ed39aa78f
1 import asyncio 1 import asyncio
2 import logging 2 import logging
3 from dataclasses import dataclass, field 3 from dataclasses import dataclass, field
4 from typing import Callable
4 5
5 from color import Color 6 from color import Color
6 7
7 log = logging.getLogger('light') 8 log = logging.getLogger('light')
8 9
31 32
32 emittingColor: Color = Color.fromHex('#000000') 33 emittingColor: Color = Color.fromHex('#000000')
33 online: bool | None = None 34 online: bool | None = None
34 latencyMs: float | None = None 35 latencyMs: float | None = None
35 36
37 notifyChanged: Callable | None = None
38
36 def __post_init__(self): 39 def __post_init__(self):
37 self.requestingDeviceColor = self.deviceColor(self.requestingColor) 40 self.requestingDeviceColor = self.deviceColor(self.requestingColor)
38 41
39 def to_dict(self): 42 def to_dict(self):
40 d = { 43 d = {
52 def deviceColor(self, c: Color) -> DeviceColor: 55 def deviceColor(self, c: Color) -> DeviceColor:
53 # do LUT here 56 # do LUT here
54 return DeviceColor(r=c.r, g=c.g, b=c.b) 57 return DeviceColor(r=c.r, g=c.g, b=c.b)
55 58
56 async def setColor(self, c: Color): 59 async def setColor(self, c: Color):
60 log.info(f'setColor from {self.requestingColor} to {c}')
61 if c == self.requestingColor:
62 return
57 self.requestingColor = c 63 self.requestingColor = c
58 self.requestingDeviceColor = self.deviceColor(self.requestingColor) 64 self.requestingDeviceColor = self.deviceColor(self.requestingColor)
65 if self.notifyChanged:
66 self.notifyChanged()
59 67
60 68
61 class Lights: 69 class Lights:
62 _d: dict[str, Light] = {} 70 _d: dict[str, Light] = {}
71 _changeSleepTask: asyncio.Task | None = None
63 72
64 def __init__(self): 73 def __init__(self):
65 self.add(Light('do-desk', 'topic1')) 74 self.add(Light('do-desk', 'topic1'))
66 self.add(Light('do-desk2', 'topic2')) 75 self.add(Light('do-desk2', 'topic2'))
67 76
68 def add(self, d: Light): 77 def add(self, d: Light):
78 d.notifyChanged = self.notifyChanged
69 self._d[d.name] = d 79 self._d[d.name] = d
80
81 self.notifyChanged()
70 82
71 def byName(self, name: str) -> Light: 83 def byName(self, name: str) -> Light:
72 return self._d[name] 84 return self._d[name]
73 85
74 async def changes(self): # yields None on any data change 86 async def changes(self): # yields None on any data change
75 while True: 87 while True:
88 log.info('Lights has a change')
76 yield None 89 yield None
77 await asyncio.sleep(1) # todo 90 self._changeSleepTask = asyncio.create_task(self._sleep())
91 try:
92 await self._changeSleepTask
93 except asyncio.CancelledError:
94 pass
95 self._changeSleepTask = None
96
97 async def _sleep(self):
98 await asyncio.sleep(100)
99
100 def notifyChanged(self):
101 log.info('Lights.notifyChanged()')
102 if self._changeSleepTask is not None:
103 self._changeSleepTask.cancel()
78 104
79 def to_dict(self): 105 def to_dict(self):
80 return {'lights': [d.to_dict() for d in sorted(self._d.values(), key=lambda r: r.name)]} 106 return {'lights': [d.to_dict() for d in sorted(self._d.values(), key=lambda r: r.name)]}