peterlu14
80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
from time import sleep, time
|
|
|
|
from biopro.impl.gpio import P3Pin, OutputPin, InputPin
|
|
from biopro.impl.led import LED
|
|
from biopro.message import DataMessage
|
|
from biopro.util.console import YELLOW
|
|
from biopro.util.sys import system_sudo
|
|
from .controller import ControlServerAPI
|
|
from .socket import ServerThread
|
|
|
|
|
|
class UPSDetectThread(ServerThread):
|
|
"""UPS power detect thread. This thread start when UPS_DETECT pin pull low, and wait N seconds.
|
|
If UPS_DETECT always low for more than N seconds, pull low UPS_ENABLE to power off controller.
|
|
|
|
"""
|
|
|
|
UPS_ENABLE: OutputPin = None
|
|
UPS_DETECT: InputPin = None
|
|
|
|
def __init__(self,
|
|
server: ControlServerAPI,
|
|
wait_time: int):
|
|
super().__init__('UPS', YELLOW)
|
|
|
|
self._init_gpio()
|
|
|
|
self._server = server
|
|
|
|
self._wait_time = wait_time
|
|
self._start_time = 0
|
|
|
|
self._prev_led_state = LED.AVAILABLE
|
|
|
|
@classmethod
|
|
def _init_gpio(cls):
|
|
if cls.UPS_ENABLE is None:
|
|
cls.UPS_ENABLE = OutputPin.get_used(P3Pin.UPS_ENABLE, initial=True)
|
|
cls.UPS_DETECT = InputPin.get_used(P3Pin.UPS_DETECT, pull_up_down=True)
|
|
return
|
|
|
|
@classmethod
|
|
def power_off(cls):
|
|
cls._init_gpio()
|
|
|
|
# power off
|
|
cls.UPS_ENABLE.output(False)
|
|
|
|
# un-reachable, iff UPS_A work functionally
|
|
# if not, shutdown OS
|
|
system_sudo('shutdown', '-h', 'now')
|
|
return
|
|
|
|
def routine(self, event = None):
|
|
if bool(self.UPS_DETECT):
|
|
if self._start_time > 0:
|
|
self._server.set_led_state(self._prev_led_state)
|
|
self._server.broadcast_command(DataMessage.low_power(DataMessage.LOW_POWER_STATE[1]))
|
|
|
|
self._start_time = 0
|
|
else:
|
|
if self._start_time == 0:
|
|
self._start_time = time()
|
|
self._prev_led_state = self._server.get_led_state()
|
|
self._server.set_led_state(LED.UPS_USING)
|
|
self._server.broadcast_command(DataMessage.low_power(DataMessage.LOW_POWER_STATE[0]))
|
|
|
|
elif time() - self._start_time > self._wait_time:
|
|
|
|
self._server.broadcast_command(DataMessage.low_power(DataMessage.LOW_POWER_STATE[2]))
|
|
|
|
event.wait(1)
|
|
|
|
self._server.hardware_device_disconnect(None, force=True)
|
|
self.power_off()
|
|
self.close()
|
|
|
|
event.wait(1)
|
|
return
|