def create_signal()

in aiops/ContraLSP/switchstate/switchgenerator.py [0:0]

• 53 lines of code
• 14 McCabe index (conditional complexity)

def create_signal(sig_len, gp_params, mean, cov):
    signal = None
    state_local = []
    y = []
    importance = []
    y_logits = []

    previous = np.random.binomial(1, P_S0)[0]
    previous_label = None
    delta_state = 1

    # Sample for "previous" state (this is current state now)
    imp_sig = np.zeros(SIG_NUM)
    imp_sig[imp_feature[previous]] = 1
    importance.append(imp_sig)
    state_local.append(previous)

    for ii in range(1, sig_len):
        next_st = next_state(previous, delta_state)
        state_n = next_st

        imp_sig = np.zeros(SIG_NUM)
        if previous != state_n:
            # this samples labels+samples until  current point - before state change at next time point
            gp_vec = [ts.signals.GaussianProcess(lengthscale=g, mean=m, variance=0.1) for g, m in
                      zip(gp_params, mean[previous])]
            sample_ii = np.array([gp.sample_vectorized(time_vector=np.array(range(delta_state))) for gp in gp_vec])
            # print(sample_ii.shape)
            # sample_ii = np.random.multivariate_normal(mean[state_n], cov[state_n])
            if signal is not None:
                signal = np.hstack((signal, sample_ii))
            else:
                signal = sample_ii

            # signal.extend(sample_ii)
            # sample_ii = (sample_ii).reshape((1, -1))

            # y_probs = state_n * generate_linear_labels(sample_ii[:, imp_feature[state_n]]) + \
            #          (1 - state_n) * generate_linear_labels(sample_ii[:, imp_feature[state_n]])

            y_probs = generate_linear_labels(sample_ii.T[:, imp_feature[previous]])

            y_logit = [yy[1] for yy in y_probs]
            y_label = [np.random.binomial(1, yy) for yy in y_logit]

            # y_logit_past = y_logit
            y.extend(y_label)
            y_logits.extend(y_logit)
            delta_state = 1
            imp_sig[imp_feature[state_n]] = 1
            imp_sig[-1] = 1
        else:
            delta_state += 1
        importance.append(imp_sig)

        # previous_label = y_label
        state_local.append(state_n)
        previous = state_n

    # sample points in the last state-change
    gp_vec = [ts.signals.GaussianProcess(lengthscale=g, mean=m, variance=0.1) for g, m in
              zip(gp_params, mean[previous])]
    sample_ii = np.array([gp.sample_vectorized(time_vector=np.array(range(delta_state))) for gp in gp_vec])

    # sometimes only one state is ever sampled
    if signal is not None:
        signal = np.hstack((signal, sample_ii))
    else:
        signal = sample_ii

    y_probs = generate_linear_labels(sample_ii.T[:, imp_feature[previous]])

    y_logit = [yy[1] for yy in y_probs]
    y_label = [np.random.binomial(1, yy) for yy in y_logit]

    y.extend(y_label)
    y_logits.extend(y_logit)

    # signal = signal
    y = np.array(y)
    importance = np.array(importance)

    # print(signal.shape, y.shape, len(state_local), importance.shape, len(y_logits))
    return signal, y, state_local, importance, y_logits