Background thread to monitor input pins for changes
()
| 126 | log_event("GPIO cleanup completed", 'info') |
| 127 | |
| 128 | def input_monitor_worker(): |
| 129 | """Background thread to monitor input pins for changes""" |
| 130 | global monitor_running |
| 131 | previous_states = {} |
| 132 | |
| 133 | while monitor_running: |
| 134 | try: |
| 135 | for pin in list(input_monitors): |
| 136 | if pin in pin_states and pin_states[pin]['mode'] == 'input': |
| 137 | try: |
| 138 | current_value = read_pin_value(pin) |
| 139 | previous_value = previous_states.get(pin, current_value) |
| 140 | |
| 141 | if current_value != previous_value: |
| 142 | edge_type = 'rising' if current_value > previous_value else 'falling' |
| 143 | log_event(f"Pin {pin} {edge_type} edge detected (value: {current_value})", 'input') |
| 144 | previous_states[pin] = current_value |
| 145 | except Exception as e: |
| 146 | log_event(f"Error monitoring pin {pin}: {str(e)}", 'error') |
| 147 | |
| 148 | time.sleep(0.1) # 100ms polling interval |
| 149 | |
| 150 | except Exception as e: |
| 151 | log_event(f"Monitor thread error: {str(e)}", 'error') |
| 152 | time.sleep(1) |
| 153 | |
| 154 | def read_pin_value(pin): |
| 155 | """Read current value from a pin""" |
nothing calls this directly
no test coverage detected