Compare commits

...

2 Commits

Author SHA1 Message Date
Cal W
bdf25d5736 Add canny args to _info.yaml 2024-10-18 00:03:22 +10:00
Cal W
ba1136b0ea quick data load refactor 2024-10-17 22:58:30 +10:00
6 changed files with 256 additions and 210 deletions

View File

@ -12,6 +12,11 @@ shot-info:
config: "x2s5823.config"
info: "x2s5823.txt"
# Canny Args
canny-args:
- sigma: 2
post_pres: 0.03
pcb-refs:
- "st1"
- "st3"

View File

@ -12,6 +12,11 @@ shot-info:
config: "x2s5824.config"
info: "x2s5824.txt"
# Canny Args
canny-args:
- sigma: 2
post_pres: 0.03
pcb-refs:
- "st1"
- "st3"

View File

@ -12,6 +12,14 @@ shot-info:
config: "x2s5827.config"
info: "x2s5827.txt"
# Canny Args
canny-args:
- sigma: 2
post_pres: 0.03
- sigma: 0.75
post_pres: 0.1
pcb-refs:
- "st1"

View File

@ -12,6 +12,14 @@ shot-info:
config: "x2s5829.config"
info: "x2s5829.txt"
# Canny Args
canny-args:
- sigma: 2
post_pres: 0.03
- sigma: 1
post_pres: 0.2
pcb-refs:
- "st1"

92
main.py
View File

@ -24,27 +24,25 @@ DATA_PATH = "./data"
DATA_INFO = "_info.yaml"
TUNNEL_INFO_FILE = "./tunnel-info.yaml"
SAMPLES_TO_AVG = 500
CANNY_TIME_OFFSET = 50 #us
with open(TUNNEL_INFO_FILE, 'r') as file:
TUNNEL_INFO = yaml.safe_load(file)
data_to_load = [
#"x2s5823",
#"x2s5824",
#"x2s5827",
"x2s5823",
"x2s5824",
"x2s5827",
"x2s5829",
]
# ==== Data Loading & Processing ====
def load_data(data_to_load: list[str]) -> dict:
data = {}
for dp in data_to_load:
data_path = f"{DATA_PATH}/{dp}/"
def load_data(data_path: str, data={}) -> dict:
data_info_path = data_path + DATA_INFO
if not os.path.exists(data_info_path):
print(f"[ERR] Could not find data info file: '{data_info_path}'")
print(f"[WARN] Not Loading Data '{dp}'")
continue
print(f"[WARN] Not Loading Data '{data_path}'")
return None
# Load Shot Data Info YAML File (Cal)
with open(data_info_path, 'r') as file:
@ -106,7 +104,8 @@ def load_data(data_to_load: list[str]) -> dict:
},
"data": {
"x2": {} # Only pop channels with a voltage scale in ./tunnel-info.yaml
}
},
"shock-speed": {}
}
# === Process the data ===
@ -191,38 +190,36 @@ def load_data(data_to_load: list[str]) -> dict:
probeCh1 = data[x2_shot]["data"]["scope"][1]
probeCh2 = data[x2_shot]["data"]["scope"][2]
#[HACK] For detection
TIME_OFFSET = 50 #us
# Get the canny-args
cArgs = dataInfo["canny-args"]
doCannyPlot = False
doCannyPlot = i > 0 and x2_shot == "x2s5827"
if i in range(len(cArgs)):
sigma = cArgs[i]["sigma"]
post_pres = cArgs[i]["post_pres"]
else:
sigma = cArgs[-1]["sigma"]
post_pres = cArgs[-1]["post_pres"]
# If this _isn't_ the first probe then apply a time offset
if i > 0:
privPoint = dataInfo["probe-info"]["locations"][i-1]
time_offset = data[x2_shot]["shock-point"][f"{privPoint}-g1"][1] + TIME_OFFSET
post_pres = 0.03
sigma = 7
doPlot = True
else: # These work for the first probe int he chain
time_offset = data[x2_shot]["shock-point"][f"{privPoint}-g1"][1] + CANNY_TIME_OFFSET
else:
time_offset = None
post_pres = 0.03
sigma = 2
doPlot = False
shock_point = np.where(probeCh1 >= 0.3)[0]
first_value = scope_time[shock_point]
first_value, first_value_uncertainty, _, _ = canny_shock_finder(scope_time, probeCh1, sigma=sigma, post_suppression_threshold=post_pres, plot=doPlot, start_time=time_offset, print_func=None)
# Find G1 Shock Time
first_value, first_value_uncertainty, _, _ = canny_shock_finder(scope_time, probeCh1, sigma=sigma, post_suppression_threshold=post_pres, plot=doCannyPlot, start_time=time_offset, print_func=None)
shock_point = np.where(scope_time >= first_value)[0][0] # [BUG] Seems to give n+1
data[x2_shot]["shock-point"][f"{probe}-g1"] = shock_point, first_value
#[HACK] For detection
# Do the same for G2
if i > 0:
time_offset = data[x2_shot]["shock-point"][f"{privPoint}-g2"][1] + TIME_OFFSET
time_offset = data[x2_shot]["shock-point"][f"{privPoint}-g2"][1] + CANNY_TIME_OFFSET
shock_point = np.where(probeCh2 >= 0.3)[0] # + offset
first_value = scope_time[shock_point]
first_value, first_value_uncertainty, _, _ = canny_shock_finder(scope_time, probeCh2, sigma=sigma, post_suppression_threshold=post_pres, plot=doPlot, start_time=time_offset, print_func=None)
# Find G2 Shock Time
first_value, first_value_uncertainty, _, _ = canny_shock_finder(scope_time, probeCh2, sigma=sigma, post_suppression_threshold=post_pres, plot=doCannyPlot, start_time=time_offset, print_func=None)
shock_point = np.where(scope_time >= first_value)[0][0] # [BUG] Seems to give n+1
data[x2_shot]["shock-point"][f"{probe}-g2"] = shock_point, first_value
@ -236,6 +233,7 @@ def load_data(data_to_load: list[str]) -> dict:
probe_velocity = c2c_dist / abs(g2_time - g1_time) # m/s
print(f"{probe} Measured a shock speed of {probe_velocity:.2f} m/s ({probe_velocity/1000:.2f} km/s)")
data[x2_shot]["shock-speed"][probe] = probe_velocity # m/s
if len(dataInfo["probe-info"]["locations"]) > 1:
for i in range(len(dataInfo["probe-info"]["locations"]) - 1):
@ -253,12 +251,21 @@ def load_data(data_to_load: list[str]) -> dict:
print(f"{probe_locs[i]}-{probe_locs[i + 1]} - G1 - Measured a shock speed of {p2p_1:.2f} m/s ({p2p_1/1000:.2f} km/s)")
print(f"{probe_locs[i]}-{probe_locs[i + 1]} - G2 - Measured a shock speed of {p2p_2:.2f} m/s ({p2p_2/1000:.2f} km/s)")
data[x2_shot]["shock-speed"][f"{probe_locs[i]}-{probe_locs[i + 1]}-g1"] = p2p_1
data[x2_shot]["shock-speed"][f"{probe_locs[i]}-{probe_locs[i + 1]}-g2"] = p2p_2
print()
# Return the data & the successfully loaded data keys
return data, tuple(data.keys())
return data #, tuple(data.keys())
data = {}
for dp in data_to_load:
pdp = f"{DATA_PATH}/{dp}/"
load_data(pdp, data)
loaded_data = tuple(data.keys())
data, loaded_data = load_data(data_to_load)
print("Loaded Data")
@ -269,6 +276,8 @@ def genGraph(gData: dict, showPlot: bool = True):
"xLabel": "Time ($\\mu$s)",
"yLabel": "Voltage Reading (V)",
"grid": True,
"figSize": (8,6.5),
"ledgLoc": 'upper left',
"plots": []
}
@ -318,9 +327,20 @@ def genGraph(gData: dict, showPlot: bool = True):
lims.append(gData["shock-point"][f"{probe}-g2"][1])
lims.append(gData["shock-point"][f"{probe}-g1"][1])
probeText = ""
for shock_speed_loc in gData["shock-speed"]:
probeText += f"\n{shock_speed_loc} - {gData["shock-speed"][shock_speed_loc]/1000:.2f} km/s"
graphData["plots"].append({
"type": "text",
"text": f"Measured Shock Speeds{probeText}",
"align": ("top", "right"),
"x": 0.94, "y": 0.94
})
if len(lims) > 1:
OFFSET = 10
#graphData["xLim"] = (float(min(lims) - OFFSET), float(max(lims) + OFFSET))
graphData["xLim"] = (float(min(lims) - OFFSET), float(max(lims) + OFFSET))
makeGraph(graphData, doProgramBlock=False, showPlot=showPlot, figSavePath="./images/{0}.png")
@ -329,7 +349,7 @@ def genGraph(gData: dict, showPlot: bool = True):
print("Graphing Data")
for shot in loaded_data:
#if shot != loaded_data[-1]: continue
if shot != loaded_data[-2]: continue
genGraph(data[shot], showPlot=False)
# This forces matplotlib to hang until I tell it to close all windows

View File

@ -378,8 +378,8 @@ def makeGraph(graphData, showPlot=True, doProgramBlock=True, figSavePath=None, h
}
align = (
getSafeValue("valign", None),
getSafeValue("halign", None),
getSafeValue("valign", "center"),
getSafeValue("halign", "center"),
)
align = getSafeValue("align", align)