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

oud algoritme van 17-mrt-2026 maar dan met versienummer support

Browse source
This commit is contained in:
gertjan 2026-04-03 16:19:30 +02:00
parent f0cc0a2b93
commit 73e8582835
2 changed files with 56 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

11
config.yaml
View file

@ -1,9 +1,12 @@
# Configuration file for scanner service # Configuration file for scanner service
# it is automatically re-loaded between scans # it is automatically re-loaded between scans
start_freq: 1000 start_freq: 4000
stop_freq: 0.2 stop_freq: 0.4
freq_step_multiply: 0.95 freq_step_multiply: 0.95
allowed_noise_level: 9 allowed_noise_level: 15
skip_scans_for_noisy_freqs: 5 skip_scans_for_noisy_freqs: 3
ampl: 1600 ampl: 1600
noise_scan: False noise_scan: False
hw_version: "2026-04-02" # dit was met 3.33 Ohm weerstand voor stromen boven 1000A
scan_comment: "Paasweekend scan met varierende stoom levels"

79
scanner.py
View file

@ -12,23 +12,22 @@ from datetime import datetime
import yaml import yaml
import os import os
VersionScripSoftware = "2026-03-17"
VersionDspSoftware = "Unknown"
data_rows = [] # global 2-D list data_rows = [] # global 2-D list
state = {} state = {}
blacklist = [] blacklist = []
Inoise_baseline = 0.1 Inoise_baseline = 0.1
PATTERN = re.compile( PATTERN = re.compile(
r"Va=(?P<Va>-?\d+\.?\d*)\s+" r"Va=(?P<Va>[\d.-]+)\s+Vp=(?P<Vp>[\d.-]+)\s*\|\s*"
r"Vp=(?P<Vp>-?\d+\.?\d*)\s+\|\s+" r"Ia=(?P<Ia>[\d.-]+)\s+Ip=(?P<Ip>[\d.-]+)\s*\|\s*"
r"Ia=(?P<Ia>-?\d+\.?\d*)\s+" r"ZR=(?P<ZR>[\d.-]+)\s+ZX=(?P<ZX>[\d.-]+).*?\|\s*"
r"Ip=(?P<Ip>-?\d+\.?\d*)\s+\|\s+" r".*?irq=\w+\s+"
r"ZR=(?P<ZR>-?\d+\.?\d*)\s+" r"(?P<adc_vmin>[0-9a-fA-F]+)-(?P<adc_vmax>[0-9a-fA-F]+)\s+"
r"ZX=(?P<ZX>-?\d+\.?\d*)" r"(?P<adc_imin>[0-9a-fA-F]+)-(?P<adc_imax>[0-9a-fA-F]+).*?\|\s*"
r".*?irq=\d+\s+" # Skip to 'irq=', match the first digits and space r"nv=(?P<nv>[\d.-]+)\s+ni=(?P<ni>[\d.-]+)\s*\|\s*"
r"(?P<adc_vmin>[0-9a-fA-F]+)-" # First hex
r"(?P<adc_vmax>[0-9a-fA-F]+)\s+" # Second hex
r"(?P<adc_imin>[0-9a-fA-F]+)-" # Third hex
r"(?P<adc_imax>[0-9a-fA-F]+)" # Fourth hex
) )
config_path = '/home/bart/python-scanner/config.yaml' config_path = '/home/bart/python-scanner/config.yaml'
@ -56,11 +55,26 @@ conn_str = (
) )
def dump_into_database(): def dump_into_database():
print("dump into database\n")
try: try:
with pyodbc.connect(conn_str) as conn: with pyodbc.connect(conn_str) as conn:
cursor = conn.cursor() cursor = conn.cursor()
sweep_insert_time = datetime.now() sweep_insert_time = datetime.now()
print("dump into database\n") # Add scan details
cursor.execute("""
IF NOT EXISTS (
SELECT 1 FROM ScanDetails WHERE StartTimeOfSweep = ?
)
INSERT INTO ScanDetails
(StartTimeOfSweep, VersionScanScript, VersionDspSoftware, VersionHardware, Comment)
VALUES (?, ?, ?, ?, ?)
""", sweep_insert_time,
sweep_insert_time, VersionScripSoftware, VersionDspSoftware, state["hw_version"], state["scan_comment"])
print("commit1\n")
conn.commit()
with pyodbc.connect(conn_str) as conn:
cursor = conn.cursor()
# Prepare the data: SQL Server expects the columns in order. # Prepare the data: SQL Server expects the columns in order.
if state["noise_scan"]==False: if state["noise_scan"]==False:
# regular scan data # regular scan data
@ -89,7 +103,7 @@ def dump_into_database():
cursor.fast_executemany = False # needs less RAM cursor.fast_executemany = False # needs less RAM
cursor.executemany(sql, formatted_rows) cursor.executemany(sql, formatted_rows)
print("commit\n") print("commit2\n")
conn.commit() conn.commit()
print(f"Successfully inserted {len(formatted_rows)} rows.") print(f"Successfully inserted {len(formatted_rows)} rows.")
@ -136,17 +150,19 @@ def extract_to_dataframe(line):
if not match: if not match:
return # silently ignore malformed lines return # silently ignore malformed lines
row = [ row = [
float(state["freq"]), float(state["freq"]), # row 0
float(match.group("Va")), float(match.group("Va")), # row 1
float(match.group("Vp")), float(match.group("Vp")), # row 2
float(match.group("Ia")), float(match.group("Ia")), # row 3
float(match.group("Ip")), float(match.group("Ip")), # row 4
float(match.group("ZR")), float(match.group("ZR")), # row 5
float(match.group("ZX")), float(match.group("ZX")), # row 6
int(match.group("adc_vmin"), 16), int(match.group("adc_vmin"), 16), # row 7
int(match.group("adc_vmax"), 16), int(match.group("adc_vmax"), 16), # row 8
int(match.group("adc_imin"), 16), int(match.group("adc_imin"), 16), # row 9
int(match.group("adc_imax"), 16), 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[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[5] = row[1]/row[3] * math.cos(0.01745*(row[2]-row[4]))
@ -219,7 +235,10 @@ class TelnetReader:
def process_line(self, line): def process_line(self, line):
global state global state
global Inoise_baseline global Inoise_baseline
# print(f"RAW: {line}") global VersionDspSoftware
if line.startswith("SW-Version"):
VersionDspSoftware = line.split(":", 1)[1].strip() # remember the DSP software version from its reporting
print(f"DSP SW Version={VersionDspSoftware}\r\n")
if not line.startswith("Va="): # skip lines that are not to be analyzed if not line.startswith("Va="): # skip lines that are not to be analyzed
return return
if state["remaining_receive_lines"] > 0 : if state["remaining_receive_lines"] > 0 :
@ -228,7 +247,7 @@ class TelnetReader:
# prepare new frequency measurement # prepare new frequency measurement
if state["initializing"] == 1: if state["initializing"] == 1:
# since we just start a frequency scan, let's set the R, Max and Amplitude # 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 response = f"\rr344\r" # set Resistance value for scaling HAL sensor
self.tn.write(response.encode("utf-8")) self.tn.write(response.encode("utf-8"))
response = f"m1600\r" # set max amplitude value response = f"m1600\r" # set max amplitude value
self.tn.write(response.encode("utf-8")) self.tn.write(response.encode("utf-8"))
@ -263,8 +282,8 @@ class TelnetReader:
state["freq"] *= freq_multiplier state["freq"] *= freq_multiplier
if (state["freq"]>40) and (state["freq"]<660) and ((state["freq"]%50<2.5) or (-state["freq"]%50<2.5)): 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 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"])): # 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 # state["freq"] *= freq_multiplier # if near a 0.052Hz harmonic, skip to the next frequency
if state["freq"] <= state["stop_freq"]: if state["freq"] <= state["stop_freq"]:
print("Reached stop frequency") print("Reached stop frequency")
else: else:
@ -279,8 +298,8 @@ class TelnetReader:
response = f"b{bandwidth:.3f}\r" # new freq response = f"b{bandwidth:.3f}\r" # new freq
self.tn.write(response.encode("utf-8")) self.tn.write(response.encode("utf-8"))
state["remaining_receive_lines"] = 1 + 4/bandwidth state["remaining_receive_lines"] = 1 + 4/bandwidth
print(line) print(line)
print(state) print(state)
state["initializing"] = 0 state["initializing"] = 0
# Example: # Example: