Merge pull request #72 from Snailed/docs

Added docs through MKDocs, hosted on Github Pages

.github/workflows/deploy-docs.yml

@@ -0,0 +1,16 @@
+name: deploy-docs
+on:
+  push:
+    branches:
+      - main
+      - docs
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+        with:
+          python-version: 3.x
+      - run: pip install mkdocs mkdocstrings[python]
+      - run: mkdocs gh-deploy --force --clean --verbose

README.md

@@ -5,6 +5,7 @@
 [![Unit Tests](https://github.com/lfwa/carbontracker/actions/workflows/test.yml/badge.svg)](https://github.com/lfwa/carbontracker/actions)
 [![License](https://img.shields.io/github/license/lfwa/carbontracker)](https://github.com/lfwa/carbontracker/blob/master/LICENSE)
 
+[Website](https://carbontracker.info)
 
 ## About
 **carbontracker** is a tool for tracking and predicting the energy consumption and carbon footprint of training deep learning models as described in [Anthony et al. (2020)](https://arxiv.org/abs/2007.03051).

carbontracker/cli.py

@@ -3,20 +3,48 @@
 from carbontracker.tracker import CarbonTracker
 import ast
 
+
 def main():
+    """
+
+    The **carbontracker** CLI allows the user to track the energy consumption and carbon intensity of any program.
+    [Make sure that you have relevant permissions before running this.](/#permissions)
+
+    Args:
+        --log_dir (path, optional): Log directory. Defaults to `./logs`.
+        --api_keys (str, optional): API keys in a dictionary-like format, e.g. `\'{"electricitymaps": "YOUR_KEY"}\'`
+
+    Example:
+        Tracking the carbon intensity of `script.py`.
+
+            $ carbontracker python script.py
+
+        With example options
+
+            $ carbontracker --log_dir='./logs' --api_keys='{"electricitymaps": "API_KEY_EXAMPLE"}' python script.py
+
+    """
+
     # Create a parser for the known arguments
     parser = argparse.ArgumentParser(description="CarbonTracker CLI", add_help=True)
     parser.add_argument("--log_dir", type=str, default="./logs", help="Log directory")
-    parser.add_argument("--api_keys", type=str, help="API keys in a dictionary-like format, e.g., "
-                                                 "'{\"electricitymaps\": \"YOUR_KEY\"}'", default=None)
+    parser.add_argument(
+        "--api_keys",
+        type=str,
+        help="API keys in a dictionary-like format, e.g., "
+        '\'{"electricitymaps": "YOUR_KEY"}\'',
+        default=None,
+    )
 
     # Parse known arguments only
     known_args, remaining_args = parser.parse_known_args()
 
     # Parse the API keys string into a dictionary
     api_keys = ast.literal_eval(known_args.api_keys) if known_args.api_keys else None
 
-    tracker = CarbonTracker(epochs=1, log_dir=known_args.log_dir, epochs_before_pred=0, api_keys=api_keys)
+    tracker = CarbonTracker(
+        epochs=1, log_dir=known_args.log_dir, epochs_before_pred=0, api_keys=api_keys
+    )
     tracker.epoch_start()
 
     # The remaining_args are considered as the command to execute

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

@@ -3,7 +3,12 @@
 from carbontracker import exceptions
 from carbontracker.components.gpu import nvidia
 from carbontracker.components.cpu import intel
-from carbontracker.components.apple_silicon.powermetrics import AppleSiliconCPU, AppleSiliconGPU
+from carbontracker.components.apple_silicon.powermetrics import (
+    AppleSiliconCPU,
+    AppleSiliconGPU,
+)
+from carbontracker.components.handler import Handler
+from typing import Iterable, List, Union, Type, Sized
 
 COMPONENTS = [
     {
@@ -19,52 +24,60 @@
 ]
 
 
-def component_names():
+def component_names() -> List[str]:
     return [comp["name"] for comp in COMPONENTS]
 
 
-def error_by_name(name):
+def error_by_name(name) -> Exception:
     for comp in COMPONENTS:
         if comp["name"] == name:
             return comp["error"]
+    raise exceptions.ComponentNameError()
 
 
-def handlers_by_name(name):
+def handlers_by_name(name) -> List[Type[Handler]]:
     for comp in COMPONENTS:
         if comp["name"] == name:
             return comp["handlers"]
+    raise exceptions.ComponentNameError()
 
 
 class Component:
-    def __init__(self, name, pids, devices_by_pid):
+    def __init__(self, name: str, pids: Iterable[int], devices_by_pid: bool):
         self.name = name
         if name not in component_names():
-            raise exceptions.ComponentNameError(f"No component found with name '{self.name}'.")
-        self._handler = self._determine_handler(pids=pids, devices_by_pid=devices_by_pid)
-        self.power_usages = []
-        self.cur_epoch = -1  # Sentry
+            raise exceptions.ComponentNameError(
+                f"No component found with name '{self.name}'."
+            )
+        self._handler = self._determine_handler(
+            pids=pids, devices_by_pid=devices_by_pid
+        )
+        self.power_usages: List[List[float]] = []
+        self.cur_epoch: int = -1  # Sentry
 
     @property
-    def handler(self):
+    def handler(self) -> Handler:
         if self._handler is None:
             raise error_by_name(self.name)
         return self._handler
 
-    def _determine_handler(self, pids, devices_by_pid):
+    def _determine_handler(
+        self, pids: Iterable[int], devices_by_pid: bool
+    ) -> Union[Handler, None]:
         handlers = handlers_by_name(self.name)
         for h in handlers:
             handler = h(pids=pids, devices_by_pid=devices_by_pid)
             if handler.available():
                 return handler
         return None
 
-    def devices(self):
+    def devices(self) -> List[str]:
         return self.handler.devices()
 
-    def available(self):
+    def available(self) -> bool:
         return self._handler is not None
 
-    def collect_power_usage(self, epoch):
+    def collect_power_usage(self, epoch: int):
         if epoch < 1:
             return
 
@@ -77,11 +90,13 @@ def collect_power_usage(self, epoch):
             if diff != 0:
                 for _ in range(diff):
                     # Copy previous measurement lists.
-                    latest_measurements = self.power_usages[-1] if self.power_usages else []
+                    latest_measurements = (
+                        self.power_usages[-1] if self.power_usages else []
+                    )
                     self.power_usages.append(latest_measurements)
             self.power_usages.append([])
         try:
-            self.power_usages[-1].append(self.handler.power_usage())
+            self.power_usages[-1] += self.handler.power_usage()
         except exceptions.IntelRaplPermissionError:
             # Only raise error if no measurements have been collected.
             if not self.power_usages[-1]:
@@ -100,7 +115,7 @@ def collect_power_usage(self, epoch):
                 # Append zero measurement to avoid further errors.
                 self.power_usages.append([0])
 
-    def energy_usage(self, epoch_times):
+    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
@@ -138,11 +153,17 @@ def shutdown(self):
         self.handler.shutdown()
 
 
-def create_components(components, pids, devices_by_pid):
+def create_components(
+    components: str, pids: Iterable[int], devices_by_pid: bool
+) -> List[Component]:
     components = components.strip().replace(" ", "").lower()
     if components == "all":
-        return [Component(name=comp_name, pids=pids, devices_by_pid=devices_by_pid) for comp_name in component_names()]
+        return [
+            Component(name=comp_name, pids=pids, devices_by_pid=devices_by_pid)
+            for comp_name in component_names()
+        ]
     else:
         return [
-            Component(name=comp_name, pids=pids, devices_by_pid=devices_by_pid) for comp_name in components.split(",")
+            Component(name=comp_name, pids=pids, devices_by_pid=devices_by_pid)
+            for comp_name in components.split(",")
         ]