PR-2501/scanner_script
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
scanner_script/scanner.py

323 lines
13 KiB
Python
Raw Blame History

import gc
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import re
import telnetlib3
import csv
import time
import pyodbc # XXXDB
import math
from datetime import datetime
import yaml
import os
data_rows = [] # global 2-D list
state = {}
blacklist = []
ALLOWED_NOISE_LEVEL = 100.0
PATTERN = re.compile(
r"Va=(?P<Va>[\d.-]+)\s+Vp=(?P<Vp>[\d.-]+)\s+\|\s+"
r"Ia=(?P<Ia>[\d.-]+)\s+Ip=(?P<Ip>[\d.-]+)\s+\|\s+"
r"ZR=(?P<ZR>[\d.-]+)\s+ZX=(?P<ZX>[\d.-]+).*?\|\s+" # Skips R=...
r".*?irq=\w+\s+" # Skips the first irq value (59)
r"(?P<adc_vmin>[0-9a-fA-F]+)-(?P<adc_vmax>[0-9a-fA-F]+)\s+"
r"(?P<adc_imin>[0-9a-fA-F]+)-(?P<adc_imax>[0-9a-fA-F]+).*?\|\s+"
r"nv=(?P<nv>[\d.-]+)\s+ni=(?P<ni>[\d.-]+)"
)
config_path = '/home/bart/python-scanner/config.yaml' # XXXDB
#config_path = 'config.yaml'
def load_yaml_config(state_dict):
if os.path.exists(config_path):
with open(config_path, 'r') as f:
# Loader=yaml.SafeLoader is the secure way to load YAML
config_data = yaml.load(f, Loader=yaml.SafeLoader)
if config_data:
state_dict.update(config_data)
else:
print(f"Error: {config_path} not found.")
# =============================================
# Database connection parameters
conn_str = (
"DRIVER={FreeTDS};"
"SERVER=10.1.20.18;"
"PORT=1433;"
"DATABASE=H2_HealthMonitoring01;"
"UID=rh\\sa_H2_HealthMon01;"
"PWD=AAo6EM2pttfV5EVZrWqB;"
"TDS_Version=7.4;"
)
def dump_into_database():
try:
with pyodbc.connect(conn_str) as conn:
cursor = conn.cursor()
sweep_insert_time = datetime.now()
print("dump into database\n")
# Prepare the data: SQL Server expects the columns in order.
if state["noise_scan"]==False:
# regular scan data
formatted_rows = [
[sweep_insert_time, r[0], r[5], r[6], r[1], r[2], r[3], r[4], r[11], r[12]]
for r in data_rows
]
sql = """
INSERT INTO SequenceValues (StartTimeOfSweep, Freq, ZR, ZX, Vampl, Vphase, Iampl, Iphase, Vnoise, Inoise)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
"""
else:
# noise scan data
formatted_rows = [
[sweep_insert_time, r[0], r[1], r[3]]
for r in data_rows
]
sql = """
INSERT INTO SequenceNoiseFloor (StartTimeOfSweep, Freq, Vampl, Iampl)
VALUES (?, ?, ?, ?) \
"""
print("execute\n")
# Execute in bulk
cursor.fast_executemany = False # needs less RAM
cursor.executemany(sql, formatted_rows)
print("commit\n")
conn.commit()
print(f"Successfully inserted {len(formatted_rows)} rows.")
except Exception as e:
print(f"Database error: {e}")
def dump_csv(filename="data.csv"):
df = pd.DataFrame(
data_rows,
columns=["Freq", "Va", "Vp", "Ia", "Ip", "ZR", "ZX", "adc_vmin", "adc_vmax", "adc_imin", "adc_imax", "nv", "ni"]
)
df.to_csv(filename, index=False)
def append_line_to_file(column_nr, filename):
# Extract freq column (index 0)
# Extract ZR column (index 5)
# Extract ZX column (index 6)
zr_values = [row[column_nr] for row in data_rows if len(row) > column_nr]
if not zr_values:
print("No ZR data to write")
return
with open(filename, "a", newline="") as f:
writer = csv.writer(f)
writer.writerow(zr_values)
def append_Nyquist_run(filename):
# Flatten ZR/ZX pairs into one row
row = []
for r in data_rows:
if len(r) > 6:
row.extend([r[5], r[6]])
if not row:
print("No ZR/ZX data to write")
return
with open(filename, "a", newline="") as f:
writer = csv.writer(f)
writer.writerow(row)
def extract_to_dataframe(line):
# Extract Va, Vp, Ia, Ip, ZR, ZX from a line and append them as a row to the global data_rows list.
global state
match = PATTERN.search(line)
if not match:
return # silently ignore malformed lines
row = [
float(state["freq"]), # row 0
float(match.group("Va")), # row 1
float(match.group("Vp")), # row 2
float(match.group("Ia")), # row 3
float(match.group("Ip")), # row 4
float(match.group("ZR")), # row 5
float(match.group("ZX")), # row 6
int(match.group("adc_vmin"), 16), # row 7
int(match.group("adc_vmax"), 16), # row 8
int(match.group("adc_imin"), 16), # row 9
int(match.group("adc_imax"), 16), # row 10
float(match.group("nv")), # row 11
float(match.group("ni")), # row 12
]
row[1] = row[1]/(2*3.14159*row[0]*0.00005 + 1) # compensate Va for pole at 20kHz
row[5] = row[1]/row[3] * math.cos(0.01745*(row[2]-row[4]))
row[6] = row[1]/row[3] * math.sin(0.01745*(row[2]-row[4]))
data_rows.append(row)
def process_line(line):
extract_to_dataframe(line)
# Example: print last row
if data_rows:
print("Last row:", data_rows[-1])
def get_dataframe():
return pd.DataFrame(
data_rows,
columns=["Va", "Vp", "Ia", "Ip", "ZR", "ZX", "adc_vmin", "adc_vmax", "adc_imin", "adc_imax", "nv", "ni"]
)
class TelnetReader:
def __init__(self, host, port=23, timeout=10):
self.host = host
self.port = port
self.timeout = timeout
self.tn = None
def connect(self):
print(f"Connecting to {self.host}:{self.port} ...")
self.tn = telnetlib3.Telnet(self.host, self.port, self.timeout)
print("Connected.")
def disconnect(self):
if self.tn:
self.tn.close()
self.tn = None
print("Disconnected.")
def read_loop(self):
global state
# Main loop: reads incoming lines forever
try:
while state["freq"] > state["stop_freq"]:
line = self.tn.read_until(b"\n") # read line
if not line:
break
decoded = line.decode("utf-8", errors="ignore").strip()
self.process_line(decoded) # extract all info from line
except KeyboardInterrupt:
print("Interrupted by user.")
print(data_rows)
if state["freq"]>0.00001: # check if scan ended normally (e.g. no abort due to clipping)
if False:
dump_into_database()
else:
dump_csv("measurements.csv")
#append_line_to_file(0, "scan_freq.csv")
#append_line_to_file(1, "scan_Va.csv")
#append_line_to_file(2, "scan_Vp.csv")
#append_line_to_file(3, "scan_Ia.csv")
#append_line_to_file(4, "scan_Ip.csv")
#append_line_to_file(5, "scan_ZR.csv")
#append_line_to_file(6, "scan_ZX.csv")
#append_Nyquist_run("scan_Nyquist.csv")
#state["noise_scan"] = not state["noise_scan"]
else:
print("scan aborted\r")
gc.collect() # clean up internal memory (garbage collect)
def process_line(self, line):
global state
# print(f"RAW: {line}")
if not line.startswith("Va="): # skip lines that are not to be analyzed
return
if state["remaining_receive_lines"] > 0 :
state["remaining_receive_lines"] -= 1 # we are waiting for settling - just skip the line
else:
# prepare new frequency measurement
if state["initializing"] == 1:
# since we just start a frequency scan, let's set the R, Max and Amplitude
response = f"\rr516\r" # set Resistance value for scaling HAL sensor
self.tn.write(response.encode("utf-8"))
response = f"m1600\r" # set max amplitude value
self.tn.write(response.encode("utf-8"))
# for noise scans the amplitude is zero, else the state["ampl"]
if state["noise_scan"]==True:
response = f"a0\r" # zero amplitude for noise measurement
else:
response = f"a{state["ampl"]:.1f}\r" # set amplitude
self.tn.write(response.encode("utf-8"))
# there will be a lot of lines, but they will be skipped as they do not match the pattern
else: # regular loop (not initializing)
extract_to_dataframe(line) # capture the measurement
if data_rows and (data_rows[-1][11]>ALLOWED_NOISE_LEVEL or data_rows[-1][12]>ALLOWED_NOISE_LEVEL):
# Too much noise: add to blacklist
print("Too much noise - adding to blacklist")
blacklist.append({"Freq": state["freq"], "NrToSkip": 10})
print(f"Blacklist: {blacklist}\r")
del data_rows[-1] # remove this last entry from the list (for DB it is okay, but for CSV things will shift)
# print(f"minmax: {data_rows[7]},{data_rows[8]},{data_rows[9]},{data_rows[10] }\n")
if data_rows and (data_rows[-1][7]<5 or data_rows[-1][8]>250 or data_rows[-1][9]<5 or data_rows[-1][10]>250): # XXXDB
state["freq"] = 0 # force ending of the scan, and write no data in the database
# calculate next freq
if state["freq"] > 1.0:
freq_multiplier = state["freq_step_multiply"] # calc next freq
else:
if state["freq"] > 0.01:
freq_multiplier = state["freq_step_multiply"] ** 2 # skip 1/2 steps to speed up
else:
freq_multiplier = state["freq_step_multiply"] ** 4 # skip 3/4 steps to speed up
state["freq"] *= freq_multiplier
# if (state["freq"]>40) and (state["freq"]<660) and ((state["freq"]%50<2.5) or (-state["freq"]%50<2.5)):
# state["freq"] *= freq_multiplier # if near a 50Hz harmonic, skip to the next frequency
# while (state["freq"]<0.6) and ((state["freq"]%state["interference_freq"]<state["interference_bandwidth"]) or (-state["freq"]<state["interference_freq"]<state["interference_bandwidth"])):
# state["freq"] *= freq_multiplier # if near a 0.052Hz harmonic, skip to the next frequency
while (any(item["Freq"] == state["freq"] for item in blacklist)):
print("skipping freq")
state["freq"] *= freq_multiplier # skip all frequencies that are blacklisted
if state["freq"] <= state["stop_freq"]:
print("Reached stop frequency")
else:
# program the health monitor to go to the next frequency:
if state["freq"] > 0.01:
response = f"\rf{state["freq"]:.1f}\r" # new freq
else:
response = f"\rf{state["freq"]:.3f}\r" # new freq
self.tn.write(response.encode("utf-8"))
bandwidth = state["freq"]/20
if bandwidth > 0.5 :
bandwidth = 0.5
response = f"b{bandwidth:.3f}\r" # new freq
self.tn.write(response.encode("utf-8"))
state["remaining_receive_lines"] = 2 + 4/bandwidth
print(line)
# print(state)
state["initializing"] = 0
# Example:
# value = self.extract_value(line)
# self.store_value(value)
# Example placeholder methods
def extract_value(self, line):
pass
def store_value(self, value):
pass
if __name__ == "__main__":
reader = TelnetReader(host="localhost", port=2002)
# reader = TelnetReader(host="10.1.122.152", port=2002)
# reader = TelnetReader(host="192.168.1.235", port=2002)
try:
reader.connect()
while True:
data_rows.clear()
load_yaml_config(state) # Load configuration settings
state["freq"] = state["start_freq"];
state["remaining_receive_lines"] = 0;
state["initializing"] = 1;
reader.read_loop()
# decrement all freq's in blacklist:
print(f"Before decrementing: {blacklist}")
for item in blacklist:
item["NrToSkip"] -= 1
blacklist = [item for item in blacklist if item["NrToSkip"] != 0] # remove all zero items
print(f"After removing zeros: {blacklist}")
finally:
reader.disconnect()
Reference in a new issue View git blame Copy permalink