refactor: update models for cell management; add handling of cell limits to I2C service and HTTP handling (not tested)

battery_measure_ctrl/config/config.yaml

@@ -13,12 +13,16 @@ i2c:
   debug: true
   bus_number: 1
   polling_interval_ms: 100  # How often to poll devices
+  retry_count: 3  # Number of retries for failed I2C communications
+  timeout_ms: 100  # Timeout for I2C communications
 
 http:
   debug: true
-  server_url: "http://localhost:8080"
+  server_url: "https://batteries.up-cell.de/cells/2224/"
   timeout: 5
-  endpoint: "/cell_info"
+  endpoint: "/cells"
+  username: "test"
+  password: "123"
 
 measurement:
   cycles: 3
@@ -30,10 +34,12 @@ measurement:
   rest_time_minutes: 30
 
 devices:
-  - i2c_address: 0x40
-    temp_sensors: [0x48, 0x49, 0x4A]
-  - i2c_address: 0x41
-    temp_sensors: [0x4B, 0x4C, 0x4D]
+  - id: 1
+    i2c_address: 0x40
+    num_slots: 8
+  - id: 2
+    i2c_address: 0x41
+    num_slots: 8
 
 logging:
   level: "INFO"

battery_measure_ctrl/requirements.txt

@@ -6,4 +6,5 @@ numpy==1.24.3
 pyyaml==6.0.1
 pytest==7.4.0
 pytest-asyncio==0.21.1
-Flask==2.3.3
+Flask==2.3.3
+requests==2.32.3

battery_measure_ctrl/src/controllers/measurement_controller.py

@@ -1,11 +1,10 @@
 import asyncio
 import logging
-from typing import Dict, List
-from datetime import datetime
+from typing import Dict
 import json
-from models.cell import Cell
-from models.device import Device
-from services.i2c_service import I2CService, DeviceStatus
+from models.cell import Cell, MeasureValues
+from models.device import Device, DeviceStatus, Slot
+from services.i2c_service import I2CService
 from services.http_service import HTTPService
 from services.mqtt_service import MQTTService
 
@@ -22,16 +21,13 @@ class MeasurementController:
 
         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.devices = [Device(conf['id'], conf) for conf in devices_config]
         
         self.subscribe_prefix = self.config['mqtt']['subscribe_prefix']
-        for device_id in self.devices.keys():
-            self.setup_mqtt_subscription(device_id)
+        for device in self.devices:
+            self.setup_mqtt_subscription(device.id)
 
     def setup_mqtt_subscription(self, device_id):
         """Setup MQTT subscriptions for each device."""
@@ -41,14 +37,11 @@ class MeasurementController:
     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."""
@@ -57,59 +50,51 @@ class MeasurementController:
         while True:
             await asyncio.sleep(polling_interval)
             try:
-                for device_id, device in self.devices.items():
+                for device in self.devices:
                     # Read slot status via I2C
-                    status_list = await self.i2c_service.request_status_list(device_id, slot)
-                    for idx, status in enumerate(status_list):
+                    new_status_list = await self.i2c_service.request_status_list(device.i2c_address)
+                    if len(device.status_list) != len(device.slots):
+                        raise IndexError(f"Invalid status list length: {len(device.status_list)} != {len(device.slots)}")
+                    
+                    # Change the (change of) status for each slot and act accordingly
+                    for idx, status in enumerate(new_status_list):
                         slot = device.slots[idx]
-                        prev_state = self.slot_states.get((device_id, slot))
-                        self.slot_states[(device_id, slot)] = status
+                        prev_state = device.status_list[idx]
+                        device.status_list[idx] = status
                         
-                        # Check for state transitions to "Done"
-                        if prev_state is DeviceStatus.MEASURING and status is DeviceStatus.DONE:
-                            self._process_done(device_id, slot)
+                        # Check for state transitions to "INSERTED"
+                        if status is DeviceStatus.INSERTED and prev_state is not DeviceStatus.INSERTED:
+                            self._update_cell_limits(device, slot)
                             continue
-                        if status is DeviceStatus.MEASURING:
-                            self._process_measurement(device_id, slot)
+                        # Check for state transitions to "DONE"
+                        if prev_state is DeviceStatus.MEASURING and status is DeviceStatus.DONE:
+                            self._process_done(device, slot)
                             continue
+                        # Check for state transitions to "ERROR"
                         if status is DeviceStatus.ERROR and prev_state is not DeviceStatus.ERROR:
-                            logger.error(f"Error detected for device {device_id}, slot {slot}")
+                            logger.error(f"Error detected for device {device.id}, slot {slot.id}")
+                            continue
+                        if status is DeviceStatus.MEASURING:
+                            self._collect_measurement(device, slot)
                             continue
 
             except Exception as e:
-                logger.error(f"Error during device polling: {str(e)}")
-                
+                logger.error(f"Error during device polling: {str(e)}")                
             
-    async def _collect_measurement_data(self):
+    async def _update_cell_limits(self, device: Device, slot: Slot):
+        """Send battery limits to the device."""
+        cell_id = slot.get_cell().id
+        limits = self.http_service.fetch_cell_info(cell_id)
+        self.i2c_service.send_cell_limits(device.i2c_address, slot, limits)
+
+    async def _collect_measurement(self, device: Device, slot: Slot):
         """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
+        try:
+            measure_values = await self.i2c_service.request_measure_values(device.i2c_address, slot.id)  
+            slot.get_cell().add_measurement(measure_values)
+                    
+        except Exception as e:
+            logger.error(f"Error collecting measurement data: {str(e)}")            
 
     def _handle_cell_insertion(self, client, userdata, message: str, device_id: int):
         """Handle MQTT message for cell insertion."""
@@ -123,58 +108,31 @@ class MeasurementController:
                 return
 
             # Create and schedule the measurement 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:
             logger.error(f"Invalid JSON in MQTT message: {message}")
         except Exception as e:
             logger.error(f"Error handling cell insertion: {str(e)}")
-
-    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] = {}
-            
-        # Cancel existing measurement if any
-        if slot in self.active_measurements[device_id]:
-            self.active_measurements[device_id][slot].cancel()
-            
-        logger.info(f"Starting measurement for device {device_id}, slot {slot}, cell {cell_id}")
         
-    def _process_done(self, device_id: str, slot: int, cell_id: int):
+    def _process_done(self, device: Device, slot: Slot):
         """Execute measurement cycles for a Cell."""
 
+        cell = slot.get_cell()
+        # Calculate health and capacity
+        estimated_capacity = cell.estimate_capacity()
         
-        # 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}%")
+        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}"
+        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,
+            "cell_id": cell.id,
+            "device_id": device.id,
             "slot_id": slot.id,
-            "capacity": device.get_capacity(slot),
+            "capacity": estimated_capacity,
             "status": DeviceStatus.DONE.name
         }))
-        
-        try:
-            # Get Cell info from HTTP service
-            cell_info = await self.http_service.get_cell_info(cell_id)
-            
-            measurements = []
-            cycles = self.config['measurement']['cycles']
-            sample_rate = self.config['measurement']['sample_rate_hz']
-            
-    
-        except Exception as e:
-            logger.error(f"Error during measurement: {str(e)}")
-        finally:
-            # Cleanup
-            if device_id in self.active_measurements and \
-               slot in self.active_measurements[device_id]:
-                del self.active_measurements[device_id][slot]
+        slot.remove_cell()
                 

battery_measure_ctrl/src/models/cell.py

@@ -1,16 +1,36 @@
 import logging
+from dataclasses import dataclass
 
 logger = logging.getLogger(__name__)
+
+@dataclass
+class MeasureValues:
+    voltage: int
+    current: int
+    temperature: int
+
+@dataclass
+class CellLimits:
+    min_volt: int
+    max_volt: int
+    max_curr: int
+
 class Cell():
     
-    def __init__(self, id: int, min_voltage: float, max_voltage: float, nom_capacity: float, estimated_health: float=-1.0):
+    def __init__(self, id: int, min_volt: float, max_volt: float, max_curr: float, nom_capacity: float, estimated_health: float=-1.0):
         self.id = id
-        self.min_voltage = min_voltage
-        self.max_voltage = max_voltage
+        self.limits = CellLimits(min_volt, max_volt, max_curr)
         self.nom_capacity = nom_capacity
         self.estimated_health = estimated_health # -1.0 indicates unknown health
         self.measurements = []
 
+    def add_measurement(self, data: MeasureValues):
+        """
+        Add a new measurement to the list of measurements.
+        """
+        self.measurements.append(data)
+        logger.debug(f"Added measurement for cell {self.id}: {data}")
+
     def estimate_capacity(self):
         """
         Estimate and update the state of maximum capacity of the cell based on current measurements.

battery_measure_ctrl/src/models/device.py

@@ -1,4 +1,12 @@
 from .cell import Cell
+from enum import Enum
+
+class DeviceStatus(Enum):
+    EMPTY = 0
+    INSERTED = 1
+    MEASURING = 2
+    DONE = 3
+    ERROR = 4
 
 class Device():
     def __init__(self, id: int, config: dict):
@@ -8,50 +16,27 @@ class Device():
         :param device_id: Unique identifier for the device.
         :param slots: Number of slots available in the device.
         """
-        self.id = id
-        self.i2c_address = config['i2c_address']
-        self.temp_sensors = config['temp_sensors']
-        self.slots = {Slot(id, adress) for id, adress in enumerate(self.temp_sensors)}
-
-    def read_i2c_data(self):
-        """
-        Reads data from the I2C bus for each slot.
-        This method should be implemented to interact with the I2C service.
-        """
-        pass
-
-    def update_slot(self, slot, cell_id):
-        """
-        Updates the specified slot with the given cell ID.
-
-        :param slot: The slot number to update.
-        :param cell_id: The ID of the cell inserted in the slot.
-        """
-        if slot in self.slot_data:
-            self.slot_data[slot] = cell_id
-        else:
-            raise ValueError("Invalid slot number.")
-
-    def get_slot_info(self):
-        """
-        Returns the current state of all slots in the device.
-        """
-        return self.slot_data
+        self.id: int = id
+        self.i2c_address: str = config['i2c_address']
+        self.slots = [Slot(idx) for idx in range(config['num_slots'])]
+        self.status_list = [DeviceStatus.EMPTY for _ in range(len(self.slots))]
+    
+    def get_slot_by_id(self, slot_id: int):
+        return next((slot for slot in self.slots if slot.id == slot_id), None)
     
 class Slot():
-    def __init__(self, id: int, sensor_address: int):
+    def __init__(self, id: int):
         self.id = id
-        self.cell = None
-        self.sensor_adress = sensor_address
+        self.curr_cell: Cell = None
 
     def insert_cell(self, cell: Cell):
-        self.cell = cell
+        self.curr_cell = cell
 
     def remove_cell(self):
-        self.cell = None
+        self.curr_cell = None
 
-    def is_empty(self):
-        return self.cell is None
+    def is_empty(self) -> bool:
+        return self.curr_cell is None
 
-    def get_cell(self):
-        return self.cell
+    def get_cell(self) -> Cell:
+        return self.curr_cell

battery_measure_ctrl/src/services/http_service.py

@@ -1,39 +1,62 @@
 import json
-from flask import Flask, request, jsonify
+from models.cell import CellLimits
+import requests
+import logging
+
+logger = logging.getLogger(__name__)
+
+DEBUG_DATA = {
+                "id": 2224,
+                "comment": None,
+                "start_voltage": 3.6,
+                "battery_id": 91,
+                "cell_type_id": 16,
+                "state": "Unvermessen",
+                "remaining_capacity": None,
+                "soh": None,
+                "created_at": "2025-01-04T11:53:01.820Z",
+                "updated_at": "2025-01-04T11:53:01.820Z",
+                "cell_type": {
+                    "id": 16,
+                    "name": "INR18650-35E",
+                    "comment": "",
+                    "manufacturer": "Samsung",
+                    "capacity": 3450,
+                    "nominal_voltage": 3.7,
+                    "max_voltage": 4.2,
+                    "min_voltage": 2.5,
+                    "created_at": "2024-11-05T15:39:15.107Z",
+                    "updated_at": "2024-11-05T15:39:19.564Z"
+                }
+            }        
 
 class HTTPService:
     def __init__(self, config: dict):
         self.config = config
         self.debug = config['http'].get('debug', False)
-        self.app = Flask(__name__)
-        self.app.add_url_rule('/cell_info', 'get_cell_info', self.get_cell_info, methods=['GET'])
+        self.base_url = config['http'].get('url')
+        self.endpoint = config['http'].get('endpoint')
 
-    def get_cell_info(self):
-        cell_id = request.args.get('cell_id')
-        if not cell_id:
-            return jsonify({"error": "cell_id is required"}), 400
-        
+    def fetch_cell_info(self, cell_id, username, password):      
         if self.debug:
-            return jsonify({
-                'cell_id': cell_id,
-                'capacity': 3000,
-                'manufacturer': 'MockManufacturer'
-            }), 200
+            return DEBUG_DATA
+        
+        url = f"{self.base_url}/{self.endpoint}/{cell_id}/"
         
-        # Here you would typically fetch the cell information from a database or another service
-        cell_info = self.fetch_cell_info(cell_id)
+        # Basic Authentication (if required)
+        auth = (username, password)
         
-        if cell_info:
-            return jsonify(cell_info), 200
+        # Headers
+        headers = {
+            "Accept": "application/json"
+        }
+        
+        # Making the GET request
+        response = requests.get(url, auth=auth, headers=headers)
+        
+        # Check if the request was successful
+        if response.status_code == 200:
+            return response.json()  # Return parsed JSON
         else:
-            return jsonify({"error": "Cell not found"}), 404
-
-    def fetch_cell_info(self, cell_id):
-        # Placeholder for fetching cell information logic
-        # This should be replaced with actual data retrieval logic
-        return {
-            "cell_id": cell_id,
-            "max_voltage": 4.2,
-            "min_voltage": 2.5,
-            "capacity": 2500
-        }
+            logger.error(f"Request failed with status {response.status_code}")
+            return None

battery_measure_ctrl/src/services/i2c_service.py

@@ -1,20 +1,7 @@
 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
+from models.cell import CellLimits, MeasureValues
+from models.device import DeviceStatus
 
 class I2CService:
     status_register = 0x01
@@ -28,13 +15,19 @@ class I2CService:
             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]:
+    async def request_status_list(self, i2c_adress: int) -> list[DeviceStatus]:
         """Request the status of a all slots."""
 
-        status_list = self.bus.read_block_data(device_id, self.status_register, 8)
-        return [DeviceStatus(value) for value in status_list]
+        status_list = self.bus.read_block_data(i2c_adress, self.status_register)
+        return [DeviceStatus(value) for value in status_list[:8]]
 
-    async def request_cell_values(self, device_id: int, slot: int) -> CellValues:
+    async def request_measure_values(self, i2c_adress: int, slot_id: int) -> MeasureValues:
         """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
+        measure_values:MeasureValues = self.bus.read_block_data(i2c_adress, self.cell_data_register, slot_id) # TODO [SG]: How do i specify the slot?
+        return measure_values
+    
+    async def send_cell_limits(self, i2c_adress: int, slot_id: int, limits: CellLimits) -> bool:
+        """Send the battery limits to the device."""
+        limit_list = [slot_id, limits.min_volt, limits.max_volt, limits.max_curr]
+        self.bus.write_block_data(i2c_adress, self.status_register, limit_list)
+        return True