145 lines
4.1 KiB
Python
145 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
FR-2: Audio Data Collector Service
|
|
Reads audio metrics from STM32, validates, and writes to TimescaleDB
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import signal
|
|
import sys
|
|
|
|
from serial_reader import SerialReader
|
|
from audio_validator import AudioValidator
|
|
from db_writer import DatabaseWriter
|
|
from protocol_parser import AudioMetrics
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CollectorService:
|
|
"""Main collector service orchestrating serial reading and database writing"""
|
|
|
|
def __init__(self, serial_port: str, db_url: str, baudrate: int = 115200):
|
|
self.serial_reader = SerialReader(
|
|
port=serial_port, baudrate=baudrate, on_packet=self._handle_packet
|
|
)
|
|
self.db_writer = DatabaseWriter(db_url=db_url)
|
|
self.validator = AudioValidator()
|
|
|
|
self._shutdown_event = asyncio.Event()
|
|
|
|
async def _handle_packet(self, packet: AudioMetrics):
|
|
"""
|
|
Process received audio packet: validate and write to database.
|
|
|
|
Args:
|
|
packet: Parsed audio metrics packet
|
|
"""
|
|
# Validate packet
|
|
validation = self.validator.validate_packet(packet.rms_db, packet.freq_hz)
|
|
|
|
if not validation.valid:
|
|
logger.warning(
|
|
f"Invalid packet: {validation.error} "
|
|
f"(rms={packet.rms_db:.1f}dB freq={packet.freq_hz}Hz)"
|
|
)
|
|
return
|
|
|
|
# Write to database
|
|
try:
|
|
await self.db_writer.add_record(
|
|
timestamp_ms=packet.timestamp_ms,
|
|
rms_db=packet.rms_db,
|
|
freq_hz=packet.freq_hz,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to add record to database: {e}")
|
|
|
|
async def start(self):
|
|
"""Start collector service"""
|
|
logger.info("Starting Audio Data Collector Service")
|
|
|
|
try:
|
|
# Connect to database
|
|
await self.db_writer.connect()
|
|
await self.db_writer.start_auto_flush()
|
|
|
|
# Connect to serial port
|
|
await self.serial_reader.connect()
|
|
await self.serial_reader.start_reading()
|
|
|
|
logger.info("Service started successfully")
|
|
|
|
# Wait for shutdown signal
|
|
await self._shutdown_event.wait()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Service startup failed: {e}")
|
|
raise
|
|
finally:
|
|
await self.stop()
|
|
|
|
async def stop(self):
|
|
"""Stop collector service gracefully"""
|
|
logger.info("Stopping Audio Data Collector Service")
|
|
|
|
# Disconnect serial reader
|
|
await self.serial_reader.disconnect()
|
|
|
|
# Close database writer (flushes remaining data)
|
|
await self.db_writer.close()
|
|
|
|
logger.info("Service stopped")
|
|
|
|
def shutdown(self):
|
|
"""Trigger graceful shutdown"""
|
|
logger.info("Shutdown requested")
|
|
self._shutdown_event.set()
|
|
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
# Read configuration from environment
|
|
SERIAL_PORT = os.getenv("SERIAL_PORT", "/dev/ttyACM0")
|
|
BAUDRATE = int(os.getenv("BAUDRATE", "115200"))
|
|
DB_HOST = os.getenv("DB_HOST", "localhost")
|
|
DB_PORT = os.getenv("DB_PORT", "5432")
|
|
DB_NAME = os.getenv("DB_NAME", "audio_analyzer")
|
|
DB_USER = os.getenv("DB_USER", "postgres")
|
|
DB_PASSWORD = os.getenv("DB_PASSWORD", "postgres")
|
|
|
|
db_url = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
|
|
|
|
# Create service
|
|
service = CollectorService(
|
|
serial_port=SERIAL_PORT, db_url=db_url, baudrate=BAUDRATE
|
|
)
|
|
|
|
# Setup signal handlers for graceful shutdown
|
|
loop = asyncio.get_event_loop()
|
|
for sig in (signal.SIGTERM, signal.SIGINT):
|
|
loop.add_signal_handler(sig, service.shutdown)
|
|
|
|
try:
|
|
# Run service
|
|
loop.run_until_complete(service.start())
|
|
except KeyboardInterrupt:
|
|
logger.info("Interrupted by user")
|
|
except Exception as e:
|
|
logger.error(f"Service error: {e}")
|
|
sys.exit(1)
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|