refactored code

Signed-off-by: Shashank Reddy Boyapally <[email protected]>

Author: shashank-boyapally

orion.py

@@ -6,13 +6,13 @@
 import logging
 import sys
 import warnings
+from typing import Any
 import click
 import uvicorn
 from fmatch.logrus import SingletonLogger
 from pkg.runTest import run
 from pkg.utils import load_config
 import pkg.constants as cnsts
-from typing import Any
 
 warnings.filterwarnings("ignore", message="Unverified HTTPS request.*")
 warnings.filterwarnings(
@@ -104,12 +104,8 @@ def cli(max_content_width=120):  # pylint: disable=unused-argument
 )
 @click.option("--lookback", help="Get data from last X days and Y hours. Format in XdYh")
 @click.option("--convert-tinyurl", is_flag=True, help="Convert buildUrls to tiny url format for better formatting")
-<<<<<<< HEAD
 @click.option("--collapse", is_flag=True, help="Only outputs changepoints, previous and later runs in the xml format")
 def cmd_analysis(**kwargs):
-=======
-def cmd_analysis(**kwargs: dict[str, dict]) -> None:
->>>>>>> 3488c73 (added typing)
     """
     Orion runs on command line mode, and helps in detecting regressions
     """

pkg/algorithm.py

@@ -0,0 +1,8 @@
+"""
+Init for pkg module
+"""
+
+from .edivisive.edivisive import EDivisive
+from .isolationforest.isolationForest import IsolationForestWeightedMean
+from .algorithmFactory import AlgorithmFactory
+from .algorithm import Algorithm

pkg/algorithms/__init__.py

@@ -0,0 +1,177 @@
+"""Module for Generic Algorithm class"""
+
+from abc import ABC, abstractmethod
+from itertools import groupby
+import json
+from typing import Any, Dict, List, Tuple, Union
+import pandas as pd
+from fmatch.matcher import Matcher
+from hunter.report import Report, ReportType
+from hunter.series import Series, Metric, ChangePoint, ChangePointGroup
+import pkg.constants as cnsts
+
+
+from pkg.utils import json_to_junit
+
+
+class Algorithm(ABC):
+    """Generic Algorithm class for algorithm factory"""
+
+    def __init__(  # pylint: disable = too-many-arguments
+        self,
+        matcher: Matcher,
+        dataframe: pd.DataFrame,
+        test: dict,
+        options: dict,
+        metrics_config: dict[str, dict],
+    ) -> None:
+        self.matcher = matcher
+        self.dataframe = dataframe
+        self.test = test
+        self.options = options
+        self.metrics_config = metrics_config
+
+    def output_json(self) -> Tuple[str, str]:
+        """Method to output json output
+
+        Returns:
+            Tuple[str, str]: returns test_name and json output
+        """
+        _, change_points_by_metric = self._analyze()
+        dataframe_json = self.dataframe.to_json(orient="records")
+        dataframe_json = json.loads(dataframe_json)
+
+        for index, entry in enumerate(dataframe_json):
+            entry["metrics"] = {
+                key: {"value": entry.pop(key), "percentage_change": 0}
+                for key in self.metrics_config
+            }
+            entry["is_changepoint"] = False
+
+        for key, value in change_points_by_metric.items():
+            for change_point in value:
+                index = change_point.index
+                percentage_change = (
+                    (change_point.stats.mean_2 - change_point.stats.mean_1)
+                    / change_point.stats.mean_1
+                ) * 100
+                if (
+                    percentage_change * self.metrics_config[key]["direction"] > 0
+                    or self.metrics_config[key]["direction"] == 0
+                ):
+                    dataframe_json[index]["metrics"][key][
+                        "percentage_change"
+                    ] = percentage_change
+                    dataframe_json[index]["is_changepoint"] = True
+
+        return self.test["name"], json.dumps(dataframe_json, indent=2)
+
+    def output_text(self) -> Tuple[str,str]:
+        """Outputs the data in text/tabular format"""
+        series, change_points_by_metric = self._analyze()
+        change_points_by_time = self.group_change_points_by_time(
+            series, change_points_by_metric
+        )
+        report = Report(series, change_points_by_time)
+        output_table = report.produce_report(
+            test_name=self.test["name"], report_type=ReportType.LOG
+        )
+        return self.test["name"], output_table
+
+    def output_junit(self) -> Tuple[str,str]:
+        """Output junit format
+
+        Returns:
+            _type_: return
+        """
+        test_name, data_json = self.output_json()
+        data_json = json.loads(data_json)
+        data_junit = json_to_junit(
+            test_name=test_name,
+            data_json=data_json,
+            metrics_config=self.metrics_config,
+            options=self.options,
+        )
+        return test_name, data_junit
+
+    @abstractmethod
+    def _analyze(self):
+        """Analyze algorithm"""
+
+    def group_change_points_by_time(
+        self, series: Series, change_points: Dict[str, List[ChangePoint]]
+    ) -> List[ChangePointGroup]:
+        """Return changepoint by time
+
+        Args:
+            series (Series): Series of data
+            change_points (Dict[str, List[ChangePoint]]): Group of changepoints wrt time
+
+        Returns:
+            List[ChangePointGroup]: _description_
+        """
+        changes: List[ChangePoint] = []
+        for metric in change_points.keys():
+            changes += change_points[metric]
+
+        changes.sort(key=lambda c: c.index)
+        points = []
+        for k, g in groupby(changes, key=lambda c: c.index):
+            cp = ChangePointGroup(
+                index=k,
+                time=series.time[k],
+                prev_time=series.time[k - 1],
+                attributes=series.attributes_at(k),
+                prev_attributes=series.attributes_at(k - 1),
+                changes=list(g),
+            )
+            points.append(cp)
+
+        return points
+
+    def setup_series(self) -> Series:
+        """
+        Returns series
+        Returns:
+            _type_: _description_
+        """
+        metrics = {
+            column: Metric(value.get("direction", 1), 1.0)
+            for column, value in self.metrics_config.items()
+        }
+        data = {column: self.dataframe[column] for column in self.metrics_config}
+        attributes = {
+            column: self.dataframe[column]
+            for column in self.dataframe.columns
+            if column in ["uuid", "buildUrl"]
+        }
+        series = Series(
+            test_name=self.test["name"],
+            branch=None,
+            time=list(self.dataframe["timestamp"]),
+            metrics=metrics,
+            data=data,
+            attributes=attributes,
+        )
+
+        return series
+
+    def output(self, output_format) -> Union[Any,None]:
+        """Method to select output method
+
+        Args:
+            output_format (str): format of the output
+
+        Raises:
+            ValueError: In case of unmatched output
+
+        Returns:
+            method: return method to be used
+        """
+        if output_format == cnsts.JSON:
+            return self.output_json()
+        if output_format == cnsts.TEXT:
+            return self.output_text()
+        if output_format == cnsts.JUNIT:
+            return self.output_junit()
+        raise ValueError("Unsupported output format {output_format} selected")

pkg/algorithms/algorithm.py

@@ -1,12 +1,14 @@
 """
 Algorithm Factory to choose avaiable algorithms
 """
-from pkg.edivisive import EDivisive
-from pkg.isolationForest import IsolationForestWeightedMean
-import pkg.constants as cnsts
 from fmatch.matcher import Matcher
 import pandas as pd
-class AlgorithmFactory: # pylint: disable= too-few-public-methods, too-many-arguments
+import pkg.constants as cnsts
+from .edivisive import EDivisive
+from .isolationforest import IsolationForestWeightedMean
+
+
+class AlgorithmFactory: # pylint: disable= too-few-public-methods, too-many-arguments, line-too-long
     """Algorithm Factory to choose algorithm
     """
     def instantiate_algorithm(self, algorithm: str, matcher: Matcher, dataframe:pd.DataFrame, test: dict, options: dict, metrics_config: dict[str,dict]):

pkg/algorithmFactory.py renamed to pkg/algorithms/algorithmFactory.py

@@ -0,0 +1,4 @@
+"""
+Init for E-Divisive Algorithm
+"""
+from .edivisive import EDivisive

pkg/algorithms/edivisive/__init__.py

@@ -0,0 +1,29 @@
+"""EDivisive Algorithm from hunter"""
+
+# pylint: disable = line-too-long
+import pandas as pd
+from pkg.algorithms.algorithm import Algorithm
+
+
+class EDivisive(Algorithm):
+    """Implementation of the EDivisive algorithm using hunter
+
+    Args:
+        Algorithm (Algorithm): Inherits
+    """
+
+
+    def _analyze(self):
+        self.dataframe["timestamp"] = pd.to_datetime(self.dataframe["timestamp"])
+        self.dataframe["timestamp"] = self.dataframe["timestamp"].astype(int) // 10**9
+        series= self.setup_series()
+        change_points_by_metric = series.analyze().change_points
+
+        # filter by direction
+        for metric, changepoint_list in change_points_by_metric.items():
+            for i in range(len(changepoint_list)-1, -1, -1):
+                if ((self.metrics_config[metric]["direction"] == 1 and changepoint_list[i].stats.mean_1 > changepoint_list[i].stats.mean_2) or
+                    (self.metrics_config[metric]["direction"] == -1 and changepoint_list[i].stats.mean_1 < changepoint_list[i].stats.mean_2) ):
+                    del changepoint_list[i]
+
+        return series, change_points_by_metric

pkg/algorithms/edivisive/edivisive.py

@@ -0,0 +1,4 @@
+"""
+init for isolation forest
+"""
+from .isolationForest import IsolationForestWeightedMean

pkg/algorithms/isolationforest/__init__.py

@@ -0,0 +1,74 @@
+# pylint: disable = too-many-locals, line-too-long
+"""The implementation module for Isolation forest and weighted mean"""
+from sklearn.ensemble import IsolationForest
+import pandas as pd
+from fmatch.logrus import SingletonLogger
+from hunter.series import  ChangePoint, ComparativeStats
+from pkg.algorithms import Algorithm
+
+
+class IsolationForestWeightedMean(Algorithm):
+    """Isolation forest with weighted mean
+
+    Args:
+        Algorithm (Algorithm): _description_
+    """
+
+    def _analyze(self):
+        """Analyzing the data
+
+        Args:
+            dataframe (pd.DataFrame): _description_
+
+        Returns:
+            pd.Dataframe, pd.Dataframe: _description_
+        """
+        self.dataframe["timestamp"] = pd.to_datetime(self.dataframe["timestamp"])
+        self.dataframe["timestamp"] = self.dataframe["timestamp"].astype(int) // 10**9
+        dataframe = self.dataframe.copy(deep=True)
+        series = self.setup_series()
+
+        logger_instance = SingletonLogger.getLogger("Orion")
+        logger_instance.info("Starting analysis using Isolation Forest")
+        metric_columns = self.metrics_config.keys()
+        dataframe_with_metrics = dataframe[metric_columns]
+        model = IsolationForest(contamination="auto", random_state=42)
+        model.fit(dataframe_with_metrics)
+        predictions = model.predict(dataframe_with_metrics)
+        dataframe["is_anomaly"] = predictions
+        anomaly_scores = model.decision_function(dataframe_with_metrics)
+        # Add anomaly scores to the DataFrame
+        dataframe["anomaly_score"] = anomaly_scores
+
+        # Calculate moving average for each metric
+        window_size = (5 if self.options.get("anomaly_window",None) is None else int(self.options.get("anomaly_window",None)))
+        moving_averages = dataframe_with_metrics.rolling(window=window_size).mean()
+
+        # Initialize percentage change columns for all metrics
+        for feature in dataframe_with_metrics.columns:
+            dataframe[f"{feature}_pct_change"] = 0.0
+
+        change_points_by_metric={ k:[] for k in metric_columns }
+
+        for idx, row in dataframe.iterrows():
+            if row["is_anomaly"] == -1:
+                for feature in metric_columns:
+                    pct_change = (
+                        (row[feature] - moving_averages.at[idx, feature])
+                        / moving_averages.at[idx, feature]
+                    ) * 100
+                    if abs(pct_change) > (10 if self.options.get("min_anomaly_percent",None) is None else int(self.options.get("min_anomaly_percent",None))):
+                        if (pct_change * self.metrics_config[feature]["direction"] > 0) or self.metrics_config[feature]["direction"]==0:
+                            change_point = ChangePoint(metric=feature,
+                                                       index=idx,
+                                                       time=row['timestamp'],
+                                                       stats=ComparativeStats(
+                                                           mean_1=moving_averages.at[idx, feature],
+                                                           mean_2=row[feature],
+                                                           std_1=0,
+                                                           std_2=0,
+                                                           pvalue=1
+                                                       ))
+                            change_points_by_metric[feature].append(change_point)
+
+        return series, change_points_by_metric