phitoduck/cloudwatch-simulator

1import random2from datetime import datetime, timedelta, date, time3import pandas as pd4import numpy as np5from typing import List, Iterator, Dict, Any, Optional6 7def generate_random_data(8    date: date,9    start_time: time,10    end_time: time,11    count: int,12    response_time_range: (int, int),13    null_percentage: float14) -> pd.DataFrame:15    start_datetime: datetime = datetime.combine(date, start_time)16    end_datetime: datetime = datetime.combine(date, end_time)17 18    random_timestamps: List[datetime] = [19        start_datetime + timedelta(seconds=random.randint(0, int((end_datetime - start_datetime).total_seconds())))20        for _ in range(count)21    ]22    random_timestamps.sort()23 24    random_response_times: List[Optional[int]] = [25        random.randint(response_time_range[0], response_time_range[1]) for _ in range(count)26    ]27 28    null_count: int = int(null_percentage * count)29    null_indices: List[int] = random.sample(range(count), null_count)30    for idx in null_indices:31        random_response_times[idx] = None32 33    data: Dict[str, Any] = {34        'timestamp': random_timestamps,35        'ResponseTime(ms)': random_response_times36    }37    df: pd.DataFrame = pd.DataFrame(data)38    return df39 40def calculate_percentile(41    df: pd.DataFrame,42    freq: str,43    percentile: float44) -> pd.DataFrame:45    percentile_df: pd.DataFrame = df.groupby(pd.Grouper(key='timestamp', freq=freq))["ResponseTime(ms)"]\46                                    .quantile(percentile).reset_index(name=f"p{int(percentile * 100)}_ResponseTime(ms)")47    percentile_df.replace(to_replace=np.nan, value=None, inplace=True)48    return percentile_df49 50def aggregate_data(51    df: pd.DataFrame,52    period_length: str,53) -> pd.DataFrame:54    if df.empty:55        return pd.DataFrame()  # Return an empty DataFrame if input is empty56 57    aggregation_funcs = {58        'p50': lambda x: np.percentile(x.dropna(), 50) if not x.dropna().empty else np.nan,59        'p95': lambda x: np.percentile(x.dropna(), 95) if not x.dropna().empty else np.nan,60        'p99': lambda x: np.percentile(x.dropna(), 99) if not x.dropna().empty else np.nan,61        'max': lambda x: np.max(x.dropna()) if not x.dropna().empty else np.nan,62        'min': lambda x: np.min(x.dropna()) if not x.dropna().empty else np.nan,63        'average': lambda x: np.mean(x.dropna()) if not x.dropna().empty else np.nan64    }65 66    summary_df = df.groupby(pd.Grouper(key='timestamp', freq=period_length)).agg(67        p50=('ResponseTime(ms)', aggregation_funcs['p50']),68        p95=('ResponseTime(ms)', aggregation_funcs['p95']),69        p99=('ResponseTime(ms)', aggregation_funcs['p99']),70        max=('ResponseTime(ms)', aggregation_funcs['max']),71        min=('ResponseTime(ms)', aggregation_funcs['min']),72        average=('ResponseTime(ms)', aggregation_funcs['average']),73    ).reset_index()74    return summary_df75 76def re_aggregate_data(77    df: pd.DataFrame,78    period_length: str,79) -> pd.DataFrame:80    if df.empty:81        return pd.DataFrame()  # Return an empty DataFrame if input is empty82 83    aggregation_funcs = {84        'p50': lambda x: np.percentile(x.dropna(), 50) if not x.dropna().empty else np.nan,85        'p95': lambda x: np.percentile(x.dropna(), 95) if not x.dropna().empty else np.nan,86        'p99': lambda x: np.percentile(x.dropna(), 99) if not x.dropna().empty else np.nan,87        'max': lambda x: np.max(x.dropna()) if not x.dropna().empty else np.nan,88        'min': lambda x: np.min(x.dropna()) if not x.dropna().empty else np.nan,89        'average': lambda x: np.mean(x.dropna()) if not x.dropna().empty else np.nan90    }91 92    summary_df = df.groupby(pd.Grouper(key='timestamp', freq=period_length)).agg(93        p50=('p50', aggregation_funcs['p50']),94        p95=('p95', aggregation_funcs['p95']),95        p99=('p99', aggregation_funcs['p99']),96        max=('max', aggregation_funcs['max']),97        min=('min', aggregation_funcs['min']),98        average=('average', aggregation_funcs['average']),99    ).reset_index()100    return summary_df101 102def downsample(df, period_minutes):103    # Create a new datetime index at specified intervals104    freq_str = f'{period_minutes}T'105    new_index = pd.date_range(start=df['timestamp'].min(), end=df['timestamp'].max(), freq=freq_str)106    107    # Create an empty DataFrame with the new index108    df_downsampled = pd.DataFrame(index=new_index)109    110    # Set the original DataFrame's index to the timestamp column111    df.set_index('timestamp', inplace=True)112    113    # Interpolate the values for each column114    for column in df.columns:115        df_downsampled[column] = df[column].resample(freq_str).interpolate(method='linear')116    117    # Reset index to have timestamp as a column again118    df_downsampled.reset_index(inplace=True)119    df_downsampled.rename(columns={'index': 'timestamp'}, inplace=True)120    121    return df_downsampled122 123def chunk_list(input_list: List[Any], size: int = 3) -> Iterator[List[Any]]:124    while input_list:125        chunk: List[Any] = input_list[:size]126        yield chunk127        input_list = input_list[size:]128 129def evaluate_alarm_state(130    summary_df: pd.DataFrame,131    threshold: int,132    datapoints_to_alarm: int,133    evaluation_range: int,134    aggregation_function: str,135    alarm_condition: str136) -> pd.DataFrame:137    data_points: List[Optional[float]] = list(summary_df[aggregation_function].values)138 139    data_table_dict: Dict[str, List[Any]] = {140        "DataPoints": [],141        "# of data points that must be filled": [],142        "MISSING": [],143        "IGNORE": [],144        "BREACHING": [],145        "NOT BREACHING": []146    }147 148    def check_condition(value, threshold, condition):149        if condition == '>':150            return value > threshold151        elif condition == '>=':152            return value >= threshold153        elif condition == '<':154            return value < threshold155        elif condition == '<=':156            return value <= threshold157 158    for chunk in chunk_list(input_list=data_points, size=evaluation_range):159        data_point_repr: str = ''160        num_dp_that_must_be_filled: int = 0161 162        for dp in chunk:163            if str(dp).lower() == "nan":164                dp_symbol = '⚫️'165            elif check_condition(dp, threshold, alarm_condition):166                dp_symbol = '🔴'167            else:168                dp_symbol = '🟢'169            data_point_repr += dp_symbol170 171        if len(chunk) < evaluation_range:172            data_point_repr += '⚫️' * (evaluation_range - len(chunk))173 174        if data_point_repr.count('⚫️') > (evaluation_range - datapoints_to_alarm):175            num_dp_that_must_be_filled = datapoints_to_alarm - sum([data_point_repr.count('🟢'), data_point_repr.count('🔴')])176 177        data_table_dict["DataPoints"].append(data_point_repr)178        data_table_dict["# of data points that must be filled"].append(num_dp_that_must_be_filled)179 180        if num_dp_that_must_be_filled > 0:181            data_table_dict["MISSING"].append("INSUFFICIENT_DATA" if data_point_repr.count('⚫️') == evaluation_range else "Retain current state")182            data_table_dict["IGNORE"].append("Retain current state")183            data_table_dict["BREACHING"].append("ALARM")184            data_table_dict["NOT BREACHING"].append("OK")185        else:186            data_table_dict["MISSING"].append("OK")187            data_table_dict["IGNORE"].append("Retain current state")188            data_table_dict["BREACHING"].append("ALARM" if '🔴' * datapoints_to_alarm in data_point_repr else "OK")189            data_table_dict["NOT BREACHING"].append("ALARM" if '🟢' * datapoints_to_alarm not in data_point_repr else "OK")190 191    return pd.DataFrame(data_table_dict)192