Lab_temp_hum_monitor/logging_setup.py at main · Aramushaa/Lab_temp_hum_monitor · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
"""Central logging configuration for the IoT backend.

Routes sensor, actuator, and control logs to separate files while keeping stdout output.
"""

from __future__ import annotations

import logging
import os
from logging.config import dictConfig
from typing import Any, Dict

_CONFIGURED = False


def configure_logging() -> None:
    """Set up log handlers only once."""
    global _CONFIGURED
    if _CONFIGURED:
        return

    log_dir = os.environ.get("LOG_DIR", "/tmp/logs")
    try:
        os.makedirs(log_dir, exist_ok=True)
    except Exception:
        log_dir = "/tmp"
        os.makedirs(log_dir, exist_ok=True)

    def _file(name: str) -> Dict[str, Any]:
        return {
            "class": "logging.handlers.RotatingFileHandler",
            "filename": os.path.join(log_dir, f"{name}.log"),
            "maxBytes": 1_000_000,
            "backupCount": 3,
            "encoding": "utf-8",
            "formatter": "detailed",
        }

    config = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "simple": {
                "format": "%(asctime)s %(levelname)s %(name)s: %(message)s",
            },
            "detailed": {
                "format": "%(asctime)s %(levelname)s %(name)s [%(filename)s:%(lineno)d]: %(message)s",
            },
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "formatter": "simple",
                "level": os.environ.get("LOG_LEVEL", "WARNING"),
            },
            "sensor_file": _file("sensors"),
            "actuator_file": _file("actuators"),
            "logic_file": _file("logic"),
        },
        "root": {
            "handlers": ["console"],
            "level": os.environ.get("LOG_LEVEL_ROOT", "WARNING"),
        },
        "loggers": {
            # Sensor pipeline
            "Device_connectors.sensor_bridge": {
                "handlers": ["sensor_file", "console"],
                "level": "INFO",
                "propagate": False,
            },
            "controller.state_memory": {
                "handlers": ["sensor_file", "console"],
                "level": "INFO",
                "propagate": False,
            },
            # Actuator pipeline
            "Device_connectors.actuator_bridge": {
                "handlers": ["actuator_file", "console"],
                "level": "INFO",
                "propagate": False,
            },
            "controller.rules": {
                "handlers": ["logic_file", "console"],
                "level": "INFO",
                "propagate": False,
            },
            "controller.control_unit": {
                "handlers": ["logic_file", "console"],
                "level": "INFO",
                "propagate": False,
            },
            "controller.cu_instancer": {
                "handlers": ["logic_file", "console"],
                "level": "INFO",
                "propagate": False,
            },
            "Device_connectors.mqtt_client": {
                "handlers": ["logic_file", "console"],
                "level": "INFO",
                "propagate": False,
            },
        },
    }

    dictConfig(config)
    _CONFIGURED = True