fix types

fix directory mess

carbontracker/components/apple_silicon/powermetrics.py

@@ -3,43 +3,49 @@
 import re
 import time
 from carbontracker.components.handler import Handler
+from typing import Union, List, Pattern
 
 
 class PowerMetricsUnified:
-    _output = None
-    _last_updated = None
+    _output: Union[None, str] = None
+    _last_updated: Union[None, float] = None
 
     @staticmethod
     def get_output():
-        if PowerMetricsUnified._output is None or time.time() - PowerMetricsUnified._last_updated > 1:
+        if (
+            PowerMetricsUnified._output is None
+            or PowerMetricsUnified._last_updated is None
+            or time.time() - PowerMetricsUnified._last_updated > 1
+        ):
             PowerMetricsUnified._output = subprocess.check_output(
-                ["sudo", "powermetrics", "-n", "1", "-i", "1000", "--samplers", "all"], universal_newlines=True
+                ["sudo", "powermetrics", "-n", "1", "-i", "1000", "--samplers", "all"],
+                universal_newlines=True,
             )
             PowerMetricsUnified._last_updated = time.time()
         return PowerMetricsUnified._output
 
 
 class AppleSiliconCPU(Handler):
-    def init(self, pids=None, devices_by_pid=None):
+    def init(self, pids=None, devices_by_pid=False):
         self.devices_list = ["CPU"]
         self.cpu_pattern = re.compile(r"CPU Power: (\d+) mW")
 
     def shutdown(self):
         pass
 
-    def devices(self):
+    def devices(self) -> List[str]:
         """Returns a list of devices (str) associated with the component."""
         return self.devices_list
 
-    def available(self):
+    def available(self) -> bool:
         return platform.system() == "Darwin"
 
-    def power_usage(self):
+    def power_usage(self) -> List[float]:
         output = PowerMetricsUnified.get_output()
         cpu_power = self.parse_power(output, self.cpu_pattern)
-        return cpu_power
+        return [cpu_power]
 
-    def parse_power(self, output, pattern):
+    def parse_power(self, output: str, pattern: Pattern[str]) -> float:
         match = pattern.search(output)
         if match:
             power = float(match.group(1)) / 1000  # Convert mW to W
@@ -49,25 +55,25 @@ def parse_power(self, output, pattern):
 
 
 class AppleSiliconGPU(Handler):
-    def init(self, pids=None, devices_by_pid=None):
+    def init(self, pids=None, devices_by_pid=False):
         self.devices_list = ["GPU", "ANE"]
         self.gpu_pattern = re.compile(r"GPU Power: (\d+) mW")
         self.ane_pattern = re.compile(r"ANE Power: (\d+) mW")
 
-    def devices(self):
+    def devices(self) -> List[str]:
         """Returns a list of devices (str) associated with the component."""
         return self.devices_list
 
-    def available(self):
+    def available(self) -> bool:
         return platform.system() == "Darwin"
 
     def power_usage(self):
         output = PowerMetricsUnified.get_output()
         gpu_power = self.parse_power(output, self.gpu_pattern)
         ane_power = self.parse_power(output, self.ane_pattern)
-        return gpu_power + ane_power
+        return [gpu_power + ane_power]
 
-    def parse_power(self, output, pattern):
+    def parse_power(self, output: str, pattern: Pattern[str]) -> float:
         match = pattern.search(output)
         if match:
             power = float(match.group(1)) / 1000  # Convert mW to W (J/s)

carbontracker/components/component.py

@@ -8,7 +8,7 @@
     AppleSiliconGPU,
 )
 from carbontracker.components.handler import Handler
-from typing import Iterable, List, Union, Type
+from typing import Iterable, List, Union, Type, Sized
 
 COMPONENTS = [
     {
@@ -115,7 +115,7 @@ def collect_power_usage(self, epoch: int):
                 # Append zero measurement to avoid further errors.
                 self.power_usages.append([0])
 
-    def energy_usage(self, epoch_times: List[int]):
+    def energy_usage(self, epoch_times: List[int]) -> List[int]:
         """Returns energy (mWh) used by component per epoch."""
         energy_usages = []
         # We have to compute each epoch in a for loop since numpy cannot

carbontracker/components/cpu/intel.py

@@ -4,7 +4,7 @@
 
 from carbontracker import exceptions
 from carbontracker.components.handler import Handler
-from typing import List
+from typing import List, Union
 
 # RAPL Literature:
 # https://www.researchgate.net/publication/322308215_RAPL_in_Action_Experiences_in_Using_RAPL_for_Power_Measurements
@@ -20,14 +20,14 @@ def __init__(self, pids: List, devices_by_pid: bool):
         super().__init__(pids, devices_by_pid)
         self._handler = None
 
-    def devices(self):
+    def devices(self) -> List[str]:
         """Returns the name of all RAPL Domains"""
         return self._devices
 
-    def available(self):
+    def available(self) -> bool:
         return os.path.exists(RAPL_DIR) and bool(os.listdir(RAPL_DIR))
 
-    def power_usage(self):
+    def power_usage(self) -> List[float]:
         before_measures = self._get_measurements()
         time.sleep(MEASURE_DELAY)
         after_measures = self._get_measurements()
@@ -44,13 +44,13 @@ def power_usage(self):
         default = [0.0 for device in range(len(self._devices))]
         return default
 
-    def _compute_power(self, before, after):
+    def _compute_power(self, before: int, after: int) -> float:
         """Compute avg. power usage from two samples in microjoules."""
         joules = (after - before) / 1000000
         watt = joules / MEASURE_DELAY
         return watt
 
-    def _read_energy(self, path):
+    def _read_energy(self, path: str) -> int:
         with open(os.path.join(path, "energy_uj"), "r") as f:
             return int(f.read())
 
@@ -82,16 +82,16 @@ def _get_measurements(self):
 
         return measurements
 
-    def _convert_rapl_name(self, name, pattern):
+    def _convert_rapl_name(self, name, pattern) -> Union[None, str]:
         if re.match(pattern, name):
             return "cpu:" + name[-1]
 
     def init(self):
         # Get amount of intel-rapl folders
         packages = list(filter(lambda x: ":" in x, os.listdir(RAPL_DIR)))
         self.device_count = len(packages)
-        self._devices = []
-        self._rapl_devices = []
+        self._devices: List[str] = []
+        self._rapl_devices: List[str] = []
         self.parts_pattern = re.compile(r"intel-rapl:(\d):(\d)")
         devices_pattern = re.compile("intel-rapl:.")
 
@@ -101,9 +101,9 @@ def init(self):
                     name = f.read().strip()
                 if name != "psys":
                     self._rapl_devices.append(package)
-                    self._devices.append(
-                        self._convert_rapl_name(package, devices_pattern)
-                    )
+                    rapl_name = self._convert_rapl_name(package, devices_pattern)
+                    if rapl_name is not None:
+                        self._devices.append(rapl_name)
 
     def shutdown(self):
         pass

carbontracker/emissions/intensity/intensity.py

@@ -139,7 +139,7 @@ def carbon_intensity(logger, time_dur=None):
     return carbon_intensity
 
 
-def set_carbon_intensity_message(ci, time_dur):
+def set_carbon_intensity_message(ci: CarbonIntensity, time_dur):
     if ci.is_prediction:
         if ci.success:
             ci.message = (
@@ -160,4 +160,5 @@ def set_carbon_intensity_message(ci, time_dur):
             )
         else:
             ci.set_default_message()
-    ci.message += f" at detected location: {ci.address}."
+    if ci.message is not None:
+        ci.message += f" at detected location: {ci.address}."

carbontracker/loggerutil.py

@@ -1,13 +1,15 @@
 import logging
+from logging import LogRecord
 import os
 import sys
 import pathlib
 import datetime
 import importlib_metadata as metadata
 from carbontracker import constants
+from typing import Union
 
 
-def convert_to_timestring(seconds, add_milliseconds=False):
+def convert_to_timestring(seconds: int, add_milliseconds=False) -> str:
     negative = False
     if seconds < 0:
         negative = True
@@ -35,14 +37,15 @@ def convert_to_timestring(seconds, add_milliseconds=False):
 class TrackerFormatter(logging.Formatter):
     converter = datetime.datetime.fromtimestamp
 
-    def formatTime(self, record, datefmt=None):
-        ct = self.converter(record.created)
-        if datefmt:
-            s = ct.strftime(datefmt)
-        else:
-            t = ct.strftime("%Y-%m-%d %H:%M:%S")
-            s = "%s" % t
-        return s
+    def formatTime(self, record: LogRecord, datefmt: Union[str, None] = None) -> str:
+        if record.created:
+            ct = self.converter(record.created)
+            if datefmt:
+                s = ct.strftime(datefmt)
+            else:
+                t = ct.strftime("%Y-%m-%d %H:%M:%S")
+                s = "%s" % t
+            return s
 
 
 class VerboseFilter(logging.Filter):
@@ -57,7 +60,9 @@ def filter(self, record):
 class Logger:
     def __init__(self, log_dir=None, verbose=0, log_prefix=""):
         self.verbose = verbose
-        self.logger, self.logger_output, self.logger_err = self._setup(log_dir=log_dir, log_prefix=log_prefix)
+        self.logger, self.logger_output, self.logger_err = self._setup(
+            log_dir=log_dir, log_prefix=log_prefix
+        )
         self._log_initial_info()
         self.msg_prepend = "CarbonTracker: "
 
@@ -86,7 +91,9 @@ def _setup(self, log_dir=None, log_prefix=""):
 
         # Add error logging to console.
         ce = logging.StreamHandler(stream=sys.stdout)
-        ce_formatter = logging.Formatter("CarbonTracker: {levelname} - {message}", style="{")
+        ce_formatter = logging.Formatter(
+            "CarbonTracker: {levelname} - {message}", style="{"
+        )
         ce.setLevel(logging.INFO)
         ce.setFormatter(ce_formatter)
         logger_err.addHandler(ce)
@@ -103,7 +110,9 @@ def _setup(self, log_dir=None, log_prefix=""):
             f_formatter = TrackerFormatter(fmt="%(asctime)s - %(message)s")
 
             # Add output logging to file.
-            fh = logging.FileHandler(f"{log_dir}/{logger_name}_{date}_carbontracker_output.log")
+            fh = logging.FileHandler(
+                f"{log_dir}/{logger_name}_{date}_carbontracker_output.log"
+            )
             fh.setLevel(logging.INFO)
             fh.setFormatter(f_formatter)
             logger_output.addHandler(fh)
@@ -115,8 +124,12 @@ def _setup(self, log_dir=None, log_prefix=""):
             logger.addHandler(f)
 
             # Add error logging to file.
-            err_formatter = logging.Formatter("{asctime} - {threadName} - {levelname} - {message}", style="{")
-            f_err = logging.FileHandler(f"{log_dir}/{logger_name}_{date}_carbontracker_err.log", delay=True)
+            err_formatter = logging.Formatter(
+                "{asctime} - {threadName} - {levelname} - {message}", style="{"
+            )
+            f_err = logging.FileHandler(
+                f"{log_dir}/{logger_name}_{date}_carbontracker_err.log", delay=True
+            )
             f_err.setLevel(logging.DEBUG)
             f_err.setFormatter(err_formatter)
             logger_err.addHandler(f_err)

carbontracker/parser.py

@@ -4,7 +4,7 @@
 import numpy as np
 
 from carbontracker import exceptions
-from typing import Dict, Union
+from typing import Dict, Union, List
 
 
 def parse_all_logs(log_dir):
@@ -331,7 +331,7 @@ def get_all_logs(log_dir):
     return output_logs, std_logs
 
 
-def get_devices(std_log_data: str) -> Dict[str, list[str]]:
+def get_devices(std_log_data: str) -> Dict[str, List[str]]:
     """
     Retrieve dictionary of components with their device(s).
 

carbontracker/tracker.py

@@ -5,7 +5,7 @@
 import psutil
 import math
 from threading import Thread, Event
-from typing import List
+from typing import List, Union
 
 import numpy as np
 
@@ -23,11 +23,11 @@
 class CarbonIntensityThread(Thread):
     """Sleeper thread to update Carbon Intensity every 15 minutes."""
 
-    def __init__(self, logger, stop_event, update_interval=900):
+    def __init__(self, logger, stop_event, update_interval: Union[float, int] = 900):
         super(CarbonIntensityThread, self).__init__()
         self.name = "CarbonIntensityThread"
         self.logger = logger
-        self.update_interval = update_interval
+        self.update_interval: Union[float, int] = update_interval
         self.daemon = True
         self.stop_event = stop_event
         self.carbon_intensities = []
@@ -111,7 +111,7 @@ def __init__(
         logger,
         ignore_errors,
         delete,
-        update_interval=10,
+        update_interval: Union[int, float] = 10,
     ):
         super(CarbonTrackerThread, self).__init__()
         self.cur_epoch_time = time.time()