S&B Volcano vaporizer remote control with Pi Pico W
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

flow.py 2.8KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112
  1. #!/usr/bin/env python
  2. import sys
  3. import time
  4. import os
  5. from poll import (
  6. ble_conn,
  7. get_current_temp,
  8. get_target_temp, set_target_temp,
  9. get_unit_is_fahrenheit,
  10. get_state, set_state
  11. )
  12. terminal_width = os.get_terminal_size().columns - 15
  13. def print_bar(value, start, end, unit):
  14. width = terminal_width
  15. s = "\r"
  16. s += "#" * int((value - start) / (end - start) * width)
  17. s += "-" * (width - int((value - start) / (end - start) * width))
  18. s += " {}{}".format(value, unit)
  19. print(s, end="", flush=True)
  20. def sleep(t):
  21. w = terminal_width
  22. if t < w:
  23. w = int(t)
  24. print_bar(0, 0, w, "s")
  25. for i in range(0, w):
  26. time.sleep(t / w)
  27. print_bar(i + 1, 0, w, "s")
  28. print()
  29. def wait_for_temp(client, temp):
  30. print("Setting temperature {}".format(temp))
  31. set_target_temp(client, temp)
  32. print("Waiting for temperature to rise...")
  33. start = get_current_temp(client)
  34. curr = start
  35. print_bar(curr, start, temp, " degC")
  36. while curr < temp:
  37. time.sleep(1.0)
  38. curr = get_current_temp(client)
  39. print_bar(curr, start, temp, " degC")
  40. print()
  41. print("Reached temperature {}".format(temp))
  42. def flow_step(client, temp, t_wait, t_pump):
  43. wait_for_temp(client, temp)
  44. print("Waiting {}s for heat to settle...".format(t_wait))
  45. sleep(t_wait)
  46. print("Pumping for {}s".format(t_pump))
  47. set_state(client, (True, True)) # turn on pump
  48. sleep(t_pump)
  49. set_state(client, (True, False)) # turn off pump
  50. def flow(client):
  51. print("Turning on heater")
  52. set_state(client, (True, False))
  53. flow_step(client, 185.0, 10.0, 7.0)
  54. flow_step(client, 195.0, 5.0, 23.0)
  55. flow_step(client, 205.0, 5.0, 23.0)
  56. print("Notification by pumping three times...")
  57. for i in range(0, 3):
  58. time.sleep(1.0)
  59. set_state(client, (True, True)) # turn on pump
  60. time.sleep(1.0)
  61. set_state(client, (True, False)) # turn off pump
  62. print("Turning heater off")
  63. set_state(client, (False, False)) # turn off heater and pump
  64. print("Resetting temperature")
  65. set_target_temp(client, 190.0)
  66. if __name__ == "__main__":
  67. def main(address, adapter):
  68. client = ble_conn(address, adapter)
  69. try:
  70. if get_unit_is_fahrenheit(client):
  71. raise RuntimeError("Imperial American scum is currently not supported :P")
  72. print("Starting Workflow")
  73. flow(client)
  74. except:
  75. print("\nTurning heater and pump off")
  76. set_state(client, (False, False)) # turn off heater and pump
  77. print("Disconnecting")
  78. client.disconnect()
  79. raise
  80. print("Disconnecting")
  81. client.disconnect()
  82. adapter = None
  83. mac = None
  84. if len(sys.argv) > 1:
  85. adapter = int(sys.argv[1])
  86. if len(sys.argv) > 2:
  87. mac = sys.argv[2]
  88. main(mac, adapter)