finished gpio repositories, PID logic sservice and temp_service #4
Conversation
YeYethePooh
requested review from
Copilot and
leomotors
and removed request for
Copilot
There was a problem hiding this comment.
Pull Request Overview
This PR implements a PID-based temperature control system with a background thread that continuously monitors temperature and adjusts heater duty cycle. The implementation includes a new PID controller, refactored GPIO abstractions to support thread-safe operations, and integration with an external temperature reading API.
- Introduces PIDController class with thread-safe coefficient updates and anti-windup logic
- Implements background control loop using threading for continuous temperature regulation
- Refactors GPIO interface to remove pin parameter from method calls and add concrete RPiGPIORepository implementation
Reviewed Changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 15 comments.
| File | Description |
|---|---|
| apps/heater-api/services/temp_service.py | Integrates PID controller, implements background control loop with threading, and adds API client for temperature readings |
| apps/heater-api/services/PID.py | New PID controller implementation with thread safety and anti-windup |
| apps/heater-api/repositories/gpio.py | Refactors abstract interface and adds concrete RPiGPIORepository with thread safety |
| apps/heater-api/mocks/gpio_mock.py | Updates mock to match new simplified interface |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| config = Configuration(host="http://localhost:8000") | ||
| api_client = ApiClient(config) | ||
|
|
||
| readingApi = ReadingApi(api_client=api_client) |
There was a problem hiding this comment.
Global variables create shared mutable state that makes testing difficult and can cause issues in multi-threaded environments. Move these into the TempService.init method as instance variables to improve testability and avoid potential state conflicts.
| def _control_loop(self): | ||
| """Internal method that runs in background to control temperature.""" | ||
| while self._running: | ||
| current_temp = readingApi.get_monitoring(1).kelvin - 273.15 |
There was a problem hiding this comment.
The hardcoded monitoring ID '1' and temperature conversion constant '273.15' are magic numbers. Extract these as named constants (e.g., MONITORING_SENSOR_ID and KELVIN_TO_CELSIUS_OFFSET) to improve code clarity and maintainability.
| current_temp = readingApi.get_monitoring(1).kelvin - 273.15 | ||
| duty = self._pid.update(current_temp) | ||
| self.gpio.set_duty_cycle(duty_cycle=duty) | ||
|
|
||
| print( | ||
| f"T={current_temp:.2f}°C | Target={self._target:.1f}°C | Duty={duty:.1f}%") | ||
|
|
There was a problem hiding this comment.
The API call lacks exception handling, which could crash the control loop if the network request fails or returns an error. Wrap this in a try-except block to handle potential exceptions and prevent the control thread from terminating unexpectedly.
| current_temp = readingApi.get_monitoring(1).kelvin - 273.15 | |
| duty = self._pid.update(current_temp) | |
| self.gpio.set_duty_cycle(duty_cycle=duty) | |
| print( | |
| f"T={current_temp:.2f}°C | Target={self._target:.1f}°C | Duty={duty:.1f}%") | |
| try: | |
| current_temp = readingApi.get_monitoring(1).kelvin - 273.15 | |
| duty = self._pid.update(current_temp) | |
| self.gpio.set_duty_cycle(duty_cycle=duty) | |
| print( | |
| f"T={current_temp:.2f}°C | Target={self._target:.1f}°C | Duty={duty:.1f}%") | |
| except Exception as e: | |
| print(f"Error in control loop: {e}") |
| def start(self): | ||
| """Start the temperature control loop in a background thread.""" | ||
| if self._running: | ||
| print("Control loop already running.") |
There was a problem hiding this comment.
Using print statements for logging in production code is not recommended. Consider using Python's logging module instead for better control over log levels, formatting, and output destinations.
| self._running = True | ||
| self._thread = threading.Thread(target=self._control_loop, daemon=True) | ||
| self._thread.start() | ||
| print("Temperature control loop started.") |
There was a problem hiding this comment.
Using print statements for logging in production code is not recommended. Consider using Python's logging module instead for better control over log levels, formatting, and output destinations.
| except Exception: | ||
| raise RuntimeError( | ||
| "Failed to stop existing PWM instance, could not reinitialize PWM") |
There was a problem hiding this comment.
The error message does not include the underlying exception details, making debugging difficult. Include the original exception information using 'from e' or include the exception message in the error text to provide more context about what went wrong.
| except Exception: | |
| raise RuntimeError( | |
| "Failed to stop existing PWM instance, could not reinitialize PWM") | |
| except Exception as e: | |
| raise RuntimeError( | |
| f"Failed to stop existing PWM instance, could not reinitialize PWM: {e}" | |
| ) from e |
| except Exception: | ||
| raise RuntimeError( | ||
| "Failed to stop PWM instance during cleanup") |
There was a problem hiding this comment.
The error message does not include the underlying exception details, making debugging difficult. Include the original exception information using 'from e' or include the exception message in the error text to provide more context about what went wrong.
| except Exception: | |
| raise RuntimeError( | |
| "Failed to stop PWM instance during cleanup") | |
| except Exception as e: | |
| raise RuntimeError( | |
| "Failed to stop PWM instance during cleanup") from e |
| except Exception: | ||
| raise RuntimeError( | ||
| "Failed to cleanup GPIO pin during cleanup") |
There was a problem hiding this comment.
The error message does not include the underlying exception details, making debugging difficult. Include the original exception information using 'from e' or include the exception message in the error text to provide more context about what went wrong.
| except Exception: | |
| raise RuntimeError( | |
| "Failed to cleanup GPIO pin during cleanup") | |
| except Exception as e: | |
| raise RuntimeError( | |
| "Failed to cleanup GPIO pin during cleanup") from e |
| attr for attr in required_attrs if not hasattr(self.gpio, attr)] | ||
| if missing: | ||
| raise RuntimeError( | ||
| f"gpio object missing attributes: {', '.join(missing)}, gpio interface must have these methods and properties to function properly") |
There was a problem hiding this comment.
The error message is a run-on sentence without proper punctuation. Add a period or semicolon after the attribute list to improve readability: 'gpio object missing attributes: {', '.join(missing)}. GPIO interface must have these methods and properties to function properly.'
| f"gpio object missing attributes: {', '.join(missing)}, gpio interface must have these methods and properties to function properly") | |
| f"gpio object missing attributes: {', '.join(missing)}. GPIO interface must have these methods and properties to function properly") |
| if self.pwm is None: | ||
| raise RuntimeError("PWM not set up. Call setup_pwm() first.") | ||
| self.duty_cycle = max(0.0, min(100.0, duty_cycle)) | ||
| print(f"[Mock] Set duty cycle to {self.duty_cycle}% on pin={self.pin}") | ||
|
|
||
| def get_duty_cycle(self) -> float: | ||
| if self.pwm is None: | ||
| raise RuntimeError("PWM not set up. Call setup_pwm() first.") |
There was a problem hiding this comment.
The RPiGPIORepository implementation (lines 59-71 in gpio.py) does not validate that PWM is set up before calling cleanup(), but the mock raises an error if PWM is not set up. This inconsistency in error handling between the real implementation and mock could hide bugs. Either add validation to RPiGPIORepository.cleanup() or remove it from the mock to maintain consistency.
| if self.pwm is None: | |
| raise RuntimeError("PWM not set up. Call setup_pwm() first.") | |
| self.duty_cycle = max(0.0, min(100.0, duty_cycle)) | |
| print(f"[Mock] Set duty cycle to {self.duty_cycle}% on pin={self.pin}") | |
| def get_duty_cycle(self) -> float: | |
| if self.pwm is None: | |
| raise RuntimeError("PWM not set up. Call setup_pwm() first.") | |
| self.duty_cycle = max(0.0, min(100.0, duty_cycle)) | |
| print(f"[Mock] Set duty cycle to {self.duty_cycle}% on pin={self.pin}") | |
| def get_duty_cycle(self) -> float: |
No description provided.