Tests Blocked and Timed Out Tests

import gzip
import json
import os
import pandas as pd
import datetime

from ipynb.fs.defs.metric_template import testgrid_labelwise_encoding
from ipynb.fs.defs.metric_template import CephCommunication
from ipynb.fs.defs.metric_template import save_to_disk, read_from_disk
from dotenv import load_dotenv, find_dotenv

load_dotenv(find_dotenv())

## Specify variables

METRIC_NAME = "blocked_timed_out"

# Specify the path for input grid data,
INPUT_DATA_PATH = "../../../../data/raw/testgrid_183.json.gz"

# Specify the path for output metric data
OUTPUT_DATA_PATH = f"../../../../data/processed/metrics/{METRIC_NAME}"

## CEPH Bucket variables
## Create a .env file on your local with the correct configs,
s3_endpoint_url = os.getenv("S3_ENDPOINT")
s3_access_key = os.getenv("S3_ACCESS_KEY")
s3_secret_key = os.getenv("S3_SECRET_KEY")
s3_bucket = os.getenv("S3_BUCKET")
s3_path = os.getenv("S3_PROJECT_KEY", "ai4ci/testgrid/metrics")
s3_input_data_path = "raw_data"
AUTOMATION = os.getenv("IN_AUTOMATION")

## Import data
timestamp = datetime.datetime.today()

if AUTOMATION:

    filename = f"testgrid_{timestamp.day}{timestamp.month}.json"

    cc = CephCommunication(s3_endpoint_url, s3_access_key, s3_secret_key, s3_bucket)
    s3_object = cc.s3_resource.Object(s3_bucket, f"{s3_input_data_path}/{filename}")
    file_content = s3_object.get()["Body"].read().decode("utf-8")
    testgrid_data = json.loads(file_content)

else:
    with gzip.open(INPUT_DATA_PATH, "rb") as read_file:
        testgrid_data = json.load(read_file)

blocked_tests_list = testgrid_labelwise_encoding(testgrid_data, 8)[0:1000000]

len(blocked_tests_list)

blocked_tests_list[0]

# Convert to dataframe
blocked_tests_df = pd.DataFrame(
    blocked_tests_list,
    columns=["timestamp", "tab", "grid", "test", "test_duration", "test_blocked"],
)
blocked_tests_df.head()

timed_out_tests_list = testgrid_labelwise_encoding(testgrid_data, 9)[0:1000000]

len(timed_out_tests_list)

timed_out_tests_list[0]

# Convert to dataframe
timed_out_tests_df = pd.DataFrame(
    timed_out_tests_list,
    columns=["timestamp", "tab", "grid", "test", "test_duration", "test_timed_out"],
)
timed_out_tests_df.head()

no_tests = blocked_tests_df.test.count()
print("Total number of tests: %i" % (no_tests))
no_blocked = blocked_tests_df.test_blocked.sum()
print("Total number of tests blocked: %i" % (no_blocked))
test_blocked_percentage = (
    (blocked_tests_df.test_blocked.sum() / blocked_tests_df.test.count())
) * 100
print("Tests blocked percentage: %f" % (test_blocked_percentage))
no_timed_out = timed_out_tests_df.test_timed_out.sum()
print("Total number of timed out tests: %i" % (no_timed_out))
test_timed_out_percentage = (
    (timed_out_tests_df.test_timed_out.sum() / timed_out_tests_df.test.count())
) * 100
print("Test timed out percentage: %f" % (test_timed_out_percentage))

combined = pd.merge(
    blocked_tests_df,
    timed_out_tests_df,
    on=["timestamp", "tab", "grid", "test", "test_duration"],
)

combined.drop(columns="test_duration", inplace=True)
combined.head()

if AUTOMATION:
    cc = CephCommunication(s3_endpoint_url, s3_access_key, s3_secret_key, s3_bucket)
    cc.upload_to_ceph(
        combined,
        s3_path,
        f"{METRIC_NAME}/{METRIC_NAME}-{timestamp.year}-{timestamp.month}-{timestamp.day}.parquet",
    )
else:
    save_to_disk(
        combined,
        OUTPUT_DATA_PATH,
        f"{METRIC_NAME}-{timestamp.year}-{timestamp.month}-{timestamp.day}.parquet",
    )

## Sanity check to see if the dataset is the same
if AUTOMATION:
    sanity_check = cc.read_from_ceph(
        s3_path,
        f"{METRIC_NAME}/{METRIC_NAME}-{timestamp.year}-{timestamp.month}-{timestamp.day}.parquet",
    )
else:
    sanity_check = read_from_disk(
        OUTPUT_DATA_PATH,
        f"{METRIC_NAME}-{timestamp.year}-{timestamp.month}-{timestamp.day}.parquet",
    )

sanity_check