1from enum import IntEnum 

2 

3 

4class PDUState(IntEnum): 

5 UNKNOWN = 0 

6 OFF = 1 

7 ON = 2 

8 REBOOT = 3 

9 

10 @classmethod 

11 def valid_actions(cls): 

12 return [s for s in PDUState if s.value > 0] 

13 

14 @property 

15 def is_valid_action(self): 

16 return self in self.valid_actions() 

17 

18 

19class PDUPort: 

20 def __init__(self, pdu, port_id, label=None): 

21 self.pdu = pdu 

22 self.port_id = port_id 

23 self.label = label 

24 

25 def set(self, state): 

26 self.pdu.set_port_state(self.port_id, state) 

27 

28 @property 

29 def state(self): 

30 return self.pdu.get_port_state(self.port_id) 

31 

32 

33class PDU: 

34 pdus = {} 

35 

36 def __init__(self, name): 

37 self.name = name 

38 

39 self.pdus[name] = self 

40 

41 @property 

42 def ports(self): 

43 # NOTICE: Left for drivers to implement 

44 return [] 

45 

46 def set_port_state(self, port_id, state): 

47 # NOTICE: Left for drivers to implement 

48 return False 

49 

50 def get_port_state(self, port_id): 

51 # NOTICE: Left for drivers to implement 

52 return PDUState.UNKNOWN 

53 

54 def unregister(self): 

55 del self.pdus[self.name] 

56 

57 @classmethod 

58 def supported_pdus(cls): 

59 from drivers.apc import ApcMasterswitchPDU 

60 from drivers.cyberpower import PDU41004 

61 from drivers.dummy import DummyPDU 

62 from drivers.snmp import SnmpPDU 

63 

64 return { 

65 "apc_masterswitch": ApcMasterswitchPDU, 

66 "cyberpower_pdu41004": PDU41004, 

67 "dummy": DummyPDU, 

68 "snmp": SnmpPDU, 

69 } 

70 

71 @classmethod 

72 def register(cls, model_name, pdu_name, config): 

73 Driver = cls.supported_pdus().get(model_name) 

74 

75 if Driver is None: 

76 raise ValueError(f"Unknown model name '{model_name}'") 

77 

78 return Driver(pdu_name, config) 

79 

80 @classmethod 

81 def get_by_name(cls, name): 

82 return cls.pdus.get(name) 

83 

84 @classmethod 

85 def registered_pdus(cls): 

86 return cls.pdus 

87 

88 @classmethod 

89 def full_reset(cls): 

90 cls.pdus.clear()