example data generator

gensignals2.py

import math
import numpy as np

BCD_LENGTH = 100
BCD_SINE_CENTER = 300
A_CURVE = 1.5
B_SINE_AMP = 25
C_SINE_AMP = 50
D_SINE_AMP = 100
E_LENGTH = 100
SIGNAL_B_DELAY = 10
STEP_SIZE = 3

def makestep (data, step):
    stepped = []
    value = None
    for k in range(len(data)):
        if k % step == 0:
            value = data[k]
        stepped.append(value)
    return stepped

# --- generate data

partb = []
partc = []
partd = []

for k in range(BCD_LENGTH * 5 // 4):
    theta = 2.0 * math.pi * k / BCD_LENGTH
    if k < BCD_LENGTH:
        partb.append(math.sin(theta) * B_SINE_AMP + BCD_SINE_CENTER)
        partc.append(math.sin(theta) * C_SINE_AMP + BCD_SINE_CENTER)
    partd.append(math.sin(theta) * D_SINE_AMP + BCD_SINE_CENTER)

parte = []

for k in range(E_LENGTH):
    parte.append(D_SINE_AMP + BCD_SINE_CENTER)
    
parta = []

max_value = BCD_SINE_CENTER + max((B_SINE_AMP, C_SINE_AMP, D_SINE_AMP))
mid_theta_step = 2.0 * math.pi * 1 / BCD_LENGTH
for k in range(1, 1000000):
    parta.append(max(0, BCD_SINE_CENTER - math.pow(B_SINE_AMP * mid_theta_step * k, A_CURVE)))
    complete = parta + partb + partc + partd + parte
    mean = sum(complete) / len(complete)
    if mean <= max_value / 2: # keep adding until the mean is ~half the value range
        break
    
parta = list(reversed(parta))

complete = parta + partb + partc + partd + parte
complete = makestep(complete, STEP_SIZE)

# --- compute cross-correlation

signal_a = np.array(complete)
signal_b = np.array(partc)

# can test mean-centering like this
#signal_a = signal_a - np.mean(signal_a)
#signal_b = signal_b - np.mean(signal_b)

# can test differentiation like this
#signal_a = np.diff(signal_a, prepend=signal_a[0])
#signal_b = np.diff(signal_b, prepend=signal_b[0])

cc = np.correlate(signal_a, signal_b, mode="valid")

# --- write files

print("signal_a.csv...")
with open("signal_a.csv", "wt") as f:
    f.write("x,y\n")
    for x in range(len(signal_a)):
        f.write(f"{x},{signal_a[x]}\n")

print("signal_b.csv...")
partc_start = len(parta) + len(partb)
with open("signal_b.csv", "wt") as f:
    f.write("x,y\n")
    for x in range(len(signal_b)):
        f.write(f"{x+partc_start+SIGNAL_B_DELAY},{signal_b[x]}\n")

print("signal_cc.csv...")
with open("signal_cc.csv", "wt") as f:
    f.write("offset,c\n")
    start_offset = -(partc_start + SIGNAL_B_DELAY)
    for o in range(len(cc)):
        f.write(f"{o+start_offset},{cc[o]}\n")

https://stats.stackexchange.com/questions/661469/using-cross-correlation-to-find-signal-delay