def sla_hourly_miss(sla_run_detail):
    """Generate hourly SLA miss table giving us details about the hour, SLA miss % for that hour, top DAG violators
    and the longest running task and avg task queue time for the given short timeframe.

    Args:
        sla_run_detail (dataframe): Base table consiting of details of all the dag runs that happened

    Returns:
        datframe, list: observations_hourly_reccomendations list and  sla_miss_percent_past_day_hourly dataframe
    """
    try:

        sla_miss_count_past_day = sla_run_detail[sla_run_detail["duration"] > sla_run_detail["sla"]][
            sla_run_detail["start_date"].between(short_timeframe_start_date, today)]

        sla_miss_count_hourly = (sla_miss_count_past_day.groupby(
            ["run_date_hour"]).size().to_frame(name="slamiss_count_hourwise").reset_index())
        sla_count_df_past_day_hourly = (sla_miss_count_past_day.groupby(["run_date_hour", "dag_id", "task_id"
                                                                         ]).size().to_frame(name="size").reset_index())
        sla_avg_execution_time_taskwise_hourly = (sla_miss_count_past_day.groupby(
            ["run_date_hour", "dag_id", "task_id"])["duration"].mean().reset_index())
        sla_avg_execution_time_hourly = (sla_avg_execution_time_taskwise_hourly.sort_values(
            "duration", ascending=False).groupby("run_date_hour", sort=False).head(1))

        sla_pastday_run_count_df = sla_run_detail[sla_run_detail["start_date"].between(
            short_timeframe_start_date, today)]
        sla_avg_queue_time_hourly = (sla_pastday_run_count_df.groupby(["run_date_hour"
                                                                       ])["task_queue_time"].mean().reset_index())
        sla_totalcount_hourly = (sla_pastday_run_count_df.groupby(
            ["run_date_hour"]).size().to_frame(name="total_count").sort_values("run_date_hour",
                                                                               ascending=False).reset_index())
        sla_totalcount_taskwise_hourly = (sla_pastday_run_count_df.groupby(
            ["run_date_hour", "dag_id",
             "task_id"]).size().to_frame(name="totalcount").sort_values("run_date_hour", ascending=False).reset_index())
        sla_miss_pct_past_day_hourly = pd.merge(sla_miss_count_hourly, sla_totalcount_hourly, on=["run_date_hour"])
        sla_miss_pct_past_day_hourly["sla_miss_percent"] = (sla_miss_pct_past_day_hourly["slamiss_count_hourwise"] *
                                                            100 / sla_miss_pct_past_day_hourly["total_count"]).round(2)

        sla_miss_pct_past_day_hourly["sla_miss_percent(missed_tasks/total_tasks)"] = sla_miss_pct_past_day_hourly.apply(
            lambda x: "%s%s(%s/%s)" % (
                x["sla_miss_percent"].astype(int),
                "% ",
                x["slamiss_count_hourwise"].astype(int),
                x["total_count"].astype(int),
            ),
            axis=1,
        )

        sla_highest_sla_miss_hour = (sla_miss_pct_past_day_hourly[["run_date_hour", "sla_miss_percent"
                                                                   ]].sort_values("sla_miss_percent",
                                                                                  ascending=False).head(1))
        sla_highest_tasks_hour = (sla_miss_pct_past_day_hourly[["run_date_hour",
                                                                "total_count"]].sort_values("total_count",
                                                                                            ascending=False).head(1))

        sla_miss_percent_past_day = sla_miss_pct_past_day_hourly.filter(
            ["run_date_hour", "sla_miss_percent(missed_tasks/total_tasks)"], axis=1)

        sla_miss_temp_df_pct1_past_day = pd.merge(
            sla_count_df_past_day_hourly,
            sla_totalcount_taskwise_hourly,
            on=["run_date_hour", "dag_id", "task_id"],
        )

        sla_miss_temp_df_pct1_past_day["pct_violator"] = (sla_miss_temp_df_pct1_past_day["size"] * 100 /
                                                          sla_miss_temp_df_pct1_past_day["totalcount"]).round(2)
        sla_miss_pct_past_day_hourly = (sla_miss_temp_df_pct1_past_day.sort_values(
            "pct_violator", ascending=False).groupby("run_date_hour", sort=False).head(1))

        sla_miss_pct_past_day_hourly["top_pct_violator"] = sla_miss_pct_past_day_hourly.apply(
            lambda x: "%s: %s (%s%s" % (x["dag_id"], x["task_id"], x["pct_violator"], "%)"),
            axis=1,
        )

        sla_miss_percent_violator_past_day_hourly = sla_miss_pct_past_day_hourly.filter(
            ["run_date_hour", "top_pct_violator"], axis=1)
        sla_miss_absolute_kpi_past_day_hourly = (sla_miss_temp_df_pct1_past_day.sort_values(
            "size", ascending=False).groupby("run_date_hour", sort=False).head(1))
        sla_miss_absolute_kpi_past_day_hourly["top_absolute_violator"] = sla_miss_absolute_kpi_past_day_hourly.apply(
            lambda x: "%s: %s (%s/%s)" % (x["dag_id"], x["task_id"], x["size"], x["totalcount"]),
            axis=1,
        )

        sla_miss_absolute_violator_past_day_hourly = sla_miss_absolute_kpi_past_day_hourly.filter(
            ["run_date_hour", "top_absolute_violator"], axis=1)
        slamiss_pct_exectime = pd.merge(
            pd.merge(
                sla_miss_percent_past_day,
                sla_miss_percent_violator_past_day_hourly,
                on="run_date_hour",
            ),
            sla_miss_absolute_violator_past_day_hourly,
            on="run_date_hour",
        ).sort_values("run_date_hour", ascending=False)

        sla_avg_execution_time_hourly["duration"] = (
            sla_avg_execution_time_hourly["duration"].round(0).astype(int).astype(str))
        sla_avg_execution_time_hourly["longest_running_task"] = sla_avg_execution_time_hourly.apply(
            lambda x: "%s: %s (%ss)" % (x["dag_id"], x["task_id"], x["duration"]), axis=1)

        sla_longest_running_task_hourly = sla_avg_execution_time_hourly.filter(
            ["run_date_hour", "longest_running_task"], axis=1)

        sla_miss_pct = pd.merge(slamiss_pct_exectime, sla_longest_running_task_hourly, on=["run_date_hour"])
        sla_miss_percent_past_day_hourly = pd.merge(sla_miss_pct, sla_avg_queue_time_hourly, on=["run_date_hour"])
        sla_miss_percent_past_day_hourly["task_queue_time"] = (
            sla_miss_percent_past_day_hourly["task_queue_time"].round(0).astype(int).apply(str))
        sla_longest_queue_time_hourly = (sla_miss_percent_past_day_hourly[["run_date_hour", "task_queue_time"
                                                                           ]].sort_values("task_queue_time",
                                                                                          ascending=False).head(1))

        sla_miss_percent_past_day_hourly.rename(
            columns={
                "task_queue_time": "Average Task Queue Time (s)",
                "longest_running_task": "Longest Running Task",
                "top_pct_violator": "Top Violator (%)",
                "top_absolute_violator": "Top Violator (absolute)",
                "run_date_hour": "Hour",
                "sla_miss_percent(missed_tasks/total_tasks)": "SLA miss % (Missed/Total Tasks)",
            },
            inplace=True,
        )

        obs1_hourlytrend = "Hour " + (sla_highest_sla_miss_hour["run_date_hour"].apply(str) +
                                      " had the highest percentage of SLA misses").to_string(index=False)
        obs2_hourlytrend = "Hour " + (
            sla_longest_queue_time_hourly["run_date_hour"].apply(str) + " had the longest average queue time (" +
            sla_longest_queue_time_hourly["task_queue_time"].apply(str) + " seconds)").to_string(index=False)
        obs3_hourlytrend = "Hour " + (sla_highest_tasks_hour["run_date_hour"].apply(str) +
                                      " had the most tasks running").to_string(index=False)

        observations_hourly_reccomendations = [obs1_hourlytrend, obs2_hourlytrend, obs3_hourlytrend]
        return observations_hourly_reccomendations, sla_miss_percent_past_day_hourly
    except:
        sla_miss_percent_past_day_hourly = pd.DataFrame(columns=[
            "SLA Miss % (Missed/Total Tasks)",
            "Top Violator (%)",
            "Top Violator (absolute)",
            "Longest Running Task",
            "Hour",
            "Average Task Queue Time (seconds)",
        ])
        observations_hourly_reccomendations = ""
        return observations_hourly_reccomendations, sla_miss_percent_past_day_hourly
Generate hourly SLA miss table giving us details about the hour, SLA miss % for that hour, top DAG violators
    and the longest running task and avg task queue time for the given short timeframe.

    Args:
        sla_run_detail (dataframe): Base table consiting of details of all the dag runs that happened

    Returns:
        datframe, list: observations_hourly_reccomendations list and  sla_miss_percent_past_day_hourly dataframe
    
sla-miss-report/airflow-sla-miss-report.py