From 9756793dff2ff2da0d7596b0f72afc7251d0d827 Mon Sep 17 00:00:00 2001 From: Matt Davis Date: Sat, 28 Jun 2025 09:52:05 -0400 Subject: [PATCH 1/2] Track pump activity --- src/growmax/pump.py | 5 ++ src/growmax/pump_tracker.py | 127 ++++++++++++++++++++++++++++++++++++ src/growmax/routine.py | 39 ++++++++--- 3 files changed, 162 insertions(+), 9 deletions(-) create mode 100644 src/growmax/pump_tracker.py diff --git a/src/growmax/pump.py b/src/growmax/pump.py index 8fb9d16..de21278 100644 --- a/src/growmax/pump.py +++ b/src/growmax/pump.py @@ -7,6 +7,7 @@ from growmax import constants from growmax.utils.mcu import get_gpio_for_mcu +from growmax.pump_tracker import record_pump_dose PUMP_PWM_FREQ = 10000 @@ -22,6 +23,7 @@ def __init__(self, channel=1): :param channel: One of 1, 2 or 3. """ self._speed = 0 + self._channel = channel # Store channel for pump tracking rp2040_pin = constants.PUMP_GPIOS[channel - 1] self._pin = get_gpio_for_mcu(rp2040_pin) self._gpio_pin = machine.Pin(self._pin) @@ -63,6 +65,9 @@ def dose(self, speed, timeout=0.1): """ print(f"Dose pump on GPIO pin {self._pin} at speed {speed} for {timeout} seconds.") if self.set_speed(speed): + # Record pump activity for API reporting + record_pump_dose(self._channel, speed, timeout) + time.sleep(timeout) self.stop() return True diff --git a/src/growmax/pump_tracker.py b/src/growmax/pump_tracker.py new file mode 100644 index 0000000..f7f0898 --- /dev/null +++ b/src/growmax/pump_tracker.py @@ -0,0 +1,127 @@ +""" +Pump Activity Tracker for GrowMax +Tracks pump dosing events and provides data for API reporting +""" + +import utime +from typing import List, Dict, Optional + + +class PumpActivity: + """Represents a single pump activity event""" + + def __init__(self, position: int, speed: float, duration: float, timestamp: Optional[float] = None): + self.position = position # 1-8 + self.speed = speed # 0.0 - 1.0 + self.duration = duration # seconds + self.timestamp = timestamp or utime.time() + self.enabled = True + self.description = f"Pump {position} dose at {speed*100:.0f}% for {duration}s" + + +class PumpTracker: + """Tracks pump activities for reporting to OpenSensor API""" + + def __init__(self, max_activities: int = 50): + self.activities: List[PumpActivity] = [] + self.max_activities = max_activities + self.current_session_activities: List[PumpActivity] = [] + + def record_pump_activity(self, position: int, speed: float, duration: float) -> None: + """Record a pump dosing event""" + activity = PumpActivity(position, speed, duration) + + # Add to current session (for immediate reporting) + self.current_session_activities.append(activity) + + # Add to historical activities + self.activities.append(activity) + + # Keep only the most recent activities + if len(self.activities) > self.max_activities: + self.activities = self.activities[-self.max_activities:] + + print(f"Recorded pump activity: {activity.description}") + + def get_session_activities(self) -> List[Dict]: + """Get activities from current session for API reporting""" + activities_data = [] + + for activity in self.current_session_activities: + activities_data.append({ + "position": activity.position, + "enabled": activity.enabled, + "speed": activity.speed, + "duration": activity.duration, + "timestamp": activity.timestamp, + "description": activity.description + }) + + return activities_data + + def clear_session_activities(self) -> None: + """Clear current session activities after successful API report""" + self.current_session_activities.clear() + + def get_recent_activities(self, minutes: int = 60) -> List[Dict]: + """Get activities from the last N minutes""" + cutoff_time = utime.time() - (minutes * 60) + recent_activities = [] + + for activity in self.activities: + if activity.timestamp >= cutoff_time: + recent_activities.append({ + "position": activity.position, + "enabled": activity.enabled, + "speed": activity.speed, + "duration": activity.duration, + "timestamp": activity.timestamp, + "description": activity.description + }) + + return recent_activities + + def get_pump_statistics(self) -> Dict: + """Get pump usage statistics""" + stats = { + "total_activities": len(self.activities), + "session_activities": len(self.current_session_activities), + "pump_usage": {} + } + + # Calculate per-pump statistics + for position in range(1, 9): # Pumps 1-8 + pump_activities = [a for a in self.activities if a.position == position] + total_runtime = sum(a.duration for a in pump_activities) + + stats["pump_usage"][f"pump_{position}"] = { + "activations": len(pump_activities), + "total_runtime": total_runtime, + "avg_duration": total_runtime / len(pump_activities) if pump_activities else 0 + } + + return stats + + +# Global pump tracker instance +pump_tracker = PumpTracker() + + +def record_pump_dose(position: int, speed: float, duration: float) -> None: + """Convenience function to record pump activity""" + pump_tracker.record_pump_activity(position, speed, duration) + + +def get_pump_activities_for_api() -> List[Dict]: + """Get pump activities for API reporting""" + return pump_tracker.get_session_activities() + + +def clear_reported_activities() -> None: + """Clear activities after successful API report""" + pump_tracker.clear_session_activities() + + +def get_pump_stats() -> Dict: + """Get pump usage statistics""" + return pump_tracker.get_pump_statistics() diff --git a/src/growmax/routine.py b/src/growmax/routine.py index 4bba8f5..9774082 100644 --- a/src/growmax/routine.py +++ b/src/growmax/routine.py @@ -8,6 +8,7 @@ from growmax.pump import Pump from growmax.utils import api from growmax.utils.configs import get_config_value, get_moisture_threshold_for_position +from growmax.pump_tracker import get_pump_activities_for_api, clear_reported_activities from growmax.utils.displays import boot_sequence, display_basic_stats, display_ph_reading, display_scd4x_reading from growmax.utils.mcu import get_gpio_for_mcu from growmax.utils.relays import initialize_relay_board @@ -113,18 +114,38 @@ def main(): report_data["pH"] = { "pH": ph_reading } + + # Collect pump activities for reporting + pump_activities = get_pump_activities_for_api() + relay_activities = [] + + # Add pump activities as "pumps" data + if pump_activities: + report_data["pumps"] = { + "pumps": pump_activities + } + + # Add relay activities (auto-refill) if relay_refilled: + relay_activities.append({ + "position": relay_water_position, + "enabled": True, + "seconds": relay_refill_duration, + "description": "Auto refill water reservoir" + }) + + if relay_activities: report_data["relays"] = { - "relays": [ - { - "position": relay_water_position, - "enabled": True, - "seconds": relay_refill_duration, - "description": "Auto refill water reservoir" - } - ] + "relays": relay_activities } - api.report_environment_data(report_data) + + # Report to API + try: + api.report_environment_data(report_data) + # Clear pump activities after successful report + clear_reported_activities() + except Exception as e: + print(f"Error reporting to API: {e}") if scd40x: display_scd4x_reading(temp, rh, ppm_carbon_dioxide) utime.sleep(3) From 15bb0710bf1e9496989481bdd0156f2ea014d462 Mon Sep 17 00:00:00 2001 From: Matt Davis Date: Sat, 28 Jun 2025 09:57:05 -0400 Subject: [PATCH 2/2] Track pump activity --- src/growmax/pump_tracker.py | 27 +++++++++++++-------------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/src/growmax/pump_tracker.py b/src/growmax/pump_tracker.py index f7f0898..612336c 100644 --- a/src/growmax/pump_tracker.py +++ b/src/growmax/pump_tracker.py @@ -4,13 +4,12 @@ """ import utime -from typing import List, Dict, Optional class PumpActivity: """Represents a single pump activity event""" - def __init__(self, position: int, speed: float, duration: float, timestamp: Optional[float] = None): + def __init__(self, position, speed, duration, timestamp=None): self.position = position # 1-8 self.speed = speed # 0.0 - 1.0 self.duration = duration # seconds @@ -22,12 +21,12 @@ def __init__(self, position: int, speed: float, duration: float, timestamp: Opti class PumpTracker: """Tracks pump activities for reporting to OpenSensor API""" - def __init__(self, max_activities: int = 50): - self.activities: List[PumpActivity] = [] + def __init__(self, max_activities=50): + self.activities = [] self.max_activities = max_activities - self.current_session_activities: List[PumpActivity] = [] + self.current_session_activities = [] - def record_pump_activity(self, position: int, speed: float, duration: float) -> None: + def record_pump_activity(self, position, speed, duration): """Record a pump dosing event""" activity = PumpActivity(position, speed, duration) @@ -43,7 +42,7 @@ def record_pump_activity(self, position: int, speed: float, duration: float) -> print(f"Recorded pump activity: {activity.description}") - def get_session_activities(self) -> List[Dict]: + def get_session_activities(self): """Get activities from current session for API reporting""" activities_data = [] @@ -59,11 +58,11 @@ def get_session_activities(self) -> List[Dict]: return activities_data - def clear_session_activities(self) -> None: + def clear_session_activities(self): """Clear current session activities after successful API report""" self.current_session_activities.clear() - def get_recent_activities(self, minutes: int = 60) -> List[Dict]: + def get_recent_activities(self, minutes=60): """Get activities from the last N minutes""" cutoff_time = utime.time() - (minutes * 60) recent_activities = [] @@ -81,7 +80,7 @@ def get_recent_activities(self, minutes: int = 60) -> List[Dict]: return recent_activities - def get_pump_statistics(self) -> Dict: + def get_pump_statistics(self): """Get pump usage statistics""" stats = { "total_activities": len(self.activities), @@ -107,21 +106,21 @@ def get_pump_statistics(self) -> Dict: pump_tracker = PumpTracker() -def record_pump_dose(position: int, speed: float, duration: float) -> None: +def record_pump_dose(position, speed, duration): """Convenience function to record pump activity""" pump_tracker.record_pump_activity(position, speed, duration) -def get_pump_activities_for_api() -> List[Dict]: +def get_pump_activities_for_api(): """Get pump activities for API reporting""" return pump_tracker.get_session_activities() -def clear_reported_activities() -> None: +def clear_reported_activities(): """Clear activities after successful API report""" pump_tracker.clear_session_activities() -def get_pump_stats() -> Dict: +def get_pump_stats(): """Get pump usage statistics""" return pump_tracker.get_pump_statistics()