S&B Volcano vaporizer remote control with Pi Pico W

state_scan.py 3.1KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102
  1. #!/usr/bin/env python
  2. import uasyncio as asyncio
  3. from scan import ble_scan
  4. import time
  5. class StateScan:
  6. def __init__(self, lcd):
  7. self.lcd = lcd
  8. self.lock = asyncio.Lock()
  9. def enter(self, val = None):
  10. self.results = []
  11. self.current = None
  12. self.scanner = asyncio.create_task(self.scan())
  13. def exit(self):
  14. self.scanner.cancel()
  15. if self.lock.locked():
  16. self.lock.release()
  17. return self.results[self.current]
  18. async def scan(self):
  19. while True:
  20. new = await ble_scan(None, None, 0.25)
  21. for n in new:
  22. name = n.name()
  23. mac = n.device.addr_hex()
  24. rssi = n.rssi
  25. value = [name, mac, rssi, time.time()]
  26. async with self.lock:
  27. found = False
  28. for i in range(0, len(self.results)):
  29. if self.results[i][1] == mac:
  30. found = True
  31. self.results[i][0] = name
  32. self.results[i][2] = rssi
  33. self.results[i][3] = time.time()
  34. break
  35. if found == False:
  36. self.results.append(value)
  37. def draw_list(self):
  38. for i, d in enumerate(self.results):
  39. name, mac, rssi, timeout = self.results[i]
  40. s1 = "{}: {}".format(i + 1, name)
  41. s2 = "[{}] {}".format(mac, rssi)
  42. off = i * 25 + 30
  43. if off >= self.lcd.height:
  44. break
  45. c1 = self.lcd.white
  46. if name == "S&B VOLCANO H":
  47. c1 = self.lcd.green
  48. if self.current == None:
  49. self.current = i
  50. elif self.current == i:
  51. c1 = self.lcd.red
  52. c2 = self.lcd.white
  53. if self.current == i:
  54. c2 = self.lcd.red
  55. self.lcd.hline(0, off, self.lcd.width, self.lcd.blue)
  56. self.lcd.text(s1, 0, off + 2, c1)
  57. self.lcd.text(s2, 0, off + 12, c2)
  58. async def draw(self):
  59. self.lcd.text("Scanning for Bluetooth devices", 0, 10, self.lcd.red)
  60. keys = self.lcd.buttons()
  61. async with self.lock:
  62. if keys.once("enter"):
  63. if self.current < len(self.results):
  64. return 2
  65. elif keys.once("up"):
  66. if self.current == None:
  67. self.current = len(self.results) - 1
  68. elif self.current > 0:
  69. self.current -= 1
  70. elif keys.once("down"):
  71. if self.current == None:
  72. self.current = 0
  73. elif self.current < (len(self.results) - 1):
  74. self.current += 1
  75. # remove entries after timeout
  76. self.results = [x for x in self.results if (time.time() - x[3]) < 10.0]
  77. if self.current != None:
  78. if self.current >= len(self.results):
  79. self.current = len(self.results) - 1
  80. self.draw_list()
  81. return -1