codecarbon/output_methods/logger.py

import dataclasses
import json
import logging

from codecarbon.external.logger import logger
from codecarbon.output_methods.base_output import BaseOutput
from codecarbon.output_methods.emissions_data import EmissionsData


class LoggerOutput(BaseOutput):
    """
    Send emissions data to a logger
    """

    def __init__(self, logger, severity=logging.INFO):
        self.logger = logger
        self.logging_severity = severity

    def out(self, total: EmissionsData, delta: EmissionsData):
        try:
            payload = dataclasses.asdict(total)
            self.logger.log(self.logging_severity, msg=json.dumps(payload))
        except Exception as e:
            logger.error(e, exc_info=True)

    def live_out(self, total: EmissionsData, delta: EmissionsData):
        self.out(total, delta)


class GoogleCloudLoggerOutput(LoggerOutput):
    """
    Send emissions data to GCP Cloud Logging
    """

    def out(self, total: EmissionsData, delta: EmissionsData):
        try:
            payload = dataclasses.asdict(total)
            self.logger.log_struct(payload, severity=self.logging_severity)
        except Exception as e:
            logger.error(e, exc_info=True)

    def live_out(self, total: EmissionsData, delta: EmissionsData):
        self.out(total, delta)