provide comments for scripts

Author: Rustam Sadykov

monitoring/build_aggregated_data.py

@@ -3,19 +3,31 @@
 from os import listdir
 from os.path import isfile, join
 from time import time
+from typing import Iterator
 
 from monitoring_settings import JSON_VERSION
 from utils import *
 
 
-def get_file_seq(input_data_dir):
+def get_file_seq(input_data_dir: str) -> Iterator[str]:
+    """
+    Get all files from specified directory
+    :param input_data_dir: path to directory with files
+    :return: sequence of filepaths
+    """
     for filename in listdir(input_data_dir):
         path = join(input_data_dir, filename)
         if isfile(path):
             yield path
 
 
-def check_stats(stats, args):
+def check_stats(stats: dict, args: argparse.Namespace) -> bool:
+    """
+    Checks timestamp and version of given statistics
+    :param stats: dictionary with statistics and metadata
+    :param args: parsed program arguments
+    :return: is timestamp and version match
+    """
     try:
         timestamp = stats["metadata"]["timestamp"]
         timestamp_match = args.timestamp_from <= timestamp <= args.timestamp_to
@@ -25,15 +37,25 @@ def check_stats(stats, args):
         return False
 
 
-def get_stats_seq(args):
+def get_stats_seq(args: argparse.Namespace) -> Iterator[dict]:
+    """
+    Get statistics with metadata matched specified period
+    :param args: parsed program arguments
+    :return: sequence of statistics with metadata filtered by version and timestamp
+    """
     for file in get_file_seq(args.input_data_dir):
         with open(file, "r") as f:
             stats = json.load(f)
         if check_stats(stats, args):
             yield stats
 
 
-def transform_target_stats(stats):
+def transform_target_stats(stats: dict) -> dict:
+    """
+    Transform metrics by computing total coverage
+    :param stats: metrics
+    :return: transformed metrics
+    """
     common_prefix = "covered_instructions"
     denum = stats["total_instructions"]
 
@@ -49,7 +71,12 @@ def transform_target_stats(stats):
     return stats
 
 
-def aggregate_stats(stats_seq):
+def aggregate_stats(stats_seq: Iterator[dict]) -> List[dict]:
+    """
+    Aggregate list of metrics and parameters into list of transformed metrics and parameters grouped by targets
+    :param stats_seq: sequence of metrics and parameters
+    :return: list of metrics and parameters grouped by targets
+    """
     result = get_default_metrics_dict()
 
     for stats in stats_seq:

monitoring/insert_metadata.py

@@ -6,27 +6,43 @@
 from os.path import exists
 from platform import uname
 from time import time
+from typing import Optional
 
 from monitoring_settings import JSON_VERSION
 from utils import *
 
 
-def load(json_file):
+def load(json_file: str) -> Optional[any]:
+    """
+    Try load object from json file
+    :param json_file: path to json file
+    :return: object from given json file or None
+    """
     if exists(json_file):
         with open(json_file, "r") as f:
             return json.load(f)
     return None
 
 
-def try_get_output(args):
+def try_get_output(args: str) -> Optional[str]:
+    """
+    Try to run subprocess with specified arguments
+    :param args: arguments for execution
+    :return: result output of execution or None
+    """
     try:
         return subprocess.check_output(args, stderr=subprocess.STDOUT, shell=True).decode()
     except Exception as e:
         print(f'Error in command "{args}":\n\t{e}')
         return None
 
 
-def parse_gradle_version(s):
+def parse_gradle_version(s: str) -> Optional[str]:
+    """
+    Parse gradle version from given string
+    :param s: execution result of gradle --version
+    :return: parsed gradle version or None
+    """
     if s is None:
         return None
     regex = re.compile(r'^\s*(Gradle [.\d]+)\s*$', re.MULTILINE)
@@ -36,7 +52,11 @@ def parse_gradle_version(s):
     return result.group(1)
 
 
-def build_environment_data():
+def build_environment_data() -> dict:
+    """
+    Collect environment data from host
+    :return: dictionary with environment data
+    """
     uname_result = uname()
     environment = {
         'host': uname_result.node,
@@ -50,7 +70,12 @@ def build_environment_data():
     return environment
 
 
-def build_metadata(args):
+def build_metadata(args: argparse.Namespace) -> dict:
+    """
+    Collect metadata into dictionary
+    :param args: parsed program arguments
+    :return: dictionary with metadata
+    """
     metadata = {
         'source': {
             'type': args.source_type,
@@ -66,7 +91,12 @@ def build_metadata(args):
     return metadata
 
 
-def build_targets(stats_array):
+def build_targets(stats_array: List[dict]) -> List[dict]:
+    """
+    Collect and group statistics by target
+    :param stats_array: list of dictionaries with parameters and metrics
+    :return: list of metrics and parameters grouped by target
+    """
     result = get_default_metrics_dict()
     for stats in stats_array:
         target = stats['parameters']['target']
@@ -76,7 +106,12 @@ def build_targets(stats_array):
     return postprocess_targets(result)
 
 
-def insert_metadata(args):
+def insert_metadata(args: argparse.Namespace) -> dict:
+    """
+    Collect metadata and statistics from specified file and merge them into result
+    :param args: parsed program arguments
+    :return: dictionary with statistics and metadata
+    """
     stats_array = load(args.stats_file)
     if stats_array is None:
         raise FileNotFoundError("File with stats does not exist!")

monitoring/utils.py

@@ -1,10 +1,16 @@
 import re
 from collections import defaultdict
+from typing import List
 
 from monitoring_settings import DEFAULT_PROJECT_VERSION
 
 
-def parse_name_and_version(full_name):
+def parse_name_and_version(full_name: str) -> tuple[str, str]:
+    """
+    Parse from string name and version of project
+    :param full_name: string with format <name>-<version>
+    :return: pair of name and version strings
+    """
     regex = re.compile(r'([^.]+)-([\d.]+)')
     result = regex.fullmatch(full_name)
     if result is None:
@@ -14,7 +20,12 @@ def parse_name_and_version(full_name):
     return name, version
 
 
-def postprocess_targets(targets):
+def postprocess_targets(targets: dict) -> List[dict]:
+    """
+    Transform dictionary with fullname target keys into array with target objects
+    :param targets: dictionary with fullname target keys
+    :return: array of targets
+    """
     result = []
     for target in targets:
         (name, version) = parse_name_and_version(target)
@@ -26,14 +37,20 @@ def postprocess_targets(targets):
     return result
 
 
-def get_default_metrics_dict():
+def get_default_metrics_dict() -> dict:
     return defaultdict(lambda: {
         'parameters': [],
         'metrics': []
     })
 
 
-def update_target(target, stats):
+def update_target(target: dict, stats: dict) -> dict:
+    """
+    Update dictionary with target by new stats
+    :param target: dictionary with target metrics and parameters
+    :param stats: new metrics and parameters
+    :return: updated target dictionary
+    """
     target['parameters'].append(stats['parameters'])
     target['metrics'].append(stats['metrics'])
     return target