313 lines
8.7 KiB
Python
313 lines
8.7 KiB
Python
from matplotlib.pylab import sca
|
|
from PIL.ImageChops import offset
|
|
from siglent_sdg.siglent import SiglentGen
|
|
from asyncio import sleep
|
|
from rigol_dho_lib.rigol import RigolOsc, CHANNEL_COUNT
|
|
import time
|
|
from datetime import datetime
|
|
from rigol_dho_lib import rigol
|
|
from siglent_sdg import siglent
|
|
import numpy as np
|
|
import csv
|
|
from scipy.integrate import cumulative_trapezoid
|
|
|
|
import matplotlib.pyplot as plt
|
|
|
|
|
|
# R =3.4 ohm
|
|
SINE_FREQ = 100e3
|
|
|
|
CYCLE_COUNT = 1000
|
|
|
|
DIV_COUNT = 10
|
|
|
|
OSC_CHANNEL_A = 1 # GEN signal
|
|
|
|
OSC_CHANNEL_B = 2 # OUT
|
|
GEN_CHANNEL = siglent.ChannelID.CH1
|
|
|
|
UPPER_BOUND_DIV = 3.5
|
|
LOWER_BOUND_DIV = 1
|
|
TARGET_DIV = (UPPER_BOUND_DIV + LOWER_BOUND_DIV) / 2
|
|
|
|
PATH = "/home/zychlix/Desktop/pomiary/out"
|
|
PATH_BH = "/home/zychlix/Desktop/pomiary/out_BH"
|
|
PATH_BH_RAW = "/home/zychlix/Desktop/pomiary/out_BH_raw"
|
|
|
|
R0 = 19.82
|
|
|
|
AMPLITUDE = 10
|
|
COUNT = 100
|
|
|
|
|
|
class aux_ch:
|
|
def __init__(self, id):
|
|
self.id = id
|
|
self.data = []
|
|
self.done = False
|
|
|
|
|
|
class ImpedanceAnalyzer:
|
|
def __init__(self, gen_addr: str, osc_addr: str):
|
|
self.gen: siglent.SiglentGen = siglent.SiglentGen(gen_addr)
|
|
self.osc: rigol.RigolOsc = rigol.RigolOsc(osc_addr)
|
|
|
|
self.mem_depth = self.osc.getMemoryDepth()
|
|
self.osc.setPoints(self.mem_depth)
|
|
|
|
self.amplitude = AMPLITUDE
|
|
self.debug_voltages = []
|
|
|
|
self.span_a = 0
|
|
self.span_b = 0
|
|
|
|
self.dc = 0
|
|
|
|
self.scales = [1] * (CHANNEL_COUNT + 1)
|
|
|
|
# for i in self.osc.channels
|
|
# se
|
|
|
|
return
|
|
|
|
def getImpedance(self, freq: float):
|
|
setWindowSize(self.osc, CYCLE_COUNT, freq)
|
|
|
|
self.gen.channels[GEN_CHANNEL].apply_sine(freq, self.amplitude, self.dc)
|
|
self.gen.channels[GEN_CHANNEL].set_output(True)
|
|
|
|
self.autoscale()
|
|
|
|
self.osc.run()
|
|
time.sleep(1)
|
|
|
|
self.osc.single()
|
|
|
|
time.sleep(0.3)
|
|
|
|
channel_A_data = self.osc.getChannel(OSC_CHANNEL_A).getWaveform()
|
|
channel_B_data = self.osc.getChannel(OSC_CHANNEL_B).getWaveform()
|
|
|
|
time_array = self.osc.getChannel(OSC_CHANNEL_A).genTimeArray(channel_A_data)
|
|
|
|
v_a = dft(channel_A_data, time_array, freq)
|
|
v_o = dft(channel_B_data, time_array, freq)
|
|
|
|
print(f"{freq} V_a {v_a} V_o {v_o}")
|
|
|
|
# z = R0 * (v_o) / v_o #resistor to gnd
|
|
|
|
z = R0 * (v_o) / (v_a - v_o) # resistor to input
|
|
# z = v_a / (v_o / R0)
|
|
|
|
self.debug_voltages.append([np.abs(v_a), np.abs(v_o)])
|
|
|
|
return z
|
|
|
|
def getScaledWaveform(self, ch):
|
|
while True:
|
|
self.osc.single()
|
|
time.sleep(1)
|
|
data = self.osc.getChannel(ch).getWaveform()
|
|
|
|
minimum = np.min(data)
|
|
maximum = np.max(data)
|
|
|
|
scale_range = self.osc.getChannel(ch).getScaleRange()
|
|
if (
|
|
maximum / self.scales[ch] > UPPER_BOUND_DIV
|
|
or minimum / self.scales[ch] < -UPPER_BOUND_DIV
|
|
):
|
|
self.scales[ch] = self.scales[ch] * 2
|
|
if self.scales[ch] > scale_range[1]:
|
|
self.scales[ch] = scale_range[1]
|
|
self.osc.getChannel(ch).setVScale(self.scales[ch])
|
|
break
|
|
elif (
|
|
minimum / self.scales[ch] < LOWER_BOUND_DIV
|
|
and minimum / self.scales[ch] > -LOWER_BOUND_DIV
|
|
):
|
|
self.scales[ch] = self.scales[ch] / 2
|
|
if self.scales[ch] < scale_range[0]:
|
|
self.scales[ch] = scale_range[0]
|
|
self.osc.getChannel(ch).setVScale(self.scales[ch])
|
|
break
|
|
else:
|
|
break
|
|
|
|
self.osc.getChannel(ch).setVScale(self.scales[ch])
|
|
print(
|
|
f"Channel {ch} settings: min: {minimum}, max: {maximum} scale:{self.scales[ch]}"
|
|
)
|
|
|
|
return data
|
|
|
|
def autoscale(self):
|
|
channels = []
|
|
|
|
channels.append(aux_ch(1))
|
|
channels.append(aux_ch(2))
|
|
|
|
while not (channels[0].done and channels[1].done):
|
|
self.osc.single()
|
|
time.sleep(1)
|
|
for ch in channels:
|
|
ch.data = self.osc.getChannel(ch.id).getWaveform()
|
|
|
|
minimum = np.min(ch.data)
|
|
maximum = np.max(ch.data)
|
|
|
|
scale_range = self.osc.getChannel(ch.id).getScaleRange()
|
|
if (
|
|
maximum / self.scales[ch.id] > UPPER_BOUND_DIV
|
|
or minimum / self.scales[ch.id] < -UPPER_BOUND_DIV
|
|
):
|
|
self.scales[ch.id] = self.scales[ch.id] * 2
|
|
if self.scales[ch.id] > scale_range[1]:
|
|
self.scales[ch.id] = scale_range[1]
|
|
self.osc.getChannel(ch.id).setVScale(self.scales[ch.id])
|
|
ch.done = True
|
|
elif (
|
|
minimum / self.scales[ch.id] < LOWER_BOUND_DIV
|
|
and minimum / self.scales[ch.id] > -LOWER_BOUND_DIV
|
|
):
|
|
self.scales[ch.id] = self.scales[ch.id] / 2
|
|
if self.scales[ch.id] < scale_range[0]:
|
|
self.scales[ch.id] = scale_range[0]
|
|
self.osc.getChannel(ch.id).setVScale(self.scales[ch.id])
|
|
ch.done = True
|
|
else:
|
|
ch.done = True
|
|
|
|
self.osc.getChannel(ch.id).setVScale(self.scales[ch.id])
|
|
print(
|
|
f"Channel {ch.id} settings: min: {ch.id}, max: {ch.id} scale:{self.scales[ch.id]}"
|
|
)
|
|
|
|
def getSweep(self, start: float, stop: float, samples):
|
|
f = np.logspace(start, stop, samples, endpoint=True, base=10.0)
|
|
z_array = []
|
|
for i in f:
|
|
# print(f"Frequency: {f}")
|
|
z_array.append(self.getImpedance(i))
|
|
print(z_array)
|
|
return z_array, f
|
|
|
|
def calculateVRMS(self, array):
|
|
return np.sqrt(np.sum(np.pow(array, 2)) / len(array))
|
|
|
|
def PlotBH(self, freq, amplitude):
|
|
setWindowSize(self.osc, 2, freq)
|
|
|
|
self.gen.channels[GEN_CHANNEL].apply_sine(freq, amplitude, self.dc)
|
|
self.gen.channels[GEN_CHANNEL].set_output(True)
|
|
|
|
self.osc.run()
|
|
time.sleep(2)
|
|
|
|
self.osc.single()
|
|
|
|
time.sleep(0.3)
|
|
|
|
voltage_data = self.getScaledWaveform(OSC_CHANNEL_A)
|
|
current_data = self.getScaledWaveform(OSC_CHANNEL_B)
|
|
|
|
time_array = self.osc.getChannel(OSC_CHANNEL_A).genTimeArray(voltage_data)
|
|
|
|
v_l = voltage_data # Voltage induced in the inductor
|
|
|
|
Offset = np.mean(v_l)
|
|
|
|
v_l = v_l - Offset
|
|
|
|
B = (
|
|
cumulative_trapezoid(v_l, time_array, initial=0) / 1
|
|
) # Correct to proper values
|
|
H = current_data # As well as in here
|
|
|
|
plt.plot(H, B)
|
|
self.exportZtoCSV(np.array([B, H]).T, PATH_BH_RAW)
|
|
self.exportZtoCSV(np.array([B, H]).T, PATH_BH)
|
|
|
|
plt.show()
|
|
|
|
def exportZtoCSV(self, z, filename: str):
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
np.savetxt(filename + timestamp + ".csv", z, delimiter=",", fmt="%s")
|
|
|
|
|
|
def setWindowSize(osc, cycles, frequency):
|
|
period = 1 / frequency * cycles / DIV_COUNT
|
|
osc.setTimescale(period)
|
|
|
|
|
|
# def dft(data, time_array, freq):
|
|
# carrier = np.exp(-2j * np.pi * freq * time_array)
|
|
# x = carrier * data
|
|
# print(carrier)
|
|
|
|
# return sum(x) / len(carrier) * 2
|
|
|
|
|
|
def dft(data, time_array, freq):
|
|
sin_ref = np.sin(2 * np.pi * freq * time_array)
|
|
cos_ref = np.cos(2 * np.pi * freq * time_array)
|
|
|
|
i = 2 * np.mean(data * cos_ref)
|
|
q = 2 * np.mean(data * sin_ref)
|
|
|
|
return i - 1j * q
|
|
|
|
|
|
# uv
|
|
def main():
|
|
|
|
# imp = ImpedanceAnalyzer("TCPIP::10.112.1.2::INSTR", "TCPIP::10.112.1.3::INSTR")
|
|
|
|
imp = ImpedanceAnalyzer("TCPIP::192.168.1.2::INSTR", "TCPIP::192.168.1.3::INSTR")
|
|
|
|
imp.gen.channels[GEN_CHANNEL.CH1].set_output(True)
|
|
|
|
# imp.autoscale(1)
|
|
|
|
# imp.autoscale(2)
|
|
|
|
# imp.PlotBH(10e3, AMPLITUDE)
|
|
|
|
# return 0
|
|
imp.autoscale()
|
|
z, f = imp.getSweep(5, 8, 12)
|
|
|
|
z_real = np.real(z)
|
|
z_imag = np.imag(z)
|
|
fig, ax = plt.subplots(3)
|
|
ax[0].plot(f, z_real, label="Real")
|
|
ax[0].plot(f, z_imag, label="Imag")
|
|
ax[0].set_yscale("linear")
|
|
ax[1].plot(f, np.abs(z), label="Magnitude")
|
|
ax[1].set_yscale("log")
|
|
ax_phase = ax[1].twinx()
|
|
ax_phase.plot(
|
|
f, np.angle(z) / np.pi * 180, label="Phase", color="orange", linestyle="--"
|
|
)
|
|
ax_phase.legend()
|
|
ax[0].set_xscale("log")
|
|
ax[1].set_xscale("log")
|
|
|
|
ax[2].plot(f, imp.debug_voltages, label="V_a")
|
|
ax[2].set_xscale("log")
|
|
|
|
ax[0].legend()
|
|
ax[1].legend()
|
|
|
|
imp.exportZtoCSV(np.array([f, z]).T, PATH)
|
|
plt.savefig(PATH + ".png")
|
|
|
|
plt.show()
|
|
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|