<filename>pymeterreader/metrics/metrics_collector.py
import logging
import typing as tp

from prometheus_client import REGISTRY, CollectorRegistry, start_wsgi_server

from pymeterreader.device_lib import BaseReader

logger = logging.getLogger(__name__)


class MetricsJiTCollector(CollectorRegistry):
    """
    Implementation of a custom collector, that spawns it´s own thread for running the webserver.
    When collect() is called Metrics are generated by retrieving Samples from all registered Readers.
    """

    def __init__(self, listen_address: str = "0.0.0.0", http_port: int = 8080, **kwargs) -> None:
        """
        Initialize Metrics JustInTime Collector
        :param listen_address: address that the server is reachable on
        :param http_port: port that the http server runs on
        """
        self.readers: tp.List[BaseReader] = []
        if kwargs:
            logger.warning(
                f'Unknown Metrics configuration parameter{"s" if len(kwargs) > 1 else ""}:{", ".join(kwargs.keys())}'
            )
        # Initialize Registry
        super().__init__(auto_describe=True, target_info=None)
        # Register as registry
        REGISTRY.register(self)
        logger.info(f"Starting Metrics HTTP Server on {listen_address}:{http_port}")
        # Start http server in separate thread
        start_wsgi_server(addr=listen_address, port=http_port)

    def collect(self) -> None:
        for reader in self.readers:
            yield from reader.metrics()

    def register_reader(self, reader: BaseReader) -> None:
        self.readers.append(reader)
        logger.debug(f"Registered {reader.PROTOCOL} Reader for meter: {reader.meter_name}")

<filename>pymeterreader/metrics/metrics_collector.py
import logging
import typing as tp

from prometheus_client import REGISTRY, CollectorRegistry, start_wsgi_server

from pymeterreader.device_lib import BaseReader

logger = logging.getLogger(__name__)


class MetricsJiTCollector(CollectorRegistry):
    """
    Implementation of a custom collector, that spawns it´s own thread for running the webserver.
    When collect() is called Metrics are generated by retrieving Samples from all registered Readers.
    """

    def __init__(self, listen_address: str = "0.0.0.0", http_port: int = 8080, **kwargs) -> None:
        """
        Initialize Metrics JustInTime Collector
        :param listen_address: address that the server is reachable on
        :param http_port: port that the http server runs on
        """
        self.readers: tp.List[BaseReader] = []
        if kwargs:
            logger.warning(
                f'Unknown Metrics configuration parameter{"s" if len(kwargs) > 1 else ""}:{", ".join(kwargs.keys())}'
            )
        # Initialize Registry
        super().__init__(auto_describe=True, target_info=None)
        # Register as registry
        REGISTRY.register(self)
        logger.info(f"Starting Metrics HTTP Server on {listen_address}:{http_port}")
        # Start http server in separate thread
        start_wsgi_server(addr=listen_address, port=http_port)

    def collect(self) -> None:
        for reader in self.readers:
            yield from reader.metrics()

    def register_reader(self, reader: BaseReader) -> None:
        self.readers.append(reader)
        logger.debug(f"Registered {reader.PROTOCOL} Reader for meter: {reader.meter_name}")

Implementation of a custom collector, that spawns it´s own thread for running the webserver.     When collect() is called Metrics are generated by retrieving Samples from all registered Readers. Initialize Metrics JustInTime Collector         :param listen_address: address that the server is reachable on         :param http_port: port that the http server runs on # Initialize Registry # Register as registry # Start http server in separate thread