After story 1
This commit is contained in:
@@ -0,0 +1,126 @@
|
||||
"""
|
||||
Configuration loader for VideoDetect.
|
||||
|
||||
Loads config.yaml with environment variable overrides.
|
||||
Provides typed accessors and validates all required fields.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import yaml
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Default config path
|
||||
DEFAULT_CONFIG_PATH = Path(__file__).parent.parent / "config.yaml"
|
||||
|
||||
|
||||
class Config:
|
||||
"""Typed configuration manager with environment variable overrides."""
|
||||
|
||||
def __init__(self, config_path: Optional[str] = None):
|
||||
self._path = Path(config_path) if config_path else DEFAULT_CONFIG_PATH
|
||||
self._data: Dict[str, Any] = {}
|
||||
self._load()
|
||||
|
||||
def _load(self):
|
||||
"""Load config from YAML file."""
|
||||
if not self._path.exists():
|
||||
raise FileNotFoundError(f"Config file not found: {self._path}")
|
||||
|
||||
with open(self._path, "r") as f:
|
||||
self._data = yaml.safe_load(f) or {}
|
||||
|
||||
# Apply environment variable overrides
|
||||
self._apply_env_overrides()
|
||||
logger.info("Config loaded from %s", self._path)
|
||||
|
||||
def _apply_env_overrides(self):
|
||||
"""Override config values with environment variables."""
|
||||
overrides = {
|
||||
"DB_HOST": ("database.host", None),
|
||||
"DB_PORT": ("database.port", int),
|
||||
"DB_NAME": ("database.name", None),
|
||||
"DB_USER": ("database.user", None),
|
||||
"DB_PASSWORD": ("database.password", None),
|
||||
"DB_POOL_SIZE": ("database.pool_size", int),
|
||||
"DB_POOL_MIN": ("database.pool_min", int),
|
||||
"GPU_MAX_MEMORY_GB": ("gpu.max_memory_gb", float),
|
||||
"SAMPLING_INTERVAL_SECONDS": ("sampling.interval_seconds", int),
|
||||
"T_HIGH": ("thresholds.T_high", float),
|
||||
"T_LOW": ("thresholds.T_low", float),
|
||||
"CONFIG_PATH": (None, None), # handled separately
|
||||
}
|
||||
|
||||
for env_key, (config_path, type_fn) in overrides.items():
|
||||
env_val = os.environ.get(env_key)
|
||||
if env_val is not None:
|
||||
if config_path is None:
|
||||
continue
|
||||
if type_fn is not None:
|
||||
env_val = type_fn(env_val)
|
||||
self._set_nested(self._data, config_path, env_val)
|
||||
logger.debug("Config override: %s=%s (from %s)", config_path, env_val, env_key)
|
||||
|
||||
@staticmethod
|
||||
def _set_nested(data: Dict, path: str, value):
|
||||
"""Set a value in a nested dict using dot notation."""
|
||||
keys = path.split(".")
|
||||
d = data
|
||||
for key in keys[:-1]:
|
||||
d = d.setdefault(key, {})
|
||||
d[keys[-1]] = value
|
||||
|
||||
def get(self, path: str, default=None):
|
||||
"""Get a config value using dot notation."""
|
||||
keys = path.split(".")
|
||||
d = self._data
|
||||
for key in keys:
|
||||
if isinstance(d, dict):
|
||||
d = d.get(key, default)
|
||||
else:
|
||||
return default
|
||||
return d if d is not None else default
|
||||
|
||||
def get_section(self, section: str) -> Dict:
|
||||
"""Get an entire config section as a dict."""
|
||||
return self._data.get(section, {})
|
||||
|
||||
def validate(self, required_keys: list) -> list:
|
||||
"""Validate that required config keys exist. Returns list of missing keys."""
|
||||
missing = []
|
||||
for key in required_keys:
|
||||
if self.get(key) is None:
|
||||
missing.append(key)
|
||||
if missing:
|
||||
logger.error("Missing required config keys: %s", missing)
|
||||
return missing
|
||||
|
||||
@property
|
||||
def data(self) -> Dict[str, Any]:
|
||||
"""Access raw config data."""
|
||||
return self._data
|
||||
|
||||
def __repr__(self):
|
||||
return f"Config(path={self._path}, sections={list(self._data.keys())})"
|
||||
|
||||
|
||||
# Module-level singleton
|
||||
_config: Optional[Config] = None
|
||||
|
||||
|
||||
def get_config(config_path: Optional[str] = None) -> Config:
|
||||
"""Get or create the global config singleton."""
|
||||
global _config
|
||||
if _config is None:
|
||||
_config = Config(config_path)
|
||||
return _config
|
||||
|
||||
|
||||
def reset_config():
|
||||
"""Reset the config singleton (useful for testing)."""
|
||||
global _config
|
||||
_config = None
|
||||
@@ -0,0 +1,137 @@
|
||||
"""
|
||||
Database connection layer with connection pooling for VideoDetect.
|
||||
|
||||
Provides:
|
||||
- Connection pooling via DBUtils + PyMySQL
|
||||
- Automatic reconnection on disconnect
|
||||
- Context manager support
|
||||
- Prepared statements for all queries
|
||||
- Transaction support for atomic state transitions
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from dbutils.pooled_db import PooledDB
|
||||
import pymysql
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DBConnector:
|
||||
"""Thread-safe database connection pool manager."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
host: str = "mariadb",
|
||||
port: int = 3306,
|
||||
database: str = "videodetect",
|
||||
user: str = "videodetect",
|
||||
password: str = "videodetect123",
|
||||
pool_size: int = 20,
|
||||
pool_min: int = 5,
|
||||
pool_recycle: int = 3600,
|
||||
):
|
||||
self._pool = PooledDB(
|
||||
creator=pymysql,
|
||||
maxconnections=pool_size,
|
||||
mincached=pool_min,
|
||||
maxcached=pool_size,
|
||||
maxusage=200,
|
||||
blocking=True,
|
||||
max_idle_time=pool_recycle,
|
||||
connection_timeout=10,
|
||||
charset="utf8mb4",
|
||||
cursorclass=pymysql.cursors.DictCursor,
|
||||
host=host,
|
||||
port=port,
|
||||
database=database,
|
||||
user=user,
|
||||
password=password,
|
||||
read_timeout=30,
|
||||
write_timeout=30,
|
||||
)
|
||||
logger.info(
|
||||
"DB pool initialized: host=%s db=%s pool_size=%d min=%d",
|
||||
host, database, pool_size, pool_min,
|
||||
)
|
||||
|
||||
def get_connection(self):
|
||||
"""Get a connection from the pool."""
|
||||
return self._pool.connection()
|
||||
|
||||
@contextlib.contextmanager
|
||||
def get_cursor(self, transaction: bool = False):
|
||||
"""Context manager for getting a cursor with optional transaction support."""
|
||||
conn = self.get_connection()
|
||||
try:
|
||||
cursor = conn.cursor()
|
||||
if transaction:
|
||||
conn.begin()
|
||||
yield cursor
|
||||
if transaction:
|
||||
conn.commit()
|
||||
except Exception:
|
||||
if transaction:
|
||||
conn.rollback()
|
||||
raise
|
||||
finally:
|
||||
cursor.close()
|
||||
conn.close()
|
||||
|
||||
@contextlib.contextmanager
|
||||
def transaction(self):
|
||||
"""Context manager for a full transaction."""
|
||||
conn = self.get_connection()
|
||||
try:
|
||||
conn.begin()
|
||||
yield conn
|
||||
conn.commit()
|
||||
except Exception:
|
||||
conn.rollback()
|
||||
raise
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def execute(self, query: str, params=None, transaction: bool = False):
|
||||
"""Execute a query and return affected rows."""
|
||||
with self.get_cursor(transaction=transaction) as cursor:
|
||||
cursor.execute(query, params or ())
|
||||
return cursor.rowcount
|
||||
|
||||
def fetchone(self, query: str, params=None):
|
||||
"""Execute a query and return one row."""
|
||||
with self.get_cursor() as cursor:
|
||||
cursor.execute(query, params or ())
|
||||
return cursor.fetchone()
|
||||
|
||||
def fetchall(self, query: str, params=None):
|
||||
"""Execute a query and return all rows."""
|
||||
with self.get_cursor() as cursor:
|
||||
cursor.execute(query, params or ())
|
||||
return cursor.fetchall()
|
||||
|
||||
def initialize_schema(self, schema_path: str = "db/schema.sql"):
|
||||
"""Initialize the database schema from SQL file."""
|
||||
with open(schema_path, "r") as f:
|
||||
sql = f.read()
|
||||
# Split on semicolons and execute each statement
|
||||
statements = [s.strip() for s in sql.split(";") if s.strip()]
|
||||
for stmt in statements:
|
||||
if stmt.startswith("--"):
|
||||
continue
|
||||
try:
|
||||
self.execute(stmt)
|
||||
except Exception as e:
|
||||
logger.debug("Schema statement skipped (may already exist): %s", e)
|
||||
logger.info("Schema initialized from %s", schema_path)
|
||||
|
||||
def health_check(self) -> bool:
|
||||
"""Check if the database is reachable."""
|
||||
try:
|
||||
result = self.fetchone("SELECT 1")
|
||||
return result is not None
|
||||
except Exception as e:
|
||||
logger.error("Health check failed: %s", e)
|
||||
return False
|
||||
@@ -0,0 +1,89 @@
|
||||
"""
|
||||
Logging configuration for VideoDetect.
|
||||
|
||||
Sets up JSON structured logging via python-json-logger with:
|
||||
- Configurable log levels
|
||||
- Standardized field names
|
||||
- Log rotation
|
||||
- All logs to stdout for Docker capture
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import logging.handlers
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from python_json_logger import json_formatter
|
||||
|
||||
|
||||
def setup_logging(
|
||||
level: str = "INFO",
|
||||
log_format: str = "json",
|
||||
rotation_max_bytes: int = 104857600, # 100MB
|
||||
rotation_backup_count: int = 10,
|
||||
):
|
||||
"""Configure structured logging for the application."""
|
||||
log_level = getattr(logging, level.upper(), logging.INFO)
|
||||
|
||||
# Root logger
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(log_level)
|
||||
|
||||
# Remove existing handlers
|
||||
root_logger.handlers.clear()
|
||||
|
||||
# Handler: stdout (for Docker capture)
|
||||
handler = logging.StreamHandler(sys.stdout)
|
||||
handler.setLevel(log_level)
|
||||
|
||||
if log_format == "json":
|
||||
formatter = JsonFormatter()
|
||||
else:
|
||||
formatter = logging.Formatter(
|
||||
"%(asctime)s %(levelname)s %(name)s %(message)s",
|
||||
datefmt="%Y-%m-%dT%H:%M:%S%z",
|
||||
)
|
||||
|
||||
handler.setFormatter(formatter)
|
||||
root_logger.addHandler(handler)
|
||||
|
||||
# Handler: file rotation (for persistence)
|
||||
file_handler = logging.handlers.RotatingFileHandler(
|
||||
filename="/logs/videodetect.log",
|
||||
maxBytes=rotation_max_bytes,
|
||||
backupCount=rotation_backup_count,
|
||||
encoding="utf-8",
|
||||
)
|
||||
file_handler.setLevel(log_level)
|
||||
file_handler.setFormatter(formatter)
|
||||
root_logger.addHandler(file_handler)
|
||||
|
||||
logging.info("Logging configured: level=%s format=%s", level, log_format)
|
||||
|
||||
|
||||
class JsonFormatter(logging.Formatter):
|
||||
"""JSON log formatter with standardized fields."""
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
log_data = {
|
||||
"timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
|
||||
"level": record.levelname,
|
||||
"logger": record.name,
|
||||
"message": record.getMessage(),
|
||||
"module": record.module,
|
||||
"function": record.funcName,
|
||||
"line": record.lineno,
|
||||
}
|
||||
|
||||
# Add extra fields
|
||||
if hasattr(record, "video_id"):
|
||||
log_data["video_id"] = record.video_id
|
||||
if hasattr(record, "metadata"):
|
||||
log_data["metadata"] = record.metadata
|
||||
|
||||
# Add exception info if present
|
||||
if record.exc_info and record.exc_info[0] is not None:
|
||||
log_data["exception"] = self.formatException(record.exc_info)
|
||||
|
||||
return json.dumps(log_data, default=str)
|
||||
+99
@@ -0,0 +1,99 @@
|
||||
"""
|
||||
VideoDetect - Video Classification System
|
||||
|
||||
Main entry point for the worker service.
|
||||
Initializes all components and starts the processing pipeline.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
|
||||
from config_loader import get_config
|
||||
from db_connector import DBConnector
|
||||
from logging_config import setup_logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def graceful_shutdown(signum, frame):
|
||||
"""Handle shutdown signals gracefully."""
|
||||
logger.info("Received signal %d, initiating graceful shutdown...", signum)
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
def main():
|
||||
"""Initialize and start the VideoDetect worker."""
|
||||
# Register signal handlers
|
||||
signal.signal(signal.SIGTERM, graceful_shutdown)
|
||||
signal.signal(signal.SIGINT, graceful_shutdown)
|
||||
|
||||
# Load configuration
|
||||
config = get_config()
|
||||
log_config = config.get_section("logging")
|
||||
setup_logging(
|
||||
level=log_config.get("level", "INFO"),
|
||||
log_format=log_config.get("format", "json"),
|
||||
rotation_max_bytes=log_config.get("rotation_max_bytes", 104857600),
|
||||
rotation_backup_count=log_config.get("rotation_backup_count", 10),
|
||||
)
|
||||
|
||||
logger.info("VideoDetect Worker starting...")
|
||||
logger.info("Config: %s", config)
|
||||
|
||||
# Initialize database connection
|
||||
db_config = config.get_section("database")
|
||||
db = DBConnector(
|
||||
host=db_config.get("host", "mariadb"),
|
||||
port=db_config.get("port", 3306),
|
||||
database=db_config.get("name", "videodetect"),
|
||||
user=db_config.get("user", "videodetect"),
|
||||
password=db_config.get("password", "videodetect123"),
|
||||
pool_size=db_config.get("pool_size", 20),
|
||||
pool_min=db_config.get("pool_min", 5),
|
||||
pool_recycle=db_config.get("pool_recycle", 3600),
|
||||
)
|
||||
|
||||
# Verify database connectivity
|
||||
if not db.health_check():
|
||||
logger.error("Cannot connect to database. Exiting.")
|
||||
sys.exit(1)
|
||||
logger.info("Database connection established.")
|
||||
|
||||
# Initialize schema if needed
|
||||
schema_path = Path(__file__).parent.parent / "db" / "schema.sql"
|
||||
if schema_path.exists():
|
||||
db.initialize_schema(str(schema_path))
|
||||
logger.info("Schema initialized.")
|
||||
|
||||
# Verify GPU availability
|
||||
try:
|
||||
import torch
|
||||
if torch.cuda.is_available():
|
||||
gpu_count = torch.cuda.device_count()
|
||||
gpu_name = torch.cuda.get_device_name(0)
|
||||
logger.info("GPU available: %d GPUs, primary: %s", gpu_count, gpu_name)
|
||||
else:
|
||||
logger.warning("CUDA is not available! Processing will be slow.")
|
||||
except ImportError:
|
||||
logger.warning("PyTorch not installed. GPU features disabled.")
|
||||
|
||||
logger.info("Worker initialization complete. Starting processing loop...")
|
||||
|
||||
# TODO: Start scanner, processor, and monitoring services
|
||||
# This is the skeleton - actual processing logic is in subsequent stories
|
||||
|
||||
try:
|
||||
while True:
|
||||
time.sleep(60) # Main loop placeholder
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Worker shutting down.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user