Coverage for microservice_websocket/app/utils/node_settings.py: 83%

1from ..models.node_settings import DetectorSettings, SensorSettings 

2from ..services.database import Node, NodeSettings 

3from .command import send_set_hv_command, send_set_window_high, send_set_window_low 

4from .exceptions import NotFoundException 

5 

6 

7async def get_node_settings(nodeID: int) -> NodeSettings: 

8 node = await Node.find_one(Node.nodeID == nodeID) 

9 if node is None: 9 ↛ 10line 9 didn't jump to line 10, because the condition on line 9 was never true

10 raise NotFoundException("Node") 

11 

12 settings = await NodeSettings.find_one(NodeSettings.node == node.id) 

13 if settings is None: 13 ↛ 14line 13 didn't jump to line 14, because the condition on line 13 was never true

14 raise NotFoundException("NodeSettings") 

15 

16 return settings 

17 

18 

19async def send_update_settings_sensor( 

20 applicationID: str, 

21 nodeID: int, 

22 sensor_settings: SensorSettings, 

23 detector: int, 

24 sipm: int, 

25): 

26 if sensor_settings.w1_low is not None: 26 ↛ 30line 26 didn't jump to line 30, because the condition on line 26 was never false

27 await send_set_window_low( 

28 applicationID, nodeID, detector, sipm, 1, sensor_settings.w1_low 

29 ) 

30 if sensor_settings.w1_high is not None: 30 ↛ 34line 30 didn't jump to line 34, because the condition on line 30 was never false

31 await send_set_window_high( 

32 applicationID, nodeID, detector, sipm, 1, sensor_settings.w1_high 

33 ) 

34 if sensor_settings.w2_low is not None: 34 ↛ 38line 34 didn't jump to line 38, because the condition on line 34 was never false

35 await send_set_window_low( 

36 applicationID, nodeID, detector, sipm, 2, sensor_settings.w2_low 

37 ) 

38 if sensor_settings.w2_high is not None: 38 ↛ 42line 38 didn't jump to line 42, because the condition on line 38 was never false

39 await send_set_window_high( 

40 applicationID, nodeID, detector, sipm, 2, sensor_settings.w2_high 

41 ) 

42 if sensor_settings.w3_low is not None: 42 ↛ 46line 42 didn't jump to line 46, because the condition on line 42 was never false

43 await send_set_window_low( 

44 applicationID, nodeID, detector, sipm, 3, sensor_settings.w3_low 

45 ) 

46 if sensor_settings.w3_high is not None: 46 ↛ 50line 46 didn't jump to line 50, because the condition on line 46 was never false

47 await send_set_window_high( 

48 applicationID, nodeID, detector, sipm, 3, sensor_settings.w3_high 

49 ) 

50 if sensor_settings.hv is not None: 50 ↛ exitline 50 didn't return from function 'send_update_settings_sensor', because the condition on line 50 was never false

51 await send_set_hv_command( 

52 applicationID, nodeID, detector, sipm, sensor_settings.hv 

53 ) 

54 

55 

56async def send_update_settings_detector( 

57 applicationID: str, 

58 nodeID: int, 

59 detector_settings: DetectorSettings, 

60 detector: int, 

61): 

62 if detector_settings.s1 is not None: 62 ↛ 66line 62 didn't jump to line 66, because the condition on line 62 was never false

63 await send_update_settings_sensor( 

64 applicationID, nodeID, detector_settings.s1, detector, 1 

65 ) 

66 if detector_settings.s2 is not None: 66 ↛ exitline 66 didn't return from function 'send_update_settings_detector', because the condition on line 66 was never false

67 await send_update_settings_sensor( 

68 applicationID, nodeID, detector_settings.s2, detector, 2 

69 ) 

70 

71 

72async def send_update_settings( 

73 applicationID: str, 

74 nodeID: int, 

75 node_settings: NodeSettings | NodeSettings.Serialized, 

76): 

77 if node_settings.d1 is not None: 77 ↛ 79line 77 didn't jump to line 79, because the condition on line 77 was never false

78 await send_update_settings_detector(applicationID, nodeID, node_settings.d1, 1) 

79 if node_settings.d2 is not None: 79 ↛ 81line 79 didn't jump to line 81, because the condition on line 79 was never false

80 await send_update_settings_detector(applicationID, nodeID, node_settings.d2, 2) 

81 if node_settings.d3 is not None: 81 ↛ 83line 81 didn't jump to line 83, because the condition on line 81 was never false

82 await send_update_settings_detector(applicationID, nodeID, node_settings.d3, 3) 

83 if node_settings.d4 is not None: 83 ↛ exitline 83 didn't return from function 'send_update_settings', because the condition on line 83 was never false

84 await send_update_settings_detector(applicationID, nodeID, node_settings.d4, 4) 

85 

86 

87def delta_dict_recursive(a: dict, b: dict) -> dict: 

88 delta_dict = {} 

89 

90 for key in a.keys(): 

91 if isinstance(a[key], dict): 

92 val = delta_dict_recursive(a[key], b[key]) 

93 if val: 93 ↛ 90line 93 didn't jump to line 90, because the condition on line 93 was never false

94 delta_dict[key] = val 

95 else: 

96 if a[key] != b[key]: 96 ↛ 90line 96 didn't jump to line 90, because the condition on line 96 was never false

97 delta_dict[key] = a[key] 

98 

99 return delta_dict 

100 

101 

102async def update_node_settings(nodeID: int, payload: NodeSettings.Serialized): 

103 

104 node = await Node.find_one(Node.nodeID == nodeID) 

105 if node is None: 

106 raise NotFoundException("Node") 

107 

108 settings = await NodeSettings.find_one(NodeSettings.node == node.id) 

109 if settings is None: 

110 settings = NodeSettings( 

111 node=node.id, d1=payload.d1, d2=payload.d2, d3=payload.d3, d4=payload.d4 

112 ) 

113 await settings.save() 

114 

115 # Send whole payload 

116 await send_update_settings(str(node.application), nodeID, settings) 

117 

118 return 

119 

120 old_settings_dict = settings.serialize().dict() 

121 new_settings_dict = payload.dict() 

122 delta_settings: NodeSettings.Serialized = NodeSettings.Serialized.parse_obj( 

123 delta_dict_recursive(new_settings_dict, old_settings_dict) 

124 ) 

125 

126 settings.d1 = payload.d1 

127 settings.d2 = payload.d2 

128 settings.d3 = payload.d3 

129 settings.d4 = payload.d4 

130 await settings.save() 

131 

132 await send_update_settings(str(node.application), nodeID, delta_settings)