ref: adapt structure and update vs code workspace

.gitignore

@@ -1,3 +1,3 @@
-battery-health-monitor/battery_health_monitor.egg-info
+*.egg-info
 __pycache__
 .python-version

.vscode/launch.json

@@ -0,0 +1,22 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python Debugger: Current File",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${file}",
+            "console": "integratedTerminal"
+        },
+        {
+            "name": "Python: Main",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${workspaceFolder}/battery_measure_ctrl/src/main.py",
+            "console": "integratedTerminal"
+        }
+    ]
+}

README.md

@@ -36,4 +36,4 @@ python src/main.py
 
 ## Configuration
 
-The configuration for the measurement devices can be found in `src/config/config.yaml`. Modify this file to set the number of devices, cycle count, and C value for measurements.
+The configuration for the measurement devices can be found in `/config/config.yaml`. Modify this file to set the number of devices, cycle count, and C value for measurements.

battery-health-monitor/src/config/__init__.py

@@ -1,3 +0,0 @@
-# FILE: /battery-health-monitor/battery-health-monitor/src/config/__init__.py
-
-# This file is intentionally left blank.

battery-health-monitor/src/controllers/__init__.py

@@ -1,3 +0,0 @@
-# FILE: /battery-health-monitor/battery-health-monitor/src/controllers/__init__.py
-
-# This file is intentionally left blank.

battery-health-monitor/src/services/i2c_service.py

@@ -1,36 +0,0 @@
-import smbus
-import time
-
-class I2CService:
-    def __init__(self, bus_number=1):
-        self.bus = smbus.SMBus(bus_number)
-
-    def read_current(self, device_address):
-        # Replace with actual register address for current
-        current_register = 0x00
-        current = self.bus.read_word_data(device_address, current_register)
-        return self._convert_to_current(current)
-
-    def read_voltage(self, device_address):
-        # Replace with actual register address for voltage
-        voltage_register = 0x01
-        voltage = self.bus.read_word_data(device_address, voltage_register)
-        return self._convert_to_voltage(voltage)
-
-    def read_temperature(self, device_address):
-        # Replace with actual register address for temperature
-        temperature_register = 0x02
-        temperature = self.bus.read_word_data(device_address, 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
-
-    def _convert_to_temperature(self, raw_value):
-        # Conversion logic for temperature
-        return raw_value * 0.5  # Example conversion factor

battery-health-monitor/src/services/mqtt_service.py

@@ -1,32 +0,0 @@
-import paho.mqtt.client as mqtt
-import json
-
-class MQTTService:
-    def __init__(self, broker_address, topic):
-        self.broker_address = broker_address
-        self.topic = topic
-        self.client = mqtt.Client()
-
-    def connect(self):
-        self.client.connect(self.broker_address)
-
-    def on_message(self, client, userdata, message):
-        payload = json.loads(message.payload.decode())
-        self.handle_message(payload)
-
-    def handle_message(self, payload):
-        # Handle the incoming message payload
-        print(f"Received message: {payload}")
-
-    def subscribe(self):
-        self.client.subscribe(self.topic)
-        self.client.on_message = self.on_message
-
-    def publish(self, message):
-        self.client.publish(self.topic, json.dumps(message))
-
-    def loop_start(self):
-        self.client.loop_start()
-
-    def loop_stop(self):
-        self.client.loop_stop()

battery-health-monitor/src/config/config.yaml → battery_measure_ctrl/config/config.yaml

@@ -1,11 +1,17 @@
 mqtt:
-  broker: "localhost"
+  debug: true
+  broker_address: "localhost"
   port: 1883
   topic_prefix: "cells_inserted"
-  client_id: "battery_monitor"
+  client_id: "battery_measure_ctrl"
   keepalive: 60
 
+i2c:
+  debug: true
+  bus_number: 1
+
 http:
+  debug: true
   server_url: "http://localhost:8080"
   timeout: 5
   endpoint: "/cell_info"
@@ -29,4 +35,6 @@ devices:
 
 logging:
   level: "INFO"
-  file: "battery_monitor.log"
+  mode: "a" # a: append, w: overwrite
+  max_bytes: 1000000
+  file: "measure_ctrl.log"

battery-health-monitor/setup.py → battery_measure_ctrl/setup.py

@@ -1,7 +1,7 @@
 from setuptools import setup, find_packages
 
 setup(
-    name="battery-health-monitor",
+    name="battery_measure_ctrl",
     version="0.1.0",
     packages=find_packages(),
     install_requires=[
@@ -13,7 +13,7 @@ setup(
     ],
     entry_points={
         'console_scripts': [
-            'battery-monitor=src.main:main',
+            'battery_measure_ctrl=src.main:main',
         ],
     },
     python_requires='>=3.8',

battery-health-monitor/src/controllers/measurement_controller.py → battery_measure_ctrl/src/controllers/measurement_controller.py

@@ -19,7 +19,7 @@ class MeasurementController:
         self.http_service = http_service
         self.active_measurements: Dict[str, Dict[int, asyncio.Task]] = {}
         
-    async def start_measurement(self, device_id: str, slot: int, cell_id: str):
+    async def start_measurement(self, device_id: str, 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] = {}
@@ -34,7 +34,7 @@ class MeasurementController:
         )
         self.active_measurements[device_id][slot] = task
         
-    async def _measure_cycle(self, device_id: str, slot: int, cell_id: str):
+    async def _measure_cycle(self, device_id: str, slot: int, cell_id: int):
         """Execute measurement cycles for a battery."""
         try:
             # Get battery info from HTTP service
@@ -100,6 +100,6 @@ class MeasurementController:
         if not measurements:
             return False
         
-        # Add your cycle completion logic here
-        # Example: Check if voltage has reached upper limit and then lower limit
+        # TODO [SG]: Add cycle completion logic here
+        # Multiple compelte cycles or first cycle looks bad
         return len(measurements) > 100  # Simplified example

battery-health-monitor/src/main.py → battery_measure_ctrl/src/main.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import logging.handlers
 import yaml
 from pathlib import Path
 from services.mqtt_service import MQTTService
@@ -9,17 +10,29 @@ from controllers.measurement_controller import MeasurementController
 
 async def main():
     # Load config
-    config_path = Path(__file__).parent / "config" / "config.yaml"
+    config_path = Path(__file__).parent.parent / "config" / "config.yaml"
     with open(config_path) as f:
         config = yaml.safe_load(f)
 
     # Setup logging
     logging.basicConfig(
         level=getattr(logging, config['logging']['level']),
-        filename=config['logging']['file'],
-        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+        handlers=[
+            logging.handlers.RotatingFileHandler(
+                filename=config['logging']['file'], 
+                mode=config['logging']['mode'],
+                maxBytes=config['logging']['max_bytes'],
+                backupCount=2,
+                encoding=None,
+                delay=0
+            ),
+            logging.StreamHandler()
+        ]
     )
 
+    logging.info("Starting battery measurement controller...")
+
     # Initialize services
     i2c_service = I2CService(config)
     http_service = HTTPService(config)
@@ -28,7 +41,7 @@ async def main():
     # Setup MQTT client with callback
     mqtt_service = MQTTService(config)
     
-    async def on_cell_inserted(device_id: str, slot: int, cell_id: str):
+    async def on_cell_inserted(device_id: str, slot: int, cell_id: int):
         await measurement_controller.start_measurement(device_id, slot, cell_id)
 
     mqtt_service.set_callback(on_cell_inserted)

battery_measure_ctrl/src/measure_ctrl.log

@@ -0,0 +1 @@
+2025-01-31 17:24:19,325 - root - INFO - Starting battery measurement controller...

battery-health-monitor/src/services/http_service.py → battery_measure_ctrl/src/services/http_service.py

@@ -2,8 +2,10 @@ import json
 from flask import Flask, request, jsonify
 
 class HTTPService:
-    def __init__(self, app: Flask):
-        self.app = app
+    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'])
 
     def get_cell_info(self):
@@ -11,6 +13,13 @@ class HTTPService:
         if not cell_id:
             return jsonify({"error": "cell_id is required"}), 400
         
+        if self.debug:
+            return jsonify({
+                'cell_id': cell_id,
+                'capacity': 3000,
+                'manufacturer': 'MockManufacturer'
+            }), 200
+        
         # Here you would typically fetch the cell information from a database or another service
         cell_info = self.fetch_cell_info(cell_id)
         

battery_measure_ctrl/src/services/i2c_service.py

@@ -0,0 +1,46 @@
+import smbus2
+import time
+
+class I2CService:
+    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 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
+
+    def _convert_to_temperature(self, raw_value):
+        # Conversion logic for temperature
+        return raw_value * 0.5  # Example conversion factor

battery_measure_ctrl/src/services/mqtt_service.py

@@ -0,0 +1,48 @@
+import paho.mqtt.client as mqtt
+import json
+from typing import Dict, Callable
+
+class MQTTService:
+    def __init__(self, config: dict):
+        self.config = config
+        self.debug = config['mqtt'].get('debug', False)
+        self.broker_address = config['mqtt']['broker_address']
+        self.subscribe_topics = config['mqtt'].get('subscribe_topics', [])
+        self.publish_topics = config['mqtt'].get('publish_topics', [])
+        self.client = mqtt.Client()
+        self.message_handlers: Dict[str, Callable] = {}
+
+    def connect(self):
+        self.client.connect(self.broker_address)
+
+    def on_message(self, client, userdata, message):
+        topic = message.topic
+        payload = json.loads(message.payload.decode())
+        
+        if topic in self.message_handlers:
+            self.message_handlers[topic](payload)
+        else:
+            print(f"Received message on topic {topic}: {payload}")
+
+    def add_message_handler(self, topic: str, handler: Callable):
+        self.message_handlers[topic] = handler
+
+    def subscribe(self):
+        for topic in self.subscribe_topics:
+            self.client.subscribe(topic)
+        self.client.on_message = self.on_message
+
+    async def publish(self, topic: str, message: str):
+        if self.debug:
+            print(f"Debug MQTT publish to {topic}: {message}")
+            return
+        if topic in self.publish_topics:
+            self.client.publish(topic, json.dumps(message))
+        else:
+            raise ValueError(f"Topic {topic} not in configured publish topics")
+
+    def loop_start(self):
+        self.client.loop_start()
+
+    def loop_stop(self):
+        self.client.loop_stop()

battery_measure_ctrl/tests/__init__.py

@@ -0,0 +1 @@
+# This file is intentionally left blank.

measure_ctrl.code-workspace

@@ -4,5 +4,36 @@
 			"path": "."
 		}
 	],
-	"settings": {}
+	"settings": {
+		"python.analysis.extraPaths": [
+			".",
+        	"./battery_measure_ctrl/src"
+		],
+		"python.analysis.packageIndexDepths": [
+			{
+				"name": "sklearn",
+				"depth": 2
+			},
+			{
+				"name": "matplotlib",
+				"depth": 2
+			},
+			{
+				"name": "scipy",
+				"depth": 2
+			},
+			{
+				"name": "django",
+				"depth": 2
+			},
+			{
+				"name": "flask",
+				"depth": 2
+			},
+			{
+				"name": "fastapi",
+				"depth": 2
+			}
+		]
+	}
 }