CSS and JS cleanups. Rewrite battery readings ("cached results") ready

for MicroPython by LoBo, as well as PWM based on channels (separae
channels for RGB, head and tail light).

battery.py

@@ -11,10 +11,15 @@ def __init__(self, adc: int, min_value: float = 6.4, max_value: float = 8.4):
         self.max = max_value
 
     def status(self) -> dict:
-        voltage = self.adc.read() / 4095 * self.max
-        percentage = 0 if voltage < self.min else round((voltage - self.min) / (self.max - self.min) * 100, 2)
+
+        voltage_sum = 0
+        for i in range(20):
+            voltage_sum += self.adc.read()
+
+        voltage = (voltage_sum / 20) / 3134 * self.max
+        percentage = 0 if voltage < self.min else (voltage - self.min) / (self.max - self.min) * 100
 
         return {
-            'voltage': voltage,
-            'percentage': percentage
+            'voltage': round(voltage, 2),
+            'percentage': round(percentage, 2)
         }

lights.py

@@ -6,9 +6,11 @@ class Lights:
 
     DEFAULT_FREQ = 200
     DEFAULT_INTENSITY = 512
+    INTENSITY_MIN_SCALE = 0
+    INTENSITY_MAX_SCALE = 100
 
-    STATE_OFF = 0
-    STATE_ON = 1023
+    STATE_OFF = INTENSITY_MIN_SCALE
+    STATE_ON = INTENSITY_MAX_SCALE
 
     COLORS = [
         [1, 0, 0],  # red
@@ -20,32 +22,45 @@ class Lights:
     ]
 
     rgb_matrix = []
+    rgb_intensity = DEFAULT_INTENSITY
     tail_light = None
     front_light = None
 
     def __init__(self, red: int, green: int, blue: int, front: int = None, back: int = None):
         self.rgb_matrix = [
-            PWM(Pin(red, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF),
-            PWM(Pin(green, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF),
-            PWM(Pin(blue, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF)
+            PWM(Pin(red, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF, timer=0),
+            PWM(Pin(green, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF, timer=0),
+            PWM(Pin(blue, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF, timer=0)
         ]
-        self.front_light = None if front is None else PWM(Pin(front, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF)
-        self.tail_light = None if back is None else PWM(Pin(back, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF)
+        self.front_light = None if front is None else PWM(Pin(front, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF, timer=1)
+        self.tail_light = None if back is None else PWM(Pin(back, Pin.OUT, value=0), freq=self.DEFAULT_FREQ, duty=self.STATE_OFF, timer=2)
 
-    def tail(self, intensity: int):
+    def tail(self, intensity: int = DEFAULT_INTENSITY) -> None:
+        """ Manage tail light intensity
+        """
         if self.tail_light is not None:
-            self.tail_light.duty(intensity)
+            self.tail_light.freq(self.DEFAULT_FREQ)
+            self.tail_light.duty(round(intensity / self.INTENSITY_MAX_SCALE * 100, 2))
+
+    def blink_tail(self, freq: int = 5, duty: int = 50) -> None:
+        """ Make tail light blinky with specified frequency
+        """
+        self.tail_light.freq(freq)
+        self.tail_light.duty(duty)
 
-    def front(self, intensity: int):
+    def front(self, intensity: int = DEFAULT_INTENSITY) -> None:
+        """ Manage head light intensity
+        """
         if self.front_light is not None:
-            self.front_light.duty(intensity)
+            self.front_light.duty(round(intensity / self.INTENSITY_MAX_SCALE * 100, 2))
 
     @asyn.cancellable
     async def rgb_color(self, hex_color: str, intensity: int = DEFAULT_INTENSITY) -> None:
         """ Set RGB color for lights, based on HEX string
         """
         rgb = list(int(hex_color.lstrip('#')[i:i+2], 16) for i in (0, 2, 4))
-        self._set_color(rgb, intensity)
+        self._set_intensity(intensity)
+        self._set_color(rgb)
 
     @asyn.cancellable
     async def rgb_jump(self, mode: int = 2, speed: float = 0.1, intensity: int = DEFAULT_INTENSITY) -> None:
@@ -56,7 +71,8 @@ async def rgb_jump(self, mode: int = 2, speed: float = 0.1, intensity: int = DEF
                 if mode == 2 and (step + 1) % 2 == 0:
                     continue
 
-                self._set_color(map(lambda x: x * 255, color), intensity)
+                self._set_intensity(intensity)
+                self._set_color(map(lambda x: x * 255, color))
                 await asyn.sleep(speed, 10)
 
     @asyn.cancellable
@@ -80,16 +96,19 @@ async def rgb_fade(self, period: float, intensity: int = DEFAULT_INTENSITY) -> N
                     else:
                         rgb[j] = i % steps
 
-                self._set_color(map(lambda x: x / steps * 255, rgb), intensity)
+                self._set_intensity(intensity)
+                self._set_color(map(lambda x: x / steps * 255, rgb))
                 await asyn.sleep(0.01, 10)
 
     @asyn.cancellable
     async def rgb_police(self, intensity: int = DEFAULT_INTENSITY) -> None:
         """ Flash RGB like police lights (red and blue)
         """
+        self._set_intensity(intensity)
+
         while True:
-            await self._blink(self.rgb_matrix[0], intensity)
-            await self._blink(self.rgb_matrix[2], intensity)
+            await self._blink(self.rgb_matrix[0])
+            await self._blink(self.rgb_matrix[2])
 
     def rgb_reset(self) -> None:
         """ Reset RGB PWMs to neutral state
@@ -98,24 +117,34 @@ def rgb_reset(self) -> None:
             pin.freq(self.DEFAULT_FREQ)
             pin.duty(self.STATE_OFF)
 
-    async def rgb_cancel(self):
+    async def rgb_cancel(self) -> None:
         """ Cancel all working coros and reset RGB lights
         """
         await asyn.Cancellable.cancel_all()
         self.rgb_reset()
 
-    def _set_color(self, color: iter, intensity: int = DEFAULT_INTENSITY) -> None:
+    def _set_intensity(self, intensity: int) -> int:
+        self.rgb_intensity = intensity
+        return intensity
+
+    def _set_color(self, color: iter) -> None:
         """ Set specified color for RGB Pins, expected list with values of each color in range 0-255
         """
         for pin, value in enumerate(color):
-            self.rgb_matrix[pin].duty(int(float(value)/255 * intensity))
+            dty = (float(value)/255) * (self.rgb_intensity / self.INTENSITY_MAX_SCALE) * 100
+            self.rgb_matrix[pin].duty(round(dty, 2))
+
+    def _calculate_color_value(self, pin: PWM) -> int:
+        """ Return color value in range 0-255 based on duty value
+        """
+        divider = self.rgb_intensity / self.INTENSITY_MAX_SCALE
+        return 0 if divider == 0 else int(255 * (pin.duty() / divider / 100))
 
-    @staticmethod
-    async def _blink(led: PWM, intensity: int = DEFAULT_INTENSITY) -> None:
+    async def _blink(self, led: PWM) -> None:
         """ Blink led 3 times and wait
         """
         led.freq(10)
-        led.duty(intensity)
+        led.duty(round(self.rgb_intensity / self.INTENSITY_MAX_SCALE * 100, 2))
         await asyn.sleep(0.3, 10)
         led.duty(0)
         await asyn.sleep(0.2, 10)
@@ -126,11 +155,14 @@ def status(self) -> dict:
         return {
             'rgb': {
                 'mode': '',
-                'r': self.rgb_matrix[0].duty(),
-                'g': self.rgb_matrix[1].duty(),
-                'b': self.rgb_matrix[2].duty()
+                'r': self._calculate_color_value(self.rgb_matrix[0]),
+                'g': self._calculate_color_value(self.rgb_matrix[1]),
+                'b': self._calculate_color_value(self.rgb_matrix[2])
             },
-            'front': self.front_light.duty(),
-            'tail': self.tail_light.duty(),
+            'front': int(self.front_light.duty() * self.INTENSITY_MAX_SCALE / 100),
+            'tail': {
+                'mode': '',
+                'value': int(self.tail_light.duty() * self.INTENSITY_MAX_SCALE / 100),
+            }
         }
 

lights_service.py

@@ -1,20 +1,22 @@
-from uwebsocket import WebSocketConnection
 from battery import Battery
 from lights import Lights
 import uasyncio as asyncio
+import ujson
 import asyn
 
 
 class LightsService:
 
+    cache_battery = None
+
     def __init__(self, lights: Lights, battery: Battery):
         self.battery = battery
         self.lights = lights
 
     async def action_handler(self, action: str, *args, **kwargs):
         """ Run specified action after cancelled previous
         """
-        if action in ['front', 'tail']:
+        if action in ['front', 'tail', 'blink_tail']:
             getattr(self.lights, action)(*args, **kwargs)
             return
 
@@ -23,12 +25,23 @@ async def action_handler(self, action: str, *args, **kwargs):
         if action != 'cancel':
             asyncio.get_event_loop().create_task(asyn.Cancellable(getattr(self.lights, 'rgb_' + action), *args, **kwargs)())
 
-    async def status_worker(self, connection: WebSocketConnection, freq: int = 2):
+    async def status_worker(self, connection, freq: float = 2):
         """ Worker which updates current status via websocket
         """
+        i = 0
         while True:
-            connection.write(self.read_status())
+
+            if connection.ws is not None:
+                connection.write(ujson.dumps(self.read_status(i == 0)))
+                i = 0 if i > 100 else i + 1
+
             await asyncio.sleep(freq)
 
-    def read_status(self):
-        return {**self.battery.status(), **self.lights.status()}
+    def read_status(self, refresh_battery: bool = False):
+        status = self.lights.status()
+
+        if refresh_battery is True:
+            self.cache_battery = self.battery.status()
+
+        status.update(self.cache_battery)
+        return status

main.py

@@ -9,11 +9,10 @@
 ap = network.WLAN(network.AP_IF)
 ap.active(True)
 ap.config(essid=b"Evolve Bamboo GT", authmode=network.AUTH_WPA_WPA2_PSK, password=b"MyBambooGT")
-lights = Lights(14, 16, 12, 26, 27)
-battery = Battery(33, max_value=8.2)
+lights = Lights(27, 16, 12, 26, 32)
+battery = Battery(33)
 app_socket = webserver.Server(LightsService(lights, battery))
 
-
 try:
 
     app_socket.start()

webserver.py

@@ -19,7 +19,7 @@ class Client(WebSocketClient):
 
     def __init__(self, conn, lights_service: LightsService):
         self.lights_service = lights_service
-        asyncio.get_event_loop().call_soon(lights_service.status_worker(conn))
+        asyncio.get_event_loop().call_soon(self.lights_service.status_worker(conn, 0.1))
         super().__init__(conn)
 
     def process(self):
@@ -35,11 +35,9 @@ def process(self):
 
             if command is not None:
 
-                if command in ['front', 'tail', 'color', 'jump', 'fade', 'police', 'cancel']:
+                if command in ['front', 'tail', 'blink_tail', 'color', 'jump', 'fade', 'police', 'cancel']:
                     loop.call_soon(self.lights_service.action_handler(command, **msg))
                     self.connection.write(command + ' : ' + ujson.dumps(msg))
-                elif command == 'tasks':
-                    self.connection.write(ujson.dumps(asyn.Cancellable.tasks))
                 elif command == 'status':
                     self.connection.write(ujson.dumps(self.lights_service.read_status()))
                 else: