from pathlib import Path

from context import ClinicContext
from mathbox.app.signal.filter import f_lowpass_filter, moving_median
from mathbox.app.signal.outlier import noise_outlier, simple_outlier
from mathbox.na.interpolation import series_linear_interp
from mathbox.statistics.estimator import mean, median, std
from requests import Response
from yaml import Loader, load

measurement = "avg_latency_analysis"

threshold = {
    "tidb_p99_rt:total": 0.5,  # s for oltp
    "get_token_duration": 2,  # us
    "parse_duration:general": 0.01,  # s
    "compile_duration:general": 0.03,  # s
    "execute_duration:general": 0.45,  # ?
    "wait_start_tso_duration": 0.03,  # s with network
    "kv_backoff_duration": 2,  # ?
    "kv_request_duration": 0.4,  # ?
    "tikv_p99_rt:total": 0.4,  # ?
    "coprocessor_duration:total": 1,  # ?
    "coprocessor_duration:by_instance": 1,  # ?
    "small_read_duration:total": 0.2,  # ?
    "small_read_duration:by_instance": 0.2,  # ?
    "write_rpc_latency": 0.4,  # ?
    "write_rpc_duration:by_instance": 0.4,  # ?
    "copr_wait_duration": 0.05,  # s
    "copr_handle_duration": 0.25,  # ?
    "scheduler_command_duration": 0.35,  # ?
    "latch_wait_duration": 0.02,  # s
    "async_snapshot_duration": 0.2,  # ?
    "async_write_duration": 0.2,  # s
    "imbalance": 0.6,  # no evidence
}

avg_threshold = {
    "store": 0.0138 * 5,
    "apply": 0.019 * 5,
    "plan_cache": 0.01,  # ?
    "stmt_cps": 0.4,  # ?
    "qps_total": 0.1,  # ?
    "conn_idle_duration:in_txn": 0.114 * 2,
    "conn_idle_duration:not_in_txn": 0.114 * 2,
    "tidb_p99_rt:total": 0.013 * 30,  # for oltp
    "get_token_duration": 2,  # us
    "parse_duration:general": 0.01 * 1,
    "compile_duration:general": 0.038 * 1,
    "execute_duration:general": 0.013 * 20,
    "wait_start_tso_duration": 0.000639 * 50,  # s with network
    "kv_backoff_duration": 0.5,  # ?
    "kv_request_duration": 0.4,  # ?
    "tikv_p99_rt:total": 0.4,  # ?
    "coprocessor_duration:total": 1,  # ?
    "coprocessor_duration:by_instance": 1,  # ?
    "small_read_duration:total": 0.0003 * 200,  # ?
    "small_read_duration:by_instance": 0.0003 * 200,  # ?
    "write_rpc_latency": (0.0498 + 0.034 + 0.00911) * 5,
    "write_rpc_duration:by_instance": (0.0498 + 0.034 + 0.00911) * 5,
    "copr_wait_duration": 0.05,  # s
    "copr_handle_duration": 0.25,  # ?
    "scheduler_command_duration": 0.35,  # ?
    "latch_wait_duration": 0.02,  # s
    "async_snapshot_duration": 0.2,  # ?
    "async_write_duration": 0.19,  # s
    "imbalance": 0.6,  # no evidence
}


class Signal:
    # One Signal with name, timestamp and value
    def __init__(self, timestamp, value):
        self.timestamp = timestamp
        self.value = value
        self.change_points = None
        self.outliers = None
        self.details = {}

    def __truediv__(self, other):
        if self.timestamp != other.timestamp:
            raise ValueError("Signal timestamp not equal")
        value = []
        for i in range(len(self.timestamp)):
            value.append(self.value[i] / (other.value[i] + 1e-10))
        return Signal(self.timestamp, value)

    def set_name(self, name):
        self.name = name

    def interp(self, desired_timestamp):
        self.value = series_linear_interp(self.timestamp, self.value, desired_timestamp)
        self.timestamp = desired_timestamp

    # outlier point dectection, return pos index of outlier points
    def outlier_detect(self, window=5, n_top_feq=3, f_low_rate=0.01, level=3):
        # must be interpolated first!!!
        if self.outliers is not None:
            return
        sig_filtered = moving_median(self.value, window)
        moved_trend = [x - y for x, y in zip(self.value, sig_filtered)]
        sample_period = self.timestamp[1] - self.timestamp[0]
        _, seasonality = f_lowpass_filter(
            moved_trend, sample_period, n_top_feq, f_low_rate / sample_period
        )
        moved_seasonality = [x - y for x, y in zip(moved_trend, seasonality)]
        _, outlier_hi = noise_outlier(moved_seasonality, level)
        # self.outliers = [x[0] for x in outlier_hi]
        filtered_outlier = []  # modify change points: remove small changes
        for item in outlier_hi:  # modify change points: remove small changes
            if item[0] < 20:  # modify change points: remove small changes
                segment = moved_seasonality[
                    item[0] : item[0] + 40
                ]  # modify change points: remove small changes
            else:  # modify change points: remove small changes
                segment = moved_seasonality[
                    item[0] - 20 : item[0] + 20
                ]  # modify change points: remove small changes
            if max(segment) > 0.5:  # 0.1 modify change points: remove small changes
                filtered_outlier.append(item[0])  # modify change points: remove small changes
        self.outliers = filtered_outlier  # modify change points: remove small changes

    # for stationary data, return [[pos_included, pos_excluded), [pos_included, pos_excluded), ...]]
    def outlier_detect_stationary(self, window=20, density=0.9):
        seg_pos = anomaly_slice(self.value, window, density)
        self.stationary_anomaly_seg = seg_pos

    def get_slice(self, start, stop):
        slice = Signal(self.timestamp[start:stop], self.value[start:stop])
        slice.set_name(self.name)
        return slice

    def add_baseline(self, baseline):
        # baseline is a Signal(timestamp, value) and set_name(name)
        self.details["baseline"] = baseline


class Metrics:
    # One PromQL query with name, timestamp, query response and signals
    def __init__(self, name, query):
        self.name = name
        self.query = query
        self.signals = []  # list of Signal

    def to_signals(self, resp: Response):
        # response is {"status": "xxx", "data": {"resultType": "matrix", "result": [{"metrics":{<label_key>: <label_value>, ...}, "value":[[<unix>, "<value>"],...]},...] }}
        resp = resp.json()["data"]
        for item in resp["result"]:
            node = list(item["metric"].values())
            if len(node) == 1:
                node = node[0]
            else:
                print("Node is not unique")
            timestamp = []
            value = []
            for pair in item["values"]:
                timestamp.append(pair[0])
                value.append(float(pair[1]))
            signal = Signal(timestamp, value)
            signal.set_name(node)
            self.signals.append(signal)


def get_panel_data(context: ClinicContext, fname: str, start: int, end: int):
    tidb_cluster_id = str(context.cluster_id)
    p = Path(__file__).with_name(fname)
    f = p.open("r")
    cfg = load(f, Loader=Loader)

    desired_timestamp = list(range(start, end + 15, 15))
    metrics_per_panel = []

    for item in cfg["cooked"]:
        # item is a panel with id, name, obj and targets
        panel_id = item["id"]
        method = item["method"]

        query = item["object"]["expr"].replace("{tidb_cluster_id}", tidb_cluster_id)
        obj = Metrics(item["object"]["record"], query)

        metrics_resp = context.get_metrics(query, start, end)
        obj.to_signals(metrics_resp)
        # obj.signals = obj.signals[0]
        if len(obj.signals) > 0:
            for sig in obj.signals:
                if len(sig.value) >= 2:
                    sig.interp(desired_timestamp)

        candidates: list[Metrics] = []
        if item["targets"] is not None:
            for target in item["targets"]:
                query = target["expr"].replace("{tidb_cluster_id}", tidb_cluster_id)
                candidate = Metrics(target["record"], query)
                metrics_resp = context.get_metrics(query, start, end)
                candidate.to_signals(metrics_resp)
                for metric in candidate.signals:
                    metric.interp(desired_timestamp)
                candidates.append(candidate)

        metrics_per_panel.append(
            {
                "panel_id": panel_id,
                "method": method,
                "object": obj,
                "candidates": candidates,
            }
        )
    return metrics_per_panel


# offset unit is hour
def set_historical_baseline(context: ClinicContext, metrics: Metrics, offset: int, end: int):
    start = end - offset * 3600
    desired_timestamp = list(range(start, end + 15, 15))
    metrics_baseline = Metrics(metrics.name, metrics.query)
    metrics_resp = context.get_metrics(metrics.query, start, end)
    # if query res error, return None
    metrics_baseline.to_signals(metrics_resp)
    for signal in metrics_baseline.signals:
        signal.interp(desired_timestamp)
    for signal in metrics.signals:
        for sig in metrics_baseline.signals:
            if sig.name == signal.name:
                signal.add_baseline(sig)
                break


def cal_metrics_anomaly(metrics, window=20, density=0.8):
    # return anomaly_slice
    for sig in metrics.signals:
        if "baseline" in sig.details.keys():  # sig.details['baseline'] is not None:
            avg = mean(sig.details["baseline"].value)
        else:
            avg = avg_threshold[metrics.name]
        # if metrics.name == "latch_wait_duration":
        #        print("avg: ", avg, " max: ", max(sig.value))
        if max(sig.value) <= avg:
            sig.stationary_anomaly_seg = []
            continue
        if min(sig.value) >= avg * 2:
            sig.stationary_anomaly_seg = [(0, len(sig.timestamp) - 1)]
            continue
        # to be continued: there is still some case that need to be handeled (e.g. Q1 low Q3 high and MAX approx to Q3)
        sig.outlier_detect_stationary(window, density)
        for seg in sig.stationary_anomaly_seg[:]:
            if max(sig.value[seg[0] : seg[1]]) <= avg:
                sig.stationary_anomaly_seg.remove(seg)


def anomaly_slice(signal, window=40, density=0.8):
    continuous_anomaly_flag = [0 for _ in range(len(signal))]
    # assume signal is stable.
    _, outlier_hi = simple_outlier(signal, bias=1.5)
    anomaly_flag = [1 if i in ([i for (i, _) in outlier_hi]) else 0 for i in range(len(signal))]
    for i in range(len(signal) - window):
        if sum(anomaly_flag[i : i + window]) / window <= density:
            continue
        else:
            continuous_anomaly_flag[i : i + window] = [1 for _ in range(window)]
    pos_slice = cal_continuous_flag_slice(continuous_anomaly_flag, 0)
    return pos_slice


def cal_continuous_flag_slice(flag_slice, continuous_num=9):
    # [[include, exclude),[include, exclude),[include, exclude),[include, exclude),[include, exclude),...]
    # flag_slice = [1,1,1,1,1,1,1] # mock
    # flag_slice = [0,0,0,0,0,0,0]
    # flag_slice = [0,0,0,0,0,1,1,1,1,1,1]
    # flag_slice = [1,1,1,1,1,0,0,0,0,0]
    # flag_slice = [0,0,1,1,0,1,1,0,0,1,1,1,1,0,0,0,0,0,1,1,1,1,1,1,1,1,1,0,0,0] # mock
    # flag_slice = [0,0,1,1,1,1,1,0,0,1,1,1,1,0,0,0,0,0,1,1,1,1,1,1,1,1,1,1,1,1] # mock
    # flag_slice = [1,1,0,0,0,0,0,1,1,0,0,0,0,1,1,1,1,1,0,0,0,0,0,0,0,0,0,1,1,1] # mock
    # flag_slice = [1,1,0,0,0,0,0,1,0,0,0,0,0,1,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0,0] # mock
    # continuous_num = 3 # mock

    rising_edge = []
    falling_edge = []
    res = []
    for i in range(len(flag_slice) - 1):
        if flag_slice[i] == 0 and flag_slice[i + 1] == 1:
            rising_edge.append(i)
        if flag_slice[i] == 1 and flag_slice[i + 1] == 0:
            falling_edge.append(i)
    i = len(rising_edge)
    j = len(falling_edge)
    if i == 0 and j == 0:
        if flag_slice[0] == 1:
            res.append((0, len(flag_slice)))
    elif i == 0 and j == 1:
        res.append((0, falling_edge[0] + 1))
    elif i == 1 and j == 0:
        res.append((rising_edge[0] + 1, len(flag_slice)))
    elif rising_edge[0] < falling_edge[0]:  # rising fisrt
        for s in range(j):
            if falling_edge[s] - rising_edge[s] >= continuous_num:
                res.append((rising_edge[s] + 1, falling_edge[s] + 1))
        if j < i:
            if len(flag_slice) - rising_edge[-1] >= continuous_num:
                res.append((rising_edge[-1] + 1, len(flag_slice)))
    elif rising_edge[0] > falling_edge[0]:  # falling first
        if falling_edge[0] + 1 >= continuous_num:
            res.append((0, falling_edge[0] + 1))
        for s in range(j - 1):
            if falling_edge[s + 1] - rising_edge[s] >= continuous_num:
                res.append((rising_edge[s] + 1, falling_edge[s + 1] + 1))
        if i == j:
            if len(flag_slice) - rising_edge[-1] >= continuous_num:
                res.append((rising_edge[-1] + 1, len(flag_slice)))
    return res


def cal_imbalance(pieces, ratio_threshold=0.4, continuous_num=9):
    # ratio_threshold: choose a group of instance-write-rpc in some timestamp from the following url
    # https://clinic.pingcap.com:4433/grafana/d/TiKVFastTune/clinic-cluster-tikv-fasttune?orgId=1&var-tidb_cluster=6862533764151471104&var-instance=&var-store=All&from=1641332122000&to=1641339322000&refresh=5s
    # continuous_num = 4 * n + 1, n: how many minutes continuously
    num_p = len(pieces)
    num_t = len(pieces[0].timestamp)
    imbalance_ratio = []
    continuous_imbalance_flag = []
    for i in range(num_t):
        cake = []
        for j in range(num_p):
            cake.append(pieces[j].value[i])
        # ratio = (max(cake) - sum(cake) / num_p) / (sum(cake) / num_p) # not break change but important
        ratio = 1 - min(cake) / (max(cake) + 1e-10)  # not break change but important
        imbalance_ratio.append(ratio)

        if ratio > ratio_threshold:
            continuous_imbalance_flag.append(1)
        else:
            continuous_imbalance_flag.append(0)

    flag_slice = cal_continuous_flag_slice(continuous_imbalance_flag, continuous_num)
    res = []
    for item in flag_slice:
        res.append((pieces[0].timestamp[item[0]], pieces[0].timestamp[item[1] - 1]))

    stats = {}
    stats["p10"] = quantile(imbalance_ratio, 0.1)
    stats["p20"] = quantile(imbalance_ratio, 0.20)
    stats["p95"] = quantile(imbalance_ratio, 0.95)
    stats["p50"] = median(imbalance_ratio)
    stats["avg"] = mean(imbalance_ratio)
    stats["std"] = std(imbalance_ratio)
    return (
        res,
        stats,
    )  # BREAK CHANGE!!! job_fasttune depends on returned res without stats!!!


def write_annotations(context: ClinicContext, annotations):
    for item in annotations:
        start_ts = int(item["annotation"]["start"])
        end_ts = int(item["annotation"]["end"])
        duration = end_ts - start_ts
        panel_id = item["panel_id"]
        payload = {
            "timestamp": start_ts,
            "measurement": f"{measurement}_anno",
            "tidb_cluster": context.cluster_id,
            "fields": {
                "seconds": duration,
            },
            "tags": {
                "text": item["annotation"]["text"],
                "tags": ",".join(item["annotation"]["tags"]),
                "panel_id": str(panel_id),
            },
        }
        resp = context.write_timeseries_data(payload=payload)
        if not resp.ok:
            print(f"payload: {payload}")
            print(
                f"write_timeseries_data failed, status_code: {resp.status_code} reason: {resp.reason}"
            )


def write_report(context: ClinicContext, report):
    pass


def quantile(data, q):
    data.sort()
    return data[int(len(data) * q)]