update I2C service for cell value requests and status handling; update controll logic with polling (not complete)

battery_measure_ctrl/config/config.yaml

@@ -2,7 +2,8 @@ mqtt:
   debug: true
   broker_address: "localhost"
   port: 1883
-  topic_prefix: "cells_inserted"
+  subscribe_prefix: "cells_inserted"
+  publish_prefix: "measurement_done"
   client_id: "battery_measure_ctrl"
   keepalive: 60
   username: "robot"
@@ -11,6 +12,7 @@ mqtt:
 i2c:
   debug: true
   bus_number: 1
+  polling_interval_ms: 100  # How often to poll devices
 
 http:
   debug: true

battery_measure_ctrl/src/controllers/measurement_controller.py

@@ -5,7 +5,7 @@ from datetime import datetime
 import json
 from models.cell import Cell
 from models.device import Device
-from services.i2c_service import I2CService
+from services.i2c_service import I2CService, DeviceStatus
 from services.http_service import HTTPService
 from services.mqtt_service import MQTTService
 
@@ -20,20 +20,97 @@ class MeasurementController:
         self.http_service = http_service
         self.mqtt_service = mqtt_service
 
+        self.polling_task = None
+        self.measurement_data_task = None
+        self.slot_states = {}  # Track states of all slots
+
         self.active_measurements: Dict[str, Dict[int, asyncio.Task]] = {}
 
         devices_config = config['devices']
         self.devices = {id: Device(id, config) for id, config in enumerate(devices_config)}
         
-        self.topic_prefix = self.config['mqtt']['topic_prefix']
+        self.subscribe_prefix = self.config['mqtt']['subscribe_prefix']
         for device_id in self.devices.keys():
             self.setup_mqtt_subscription(device_id)
 
     def setup_mqtt_subscription(self, device_id):
         """Setup MQTT subscriptions for each device."""
-        topic = f"{self.topic_prefix}/{device_id}"
+        topic = f"{self.subscribe_prefix}/device_{device_id}"
         self.mqtt_service.add_message_handler(topic, lambda client, userdata, msg, dev_id=device_id: self._handle_cell_insertion(client, userdata, msg, dev_id))
 
+    async def start_polling(self):
+        """Start the polling tasks."""
+        self.polling_task = asyncio.create_task(self._poll_devices())
+        self.measurement_data_task = asyncio.create_task(self._collect_measurement_data())
+
+    async def stop_polling(self):
+        """Stop the polling tasks."""
+        if self.polling_task:
+            self.polling_task.cancel()
+        if self.measurement_data_task:
+            self.measurement_data_task.cancel()
+            
+    async def _poll_devices(self):
+        """Continuously poll all devices for slot status."""
+        polling_interval = self.config['i2c']['polling_interval_ms'] / 1000.0  # Convert to seconds
+        
+        while True:
+            await asyncio.sleep(polling_interval)
+            try:
+                for device_id, device in self.devices.items():
+                    # Read slot status via I2C
+                    status_list = await self.i2c_service.request_status_list(device_id, slot)
+                    for idx, status in enumerate(status_list):
+                        slot = device.slots[idx]
+                        prev_state = self.slot_states.get((device_id, slot))
+                        self.slot_states[(device_id, slot)] = status
+                        
+                        # Check for state transitions to "Done"
+                        if prev_state is DeviceStatus.MEASURING and status is DeviceStatus.DONE:
+                            self._process_done(device_id, slot)
+                            continue
+                        if status is DeviceStatus.MEASURING:
+                            self._process_measurement(device_id, slot)
+                            continue
+                        if status is DeviceStatus.ERROR and prev_state is not DeviceStatus.ERROR:
+                            logger.error(f"Error detected for device {device_id}, slot {slot}")
+                            continue
+
+            except Exception as e:
+                logger.error(f"Error during device polling: {str(e)}")
+                
+            
+    async def _collect_measurement_data(self):
+        """Collect measurement data from active slots."""
+        measurement_interval = self.config['i2c']['measurement_data_interval_ms'] / 1000.0  # Convert to seconds
+        
+        while True:
+            try:
+                for (device_id, slot), status in self.slot_states.items():
+                    if status == "MEASURING":
+                        # Collect measurement data
+                        voltage = await self.i2c_service.read_voltage(device_id, slot)
+                        current = await self.i2c_service.read_current(device_id, slot)
+                        temp = await self.i2c_service.read_temperature(device_id, slot)
+                        
+                        # Store or process the measurement data
+                        await self._process_measurement_data(device_id, slot, {
+                            "voltage": voltage,
+                            "current": current,
+                            "temperature": temp,
+                            "timestamp": datetime.now().isoformat()
+                        })
+                        
+            except Exception as e:
+                logger.error(f"Error collecting measurement data: {str(e)}")
+                
+            await asyncio.sleep(measurement_interval)
+            
+    async def _process_measurement_data(self, device_id: int, slot: int, data: dict):
+        """Process measurement data - implement your data handling logic here."""
+        # Add to measurements list, save to database, etc.
+        pass
+
     def _handle_cell_insertion(self, client, userdata, message: str, device_id: int):
         """Handle MQTT message for cell insertion."""
         try:
@@ -46,7 +123,7 @@ class MeasurementController:
                 return
 
             # Create and schedule the measurement task
-            asyncio.create_task(self.start_measurement(device_id, slot, cell_id))
+            self.start_measurement(device_id, slot, cell_id)
             logger.info(f"Initiated measurement for device {device_id}, slot {slot}, cell {cell_id}")
             
         except json.JSONDecodeError:
@@ -54,7 +131,7 @@ class MeasurementController:
         except Exception as e:
             logger.error(f"Error handling cell insertion: {str(e)}")
 
-    async def start_measurement(self, device_id: int, slot: int, cell_id: int):
+    def start_measurement(self, device_id: int, slot: int, cell_id: int):
         """Start measurement cycle for a specific slot."""
         if device_id not in self.active_measurements:
             self.active_measurements[device_id] = {}
@@ -63,14 +140,27 @@ class MeasurementController:
         if slot in self.active_measurements[device_id]:
             self.active_measurements[device_id][slot].cancel()
             
-        # Create new measurement task
-        task = asyncio.create_task(
-            self._measure_cycle(device_id, slot, cell_id)
-        )
-        self.active_measurements[device_id][slot] = task
+        logger.info(f"Starting measurement for device {device_id}, slot {slot}, cell {cell_id}")
         
-    async def _measure_cycle(self, device_id: str, slot: int, cell_id: int):
+    def _process_done(self, device_id: str, slot: int, cell_id: int):
         """Execute measurement cycles for a Cell."""
+
+        
+        # Calculate health
+        estimated_capacity = Cell.estimate_capacity([m['current'] for m in measurements])
+        
+        logger.info(f"Measurement complete for cell {cell_id}. Estimated capacity: {estimated_capacity}%")
+
+        # Publish completion message
+        topic = f"{self.config['mqtt']['measurement_done_topic']}/device_{device_id}/slot_{slot}"
+        self.mqtt_service.publish(topic, json.dumps({
+            "cell_id": device.get_cell_id(slot),
+            "device_id": device_id,
+            "slot_id": slot.id,
+            "capacity": device.get_capacity(slot),
+            "status": DeviceStatus.DONE.name
+        }))
+        
         try:
             # Get Cell info from HTTP service
             cell_info = await self.http_service.get_cell_info(cell_id)
@@ -79,43 +169,7 @@ class MeasurementController:
             cycles = self.config['measurement']['cycles']
             sample_rate = self.config['measurement']['sample_rate_hz']
             
-            for cycle in range(cycles):
-                logger.info(f"Starting cycle {cycle+1}/{cycles} for cell {cell_id}")
-                cycle_measurements = []
-                
-                while True:
-                    # Read measurements
-                    voltage = await self.i2c_service.read_voltage(device_id, slot)
-                    current = await self.i2c_service.read_current(device_id, slot)
-                    temp = await self.i2c_service.read_temperature(device_id, slot)
-                    
-                    # Check safety limits
-                    if not self._check_safety_limits(voltage, temp):
-                        raise Exception("Safety limits exceeded")
-                    
-                    cycle_measurements.append({
-                        'timestamp': datetime.now().isoformat(),
-                        'voltage': voltage,
-                        'current': current,
-                        'temperature': temp
-                    })
-                    
-                    # Check if cycle complete
-                    if self._is_cycle_complete(cycle_measurements):
-                        break
-                        
-                    await asyncio.sleep(1/sample_rate)
-                
-                measurements.extend(cycle_measurements)
-                
-                # Rest period between cycles
-                await asyncio.sleep(self.config['measurement']['rest_time_minutes'] * 60)
-            
-            # Calculate health
-            health = calculate_health([m['current'] for m in measurements])
-            
-            logger.info(f"Measurement complete for cell {cell_id}. Health: {health}%")
-            
+    
         except Exception as e:
             logger.error(f"Error during measurement: {str(e)}")
         finally:
@@ -123,18 +177,4 @@ class MeasurementController:
             if device_id in self.active_measurements and \
                slot in self.active_measurements[device_id]:
                 del self.active_measurements[device_id][slot]
-                
-    def _check_safety_limits(self, voltage: float, temperature: float) -> bool:
-        """Check if measurements are within safety limits."""
-        return (self.config['measurement']['min_voltage'] <= voltage <= 
-                self.config['measurement']['max_voltage'] and
-                temperature <= self.config['measurement']['max_temperature_c'])
-        
-    def _is_cycle_complete(self, measurements: List[dict]) -> bool:
-        """Determine if a measurement cycle is complete."""
-        if not measurements:
-            return False
-        
-        # TODO [SG]: Add cycle completion logic here
-        # Multiple compelte cycles or first cycle looks bad
-        return len(measurements) > 100  # Simplified example
+                

battery_measure_ctrl/src/main.py

@@ -37,20 +37,20 @@ async def main():
     i2c_service = I2CService(config)
     mqtt_service = MQTTService(config)
     http_service = HTTPService(config)
-    MeasurementController(config, i2c_service, http_service, mqtt_service)
+    controller = MeasurementController(config, i2c_service, http_service, mqtt_service)
 
     mqtt_service.connect()
-    mqtt_service.subscribe_defined()
     mqtt_service.loop_start()
 
+    await controller.start_polling()
+
     try:
         # Keep running until interrupted
         while True:
             await asyncio.sleep(1)
-    except KeyboardInterrupt:
-        logging.info("Shutting down...")
     finally:
-        await mqtt_service.disconnect()
+        logging.info("Shutting down...")
+        mqtt_service.disconnect()
 
 if __name__ == "__main__":
     asyncio.run(main())

battery_measure_ctrl/src/models/cell.py

@@ -11,14 +11,15 @@ class Cell():
         self.estimated_health = estimated_health # -1.0 indicates unknown health
         self.measurements = []
 
-    def estimate_health(self):
+    def estimate_capacity(self):
         """
-        Estimate and update the state of health of the cell based on current measurements.
+        Estimate and update the state of maximum capacity of the cell based on current measurements.
         """
-        # TODO [SG]: Placeholder for health estimation logic
+        # TODO [SG]: Placeholder for capacity estimation logic
         if not self.measurements:
-            logger.warning("No measurements available for health estimation.")
+            logger.warning("No measurements available for capacity estimation.")
             return
         
         avg_current = sum(self.measurements) / float(len(self.measurements))
-        self.health = max(0.0, min(100.0, (self.nom_capacity - avg_current) / self.nom_capacity * 100.0))
+        self.capacity = max(0.0, min(100.0, (self.nom_capacity - avg_current) / self.nom_capacity * 100.0))
+        return self.capacity

battery_measure_ctrl/src/services/i2c_service.py

@@ -1,46 +1,40 @@
 import smbus2
 import time
+from enum import Enum
+from dataclasses import dataclass
+
+@dataclass
+class CellValues:
+    voltage: int
+    current: int
+    temperature: int
+
+class DeviceStatus(Enum):
+    EMPTY = 0
+    INSERTED = 1
+    MEASURING = 2
+    DONE = 3
+    ERROR = 4
 
 class I2CService:
+    status_register = 0x01
+    cell_data_register = 0x02
+    battery_limit_register = 0x03
+
     def __init__(self, config: dict):
         self.config = config
         self.debug = config['i2c'].get('debug', False)
         if not self.debug:
             bus_number = config.get('i2c', {}).get('bus', 1)
             self.bus = smbus2.SMBus(bus_number)
+    
+    async def request_status_list(self, device_id: int, slot: int) -> list[DeviceStatus]:
+        """Request the status of a all slots."""
 
-    async def read_current(self, device_id: str, slot: int) -> float:
-        if self.debug:
-            return 1.0  # Return a mock current value
-        # Replace with actual register address for current
-        current_register = 0x00
-        current = self.bus.read_word_data(device_id, current_register)
-        return self._convert_to_current(current)
-
-    async def read_voltage(self, device_id: str, slot: int) -> float:
-        if self.debug:
-            return 3.7  # Return a mock voltage value
-        # Replace with actual register address for voltage
-        voltage_register = 0x01
-        voltage = self.bus.read_word_data(device_id, voltage_register)
-        return self._convert_to_voltage(voltage)
-
-    async def read_temperature(self, device_id: str, slot: int) -> float:
-        if self.debug:
-            return 25.0  # Return a mock temperature value
-        # Replace with actual register address for temperature
-        temperature_register = 0x02
-        temperature = self.bus.read_word_data(device_id, temperature_register)
-        return self._convert_to_temperature(temperature)
-
-    def _convert_to_current(self, raw_value):
-        # Conversion logic for current
-        return raw_value * 0.1  # Example conversion factor
-
-    def _convert_to_voltage(self, raw_value):
-        # Conversion logic for voltage
-        return raw_value * 0.01  # Example conversion factor
+        status_list = self.bus.read_block_data(device_id, self.status_register, 8)
+        return [DeviceStatus(value) for value in status_list]
 
-    def _convert_to_temperature(self, raw_value):
-        # Conversion logic for temperature
-        return raw_value * 0.5  # Example conversion factor
+    async def request_cell_values(self, device_id: int, slot: int) -> CellValues:
+        """Request the cell values of a specific slot."""
+        cell_values:CellValues = self.bus.read_block_data(device_id, self.cell_data_register, 8) # TODO [SG]: How do i specify the slot?
+        return cell_values

battery_measure_ctrl/src/services/mqtt_service.py

@@ -57,7 +57,7 @@ class MQTTService:
             self.client.subscribe(topic)
             self.client.message_callback_add(topic, callback)
 
-    async def publish(self, topic: str, message: str):
+    def publish(self, topic: str, message: str):
         # if self.debug:
         #     logger.info(f"Debug MQTT publish to {topic}: {message}")
         #     return