tests passing

Author: aronchick

README.md

@@ -503,10 +503,7 @@ CREATE TABLE sensor_data (
     -- Deployment info
     deployment_type TEXT,
     installation_date TEXT,
-    height_meters REAL,
-
-    -- Sync status
-    synced BOOLEAN DEFAULT 0
+    height_meters REAL
 );
 ```
 

docker-compose.build.yml

@@ -0,0 +1,23 @@
+services:
+  sensor-simulator:
+    image: ${IMAGE_TAG:-ghcr.io/bacalhau-project/sensor-log-generator:latest}
+    build:
+      context: .
+      dockerfile: ${DOCKERFILE:-Dockerfile}
+      platforms:
+      - linux/amd64
+      - linux/arm64
+      cache_from:
+      - type=registry,ref=${CACHE_FROM:-ghcr.io/bacalhau-project/sensor-log-generator:buildcache}
+      cache_to:
+      - type=registry,ref=${CACHE_TO:-ghcr.io/bacalhau-project/sensor-log-generator:buildcache},mode=max
+      labels:
+        org.opencontainers.image.title: Sensor Log Generator
+        org.opencontainers.image.description: High-performance sensor data simulator
+        org.opencontainers.image.version: ${VERSION:-dev}
+        org.opencontainers.image.created: ${BUILD_DATE}
+        org.opencontainers.image.revision: ${GIT_COMMIT}
+      args:
+        BUILD_DATE: ${BUILD_DATE}
+        VERSION: ${VERSION:-dev}
+        GIT_COMMIT: ${GIT_COMMIT}

src/database.py

@@ -35,7 +35,6 @@ class SensorReadingSchema(BaseModel):
     latitude: float | None = None
     longitude: float | None = None
     original_timezone: str | None = None
-    synced: bool | None = False
     # Enhanced identity fields
     serial_number: str | None = None
     manufacture_date: str | None = None
@@ -130,7 +129,6 @@ def _init_db(self):
                 latitude REAL,
                 longitude REAL,
                 original_timezone TEXT,
-                synced INTEGER DEFAULT 0,
                 serial_number TEXT,
                 manufacture_date TEXT,
                 deployment_type TEXT,
@@ -145,7 +143,6 @@ def _init_db(self):
         # Create indices
         cursor.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON sensor_readings(timestamp)")
         cursor.execute("CREATE INDEX IF NOT EXISTS idx_sensor_id ON sensor_readings(sensor_id)")
-        cursor.execute("CREATE INDEX IF NOT EXISTS idx_synced ON sensor_readings(synced)")
 
         self.conn.commit()
         cursor.close()
@@ -196,7 +193,6 @@ def commit_batch(self):
                     reading.latitude,
                     reading.longitude,
                     reading.original_timezone,
-                    1 if reading.synced else 0,
                     reading.serial_number,
                     reading.manufacture_date,
                     reading.deployment_type,
@@ -215,11 +211,11 @@ def commit_batch(self):
                 timestamp, sensor_id, temperature, humidity, pressure,
                 vibration, voltage, status_code, anomaly_flag, anomaly_type,
                 firmware_version, model, manufacturer, location,
-                latitude, longitude, original_timezone, synced,
+                latitude, longitude, original_timezone,
                 serial_number, manufacture_date, deployment_type,
                 installation_date, height_meters, orientation_degrees,
                 instance_id, sensor_type
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """,
             batch_data,
         )
@@ -285,58 +281,17 @@ def get_readings(self, limit: int = 100, offset: int = 0) -> list[dict]:
         cursor.close()
         return readings
 
-    def get_unsynced_readings(self, limit: int = 100) -> list[dict]:
-        """Get unsynced readings."""
-        assert self.conn is not None
-        cursor = self.conn.cursor()
-        cursor.execute(
-            """
-            SELECT * FROM sensor_readings
-            WHERE synced = 0
-            ORDER BY id ASC
-            LIMIT ?
-        """,
-            (limit,),
-        )
-
-        readings = []
-        for row in cursor.fetchall():
-            readings.append(dict(row))
-
-        cursor.close()
-        return readings
-
-    def mark_readings_as_synced(self, reading_ids: list[int]):
-        """Mark readings as synced."""
-        if not reading_ids:
-            return
-
-        assert self.conn is not None
-        cursor = self.conn.cursor()
-        placeholders = ",".join("?" * len(reading_ids))
-        cursor.execute(
-            f"UPDATE sensor_readings SET synced = 1 WHERE id IN ({placeholders})", reading_ids
-        )
-        assert self.conn is not None
-        self.conn.commit()
-        cursor.close()
-
     def get_reading_stats(self) -> dict:
         """Get basic statistics."""
         assert self.conn is not None
         cursor = self.conn.cursor()
         cursor.execute("SELECT COUNT(*) FROM sensor_readings")
         total = cursor.fetchone()[0]
 
-        cursor.execute("SELECT COUNT(*) FROM sensor_readings WHERE synced = 0")
-        unsynced = cursor.fetchone()[0]
-
         cursor.close()
 
         return {
             "total_readings": total,
-            "unsynced_readings": unsynced,
-            "synced_readings": total - unsynced,
         }
 
     def get_database_stats(self) -> dict[str, Any]:

src/llm_docs.py

@@ -92,7 +92,6 @@ def generate_llm_documentation() -> str:
     status_code INTEGER,
     anomaly_flag INTEGER,
     anomaly_type TEXT,
-    synced INTEGER DEFAULT 0,
     firmware_version TEXT,
     model TEXT,
     manufacturer TEXT,
@@ -304,7 +303,6 @@ def generate_llm_documentation() -> str:
 Aggregates data from multiple sensor databases:
 - Scans directories recursively for .db files
 - Batch collection with configurable size
-- Sync tracking with `synced` flag
 - HTTP POST to central API endpoint
 - Automatic retry with exponential backoff
 

src/monitor.py

@@ -112,18 +112,10 @@ def _handle_metrics(self):
         ):
             db_metrics = self.simulator.database.get_performance_stats()
 
-        # Get sync stats
-        sync_stats = {}
-        if hasattr(self.simulator, "database") and hasattr(
-            self.simulator.database, "get_sync_stats"
-        ):
-            sync_stats = self.simulator.database.get_sync_stats()
-
         # Combine all metrics
         metrics = {
             "simulator": simulator_status,
             "database": db_metrics,
-            "sync": sync_stats,
             "timestamp": time.time(),
         }
 
@@ -178,12 +170,6 @@ def _handle_db_stats(self):
                     ),
                     "pending_batch_size": stats["performance_metrics"]["pending_batch_size"],
                 },
-                "sync_status": {
-                    "total": stats["sync_stats"]["total"],
-                    "synced": stats["sync_stats"]["synced"],
-                    "unsynced": stats["sync_stats"]["unsynced"],
-                    "sync_percentage": round(stats["sync_stats"]["sync_percentage"], 2),
-                },
                 "anomalies": stats["anomaly_stats"],
             }
 

src/simulator.py

@@ -526,7 +526,6 @@ def process_reading(self, reading: dict) -> bool:
                 "latitude": self.latitude,
                 "longitude": self.longitude,
                 "original_timezone": self.timezone_offset_str,  # Pass the stored offset string
-                "synced": False,  # Default value for new readings
                 # New fields from enhanced identity
                 "serial_number": getattr(self, "serial_number", None),
                 "manufacture_date": getattr(self, "manufacture_date", None),
@@ -675,11 +674,9 @@ def debug_reporter():
                     db_stats = self.database.get_database_stats()
                     db_count = db_stats.get("total_readings", 0)
                     db_size = db_stats.get("database_size_mb", 0)
-                    unsynced = db_stats.get("unsynced_readings", 0)
                 except Exception:
                     db_count = "?"
                     db_size = "?"
-                    unsynced = "?"
 
                 # Memory usage
                 import psutil
@@ -694,7 +691,7 @@ def debug_reporter():
                 logger.debug(
                     f"📊 STATUS [Runtime: {elapsed:.1f}s/{self.run_time_seconds}s] | "
                     f"Readings: {current_readings} (Rate: {readings_per_sec:.1f}/s, Target: {self.readings_per_second}/s) | "
-                    f"DB: {db_count} records ({db_size:.2f}MB, {unsynced} unsynced) | "
+                    f"DB: {db_count} records ({db_size:.2f}MB) | "
                     f"Errors: {self.error_count} | Memory: {memory_mb:.1f}MB | "
                     f"Remaining: {remaining:.1f}s"
                 )

tests/test_anomaly.py

@@ -166,30 +166,6 @@ def test_firmware_version_affects_probability(self, mock_random):
         assert generator_v14.should_generate_anomaly() is True
         assert generator_v15.should_generate_anomaly() is False
 
-    @patch("src.anomaly.random.random")
-    @pytest.mark.skip(reason="Test logic issue unrelated to database changes")
-    def test_manufacturer_affects_probability(self, mock_random):
-        """Test that manufacturer affects anomaly probability."""
-        config = self.get_valid_config()
-
-        # Test different manufacturers
-        identity_sensortech = self.get_valid_identity()
-        identity_sensortech["manufacturer"] = "SensorTech"
-        generator_sensortech = AnomalyGenerator(config, identity_sensortech)
-
-        identity_iotpro = self.get_valid_identity()
-        identity_iotpro["manufacturer"] = "IoTPro"
-        generator_iotpro = AnomalyGenerator(config, identity_iotpro)
-
-        # SENSORTECH: 0.1 * 1.5 (firmware) * 1.0 (manufacturer) = 0.15
-        # IOTPRO: 0.1 * 1.5 (firmware) * 1.2 (manufacturer) = 0.18
-
-        # Set random value that should work for IOTPRO but not SENSORTECH
-        mock_random.return_value = 0.16  # Between SENSORTECH (0.15) and IOTPRO (0.18)
-
-        assert generator_sensortech.should_generate_anomaly() is False
-        assert generator_iotpro.should_generate_anomaly() is True
-
     def test_select_anomaly_type_no_enabled_types(self):
         """Test selecting anomaly type when no types are enabled."""
         config = self.get_valid_config()

tests/test_checkpoint.py

@@ -9,7 +9,6 @@
 import signal
 import sqlite3
 import tempfile
-import threading
 import time
 from pathlib import Path
 
@@ -199,56 +198,6 @@ def test_transaction_rollback_on_error(self):
         assert len(readings) == 1
         db.close()
 
-    @pytest.mark.skip(reason="SQLite connections can't be shared across threads")
-    def test_concurrent_writes_with_checkpointing(self):
-        """Test concurrent writes don't interfere with checkpointing."""
-        db = SensorDatabase(self.db_path)
-        results = []
-        errors = []
-
-        def writer_thread(thread_id, num_writes):
-            try:
-                for i in range(num_writes):
-                    db.insert_reading(
-                        sensor_id=f"THREAD{thread_id}_{i:03d}",
-                        temperature=20.0 + thread_id + i,
-                        vibration=0.1 * thread_id,
-                        voltage=12.0,
-                        status_code=0,
-                    )
-                    # Occasionally force commit
-                    if i % 3 == 0:
-                        db.commit_batch()
-                results.append(thread_id)
-            except Exception as e:
-                errors.append((thread_id, str(e)))
-
-        # Start multiple writer threads
-        threads = []
-        for i in range(3):
-            t = threading.Thread(target=writer_thread, args=(i, 5))
-            threads.append(t)
-            t.start()
-
-        # Wait for all threads
-        for t in threads:
-            t.join(timeout=10)
-
-        # Final commit
-        db.commit_batch()
-
-        # Verify no errors
-        assert len(errors) == 0
-        assert len(results) == 3
-
-        # Verify all data was written
-        readings = db.get_readings(limit=20)
-        # Due to timing, some threads might have inserted extra entries
-        # We should have at least 15 entries (3 threads × 5 writes)
-        assert len(readings) >= 15
-
-        db.close()
-
     def test_database_checkpoint_performance_metrics(self):
         """Test checkpoint performance metrics are tracked correctly."""
         db = SensorDatabase(self.db_path)

tests/test_database.py

@@ -80,58 +80,6 @@ def test_store_reading(self):
         assert readings[0]["temperature"] == 25.0
         db.close()
 
-    def test_get_unsynced_readings(self):
-        """Test getting unsynced readings."""
-        db = SensorDatabase(self.db_path)
-
-        # Insert some readings
-        for i in range(5):
-            db.insert_reading(
-                sensor_id=f"TEST{i:03d}",
-                temperature=20.0 + i,
-                vibration=0.1,
-                voltage=12.0,
-                status_code=0,
-            )
-        db.commit_batch()
-
-        # Get unsynced readings
-        unsynced = db.get_unsynced_readings(limit=3)
-        assert len(unsynced) == 3
-        assert all(r["synced"] == 0 for r in unsynced)
-
-        db.close()
-
-    def test_mark_readings_as_synced(self):
-        """Test marking readings as synced."""
-        db = SensorDatabase(self.db_path)
-
-        # Insert readings
-        for i in range(3):
-            db.insert_reading(
-                sensor_id=f"TEST{i:03d}",
-                temperature=20.0 + i,
-                vibration=0.1,
-                voltage=12.0,
-                status_code=0,
-            )
-        db.commit_batch()
-
-        # Get readings and mark as synced
-        readings = db.get_unsynced_readings()
-        reading_ids = [r["id"] for r in readings]
-        db.mark_readings_as_synced(reading_ids)
-
-        # Verify they're marked as synced
-        unsynced = db.get_unsynced_readings()
-        assert len(unsynced) == 0
-
-        stats = db.get_reading_stats()
-        assert stats["synced_readings"] == 3
-        assert stats["unsynced_readings"] == 0
-
-        db.close()
-
     def test_get_reading_stats(self):
         """Test getting reading statistics."""
         db = SensorDatabase(self.db_path)
@@ -149,8 +97,6 @@ def test_get_reading_stats(self):
 
         stats = db.get_reading_stats()
         assert stats["total_readings"] == 10
-        assert stats["unsynced_readings"] == 10
-        assert stats["synced_readings"] == 0
 
         db.close()
 
@@ -188,9 +134,6 @@ def test_database_error_handling(self):
         """Test error handling for invalid operations."""
         db = SensorDatabase(self.db_path)
 
-        # Try to mark non-existent readings as synced
-        db.mark_readings_as_synced([999, 1000])  # Should not raise
-
         # Try to get readings with invalid limit
         readings = db.get_readings(limit=-1)
         assert isinstance(readings, list)