Move the different switching and sensing methods into their own self-contained classes (see example below).
This improves clarity in the code and simplifies testing, error-handing and makes it easy to add or modify switching and sensing implementations.
Note: Implementing this will change the codebase quite a bit and will likely conflict with any other PRs written at the same time.
Possible implementation (click to expand):
```python
#
# for python 2.7 backwards compatibility, see https://stackoverflow.com/a/41622155
# ----------------------------------------------------------------
#
import sys
import abc
if sys.version_info >= (3, 4):
ABC = abc.ABC
else:
ABC = abc.ABCMeta('ABC', (), {})
#
# define general interfaces and shared fuctionality
# ----------------------------------------------------------------
#
class SwitcherBase(ABC):
"""
base class for switching the PSU on and off
"""
def __init__(self, logger):
self._logger = logger
@abc.abstractmethod
def set_psu_state(self, state):
# type: (bool) -> None
"""
switch the PSU on or off
"""
def cleanup(self):
"""
clean up any used resources
"""
class SenserBase(ABC):
"""
base class for sensing the PSU's current state
"""
def __init__(self, logger):
self._logger = logger
@abc.abstractmethod
def get_psu_state(self):
# type: () -> Optional[bool]
"""
read the current on / off state of the PSU and return it as a boolean, return None if the state is unknown
"""
def cleanup(self):
"""
clean up any used resources
"""
pass
#
# implementation of the specific methods
# ----------------------------------------------------------------
#
class GPIOSwitcher(SwitcherBase):
"""
switches PSU state via GPIO
"""
def __init__(self, path, pin_number, inverted, logger):
super(GPIOSwitcher, self).__init__(logger)
self._logger.info("Configuring GPIO for pin {}".format(pin_number))
try:
self._pin = periphery.CdevGPIO(
path=path,
line=pin_number,
direction="out",
inverted=inverted,
)
except Exception as e:
self._logger.exception(
"Exception while setting up GPIO pin {}".format(pin_number)
)
self._pin = None
raise e
def set_psu_state(self, state):
if self._pin is not None:
self._pin.write()
def cleanup(self):
self._logger.debug("Cleaning up sensing pin {}".format(self._pin.name()))
if self._pin is not None:
# should work, even if we try to clean up the same GPIO twice
self._pin.close()
class SystemSenser(SenserBase):
"""
reads PSU state via user-supplied system command
"""
def __init__(self, sense_command, logger):
super(SystemSenser, self).__init__(logger)
self._sense_command = sense_command
def get_psu_state(self):
p = subprocess.Popen(self._sense_command, shell=True)
self._logger.debug(
"Sensing system command executed. PID={}, Command={}".format(p.pid, self._sense_command))
p.wait() # TODO: set a timeout?
r = p.returncode
self._logger.debug("Sensing system command returned: {}".format(r))
return not bool(r)
#
# use a factory function to produce the right switcher and senser objects based on the config
# ----------------------------------------------------------------
#
class PSUControl(...):
...
@classmethod
def _switcher_senser_factory(self, config, logger):
# type: (...) -> Tuple[SenserBase, SwitcherBase]
if config['switchingMethod'] == 'GCODE':
...
elif config['switchingMethod'] == 'SYSTEM':
...
...
```
RE: https://github.com/kantlivelong/OctoPrint-PSUControl/issues/223#issuecomment-913915951_
Move the different switching and sensing methods into their own self-contained classes (see example below). This improves clarity in the code and simplifies testing, error-handing and makes it easy to add or modify switching and sensing implementations.
Note: Implementing this will change the codebase quite a bit and will likely conflict with any other PRs written at the same time.
Possible implementation (click to expand):
```python # # for python 2.7 backwards compatibility, see https://stackoverflow.com/a/41622155 # ---------------------------------------------------------------- # import sys import abc if sys.version_info >= (3, 4): ABC = abc.ABC else: ABC = abc.ABCMeta('ABC', (), {}) # # define general interfaces and shared fuctionality # ---------------------------------------------------------------- # class SwitcherBase(ABC): """ base class for switching the PSU on and off """ def __init__(self, logger): self._logger = logger @abc.abstractmethod def set_psu_state(self, state): # type: (bool) -> None """ switch the PSU on or off """ def cleanup(self): """ clean up any used resources """ class SenserBase(ABC): """ base class for sensing the PSU's current state """ def __init__(self, logger): self._logger = logger @abc.abstractmethod def get_psu_state(self): # type: () -> Optional[bool] """ read the current on / off state of the PSU and return it as a boolean, return None if the state is unknown """ def cleanup(self): """ clean up any used resources """ pass # # implementation of the specific methods # ---------------------------------------------------------------- # class GPIOSwitcher(SwitcherBase): """ switches PSU state via GPIO """ def __init__(self, path, pin_number, inverted, logger): super(GPIOSwitcher, self).__init__(logger) self._logger.info("Configuring GPIO for pin {}".format(pin_number)) try: self._pin = periphery.CdevGPIO( path=path, line=pin_number, direction="out", inverted=inverted, ) except Exception as e: self._logger.exception( "Exception while setting up GPIO pin {}".format(pin_number) ) self._pin = None raise e def set_psu_state(self, state): if self._pin is not None: self._pin.write() def cleanup(self): self._logger.debug("Cleaning up sensing pin {}".format(self._pin.name())) if self._pin is not None: # should work, even if we try to clean up the same GPIO twice self._pin.close() class SystemSenser(SenserBase): """ reads PSU state via user-supplied system command """ def __init__(self, sense_command, logger): super(SystemSenser, self).__init__(logger) self._sense_command = sense_command def get_psu_state(self): p = subprocess.Popen(self._sense_command, shell=True) self._logger.debug( "Sensing system command executed. PID={}, Command={}".format(p.pid, self._sense_command)) p.wait() # TODO: set a timeout? r = p.returncode self._logger.debug("Sensing system command returned: {}".format(r)) return not bool(r) # # use a factory function to produce the right switcher and senser objects based on the config # ---------------------------------------------------------------- # class PSUControl(...): ... @classmethod def _switcher_senser_factory(self, config, logger): # type: (...) -> Tuple[SenserBase, SwitcherBase] if config['switchingMethod'] == 'GCODE': ... elif config['switchingMethod'] == 'SYSTEM': ... ... ```