import neopixel from time import sleep import asyncio from typing import List import sys sys.path.append('.') import config class IndicationDriver: def __init__(self): self.pixels = neopixel.NeoPixel( config.PIXEL_PIN, config.NUM_PIXELS, brightness= config.NEOPIXEL_BRIGHTNESS, auto_write=False, pixel_order=neopixel.GRB ) self.clear() self.indication_stop_event = asyncio.Event() def stop_loop(self): self.indication_stop_event.set() def clear(self): self.pixels.fill((0,0,0)) self.pixels.show() def _get_trash_location(self, trash_type: str) -> List[int]: if trash_type in config.TRASH_LOCATION: center_location = config.TRASH_LOCATION[trash_type] return range(center_location-2, center_location+3) else: raise ValueError def indicate(self, trash_type: str): self.clear() for idx in self._get_trash_location(trash_type): self.pixels[idx] = (0, 255, 0) self.pixels.show() async def indicate_async(self, trash_type: str): self.clear() self.indication_stop_event.clear() self.indicate(trash_type) location = self._get_trash_location(trash_type) while not self.indication_stop_event.is_set(): for step in range (4): for idx in range(location[0]): if idx % 4 == step: self.pixels[idx] = (0, 0, 255) else: self.pixels[idx] = (0, 0, 0) self.pixels.show() await asyncio.sleep(.1) self.clear()