DB Writer fixes and compose.yml update

infrastructure/compose.yml

@@ -15,7 +15,7 @@ services:
       - "9001:9001"
       - "8883:8883"
     volumes:
-      - ./mosquitto/:/mosquitto/
+      - ./mosquitto/:/mosquitto/:Z
     restart: unless-stopped
   
   timescaledb:

services/db_write/src/db_writer.py

@@ -6,6 +6,7 @@ from sqlalchemy.pool import QueuePool
 
 from src.config import config
 from src.schema import TelemetryReading
+from src.models import Telemetry
 
 
 class DatabaseWriter:
@@ -37,9 +38,9 @@ class DatabaseWriter:
 
         session = self.SessionLocal()
         try:
-            # Convert to database objects using the correct field mapping
+            # Convert dataclass readings to SQLAlchemy Telemetry objects
             db_objects = [
-                TelemetryReading(
+                Telemetry(
                     time=reading.time,
                     device_id=reading.device_id,
                     metric=reading.metric,
@@ -57,7 +58,7 @@ class DatabaseWriter:
             return True
 
         except Exception as e:
-            self.logger.error(f"Failed to write batch: {e}")
+            self.logger.error(f"Failed to write batch: {e}", exc_info=True)
             session.rollback()
             return False
         finally:

services/db_write/src/models.py

@@ -0,0 +1,30 @@
+"""
+SQLAlchemy models for db_write service.
+
+These models mirror the schema in db_migrations/models.py.
+Keep them in sync when schema changes occur.
+"""
+
+from sqlalchemy import Column, Float, Text, DateTime
+from sqlalchemy.ext.declarative import declarative_base
+
+Base = declarative_base()
+
+
+class Telemetry(Base):
+    """
+    Time-series telemetry data from devices.
+    
+    This model is used by the db_write service to insert data.
+    """
+
+    __tablename__ = "telemetry"
+
+    time = Column(DateTime(timezone=True), primary_key=True, nullable=False)
+    device_id = Column(Text, primary_key=True, nullable=False)
+    metric = Column(Text, primary_key=True, nullable=False)
+    value = Column(Float, nullable=False)
+    unit = Column(Text)
+
+    def __repr__(self):
+        return f"<Telemetry(device={self.device_id}, metric={self.metric}, value={self.value})>"

services/gpt_service/config.py

@@ -3,11 +3,8 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
-class Config:
-    """Configuration settings for the GPT Service."""
-
-    API_KEY = os.getenv("API_KEY")
-    PROVIDER_NAME = os.getenv("PROVIDER_NAME", "openai")
-    MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4")
-    HOST_URL = os.getenv("HOST_URL")
-    LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
+API_KEY = os.getenv("API_KEY")
+PROVIDER_NAME = os.getenv("PROVIDER_NAME", "openai")
+MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4")
+HOST_URL = os.getenv("HOST_URL")
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")

services/gpt_service/gpt_service.py

@@ -22,21 +22,4 @@ class GPTService:
             self.logger.error(f"Unsupported provider: {self.provider_name}")
             raise ValueError(f"Unsupported provider: {self.provider_name}")
 
-    def analyze_metrics(self, metrics: dict) -> str:
-        """Analyze given metrics using GPT model and return insights."""
-        prompt = f"Analyze the following metrics and provide insights:\n{metrics}"
-        try:
-            response = openai.Completion.create(
-                engine=self.model_name,
-                prompt=prompt,
-                max_tokens=150,
-                n=1,
-                stop=None,
-                temperature=0.7,
-            )
-            insights = response.choices[0].text.strip()
-            self.logger.info("Successfully obtained insights from GPT model")
-            return insights
-        except Exception as e:
-            self.logger.error(f"Error during GPT analysis: {e}")
-            raise
+        

services/gpt_service/main.py

@@ -1,4 +1,8 @@
+from gpt_service import GPTService
+
+
 def main():
+    GPTService()
     print("Hello from gpt-service!")