In [1]:
import os, math
from dataclasses import dataclass
import numpy as np
import pandas as pd
from numba import vectorize, float64, boolean, njit
from scipy.signal import find_peaks
import matplotlib.pyplot as plt
import holoviews as hv
from IPython.display import Markdown
from holoviews.operation import decimate
from holoviews.operation.datashader import datashade,rasterize, dynspread
import dask.dataframe as dd
# import hvplot.pandas
# import hvplot.dask
hv.extension('bokeh', logo = False)
In [2]:
# turn off numpy warning for overflow in np.exp()
np.seterr(over='ignore');
In [3]:
shot_no = 0
(Internal Tektronix format) .wfm parser¶
In [4]:
# wfm reader proof-of-concept
# https://www.tek.com/sample-license
# reads volts vs. time records (including fastframes) from little-endian version 3 WFM files
# See Also
# Performance Oscilloscope Reference Waveform File Format
# Tektronix part # 077-0220-10
# https://www.tek.com/oscilloscope/dpo7000-digital-phosphor-oscilloscope-manual-4
import struct
import time
import numpy as np # http://www.numpy.org/
class WfmReadError(Exception):
"""error for unexpected things"""
pass
def read_wfm(target):
"""return sample data from target WFM file"""
with open(target, 'rb') as f:
hbytes = f.read(838)
meta = decode_header(hbytes)
# file signature checks
if meta['byte_order'] != 0x0f0f:
raise WfmReadError('big-endian not supported in this example')
if meta['version'] != b':WFM#003':
raise WfmReadError('only version 3 wfms supported in this example')
if meta['imp_dim_count'] != 1:
raise WfmReadError('imp dim count not 1')
if meta['exp_dim_count'] != 1:
raise WfmReadError('exp dim count not 1')
if meta['record_type'] != 2:
raise WfmReadError('not WFMDATA_VECTOR')
if meta['exp_dim_1_type'] != 0:
raise WfmReadError('not EXPLICIT_SAMPLE')
if meta['time_base_1'] != 0:
raise WfmReadError('not BASE_TIME')
tfrac_array = np.zeros(meta['Frames'], dtype=np.double)
tdatefrac_array = np.zeros(meta['Frames'], dtype=np.double)
tdate_array = np.zeros(meta['Frames'], dtype=np.int32)
tfrac_array[0] = meta['tfrac']
tdatefrac_array[0] = meta['tdatefrac']
tdate_array[0] = meta['tdate']
# if fastframe, read fastframe table
if meta['fastframe'] == 1:
WUSp = np.fromfile(f, dtype='i4,f8,f8,i4', count=(meta['Frames'] - 1))
# merge first frame trigger infos with frames > 1
tfrac_array[1:] = WUSp['f1']
tdatefrac_array[1:] = WUSp['f2']
tdate_array[1:] = WUSp['f3']
# read curve block
bin_wave = np.memmap(filename = f,
dtype = meta['dformat'],
mode = 'r',
offset = meta['curve_offset'],
shape = (meta['avilable_values'], meta['Frames']),
order = 'F')
# close file
# slice out buffer values
bin_wave = bin_wave[meta['pre_values']:meta['avilable_values'] - meta['post_values'],:]
scaled_array = bin_wave * meta['vscale'] + meta['voffset']
return scaled_array, meta['tstart'], meta['tscale'], tfrac_array, tdatefrac_array, tdate_array
def decode_header(header_bytes):
"""returns a dict of wfm metadata"""
wfm_info = {}
if len(header_bytes) != 838:
raise WfmReadError('wfm header bytes not 838')
wfm_info['byte_order'] = struct.unpack_from('H', header_bytes, offset=0)[0]
wfm_info['version'] = struct.unpack_from('8s', header_bytes, offset=2)[0]
wfm_info['imp_dim_count'] = struct.unpack_from('I', header_bytes, offset=114)[0]
wfm_info['exp_dim_count'] = struct.unpack_from('I', header_bytes, offset=118)[0]
wfm_info['record_type'] = struct.unpack_from('I', header_bytes, offset=122)[0]
wfm_info['exp_dim_1_type'] = struct.unpack_from('I', header_bytes, offset=244)[0]
wfm_info['time_base_1'] = struct.unpack_from('I', header_bytes, offset=768)[0]
wfm_info['fastframe'] = struct.unpack_from('I', header_bytes, offset=78)[0]
wfm_info['Frames'] = struct.unpack_from('I', header_bytes, offset=72)[0] + 1
wfm_info['summary_frame'] = struct.unpack_from('h', header_bytes, offset=154)[0]
wfm_info['curve_offset'] = struct.unpack_from('i', header_bytes, offset=16)[0] # 838 + ((frames - 1) * 54)
# scaling factors
wfm_info['vscale'] = struct.unpack_from('d', header_bytes, offset=168)[0]
wfm_info['voffset'] = struct.unpack_from('d', header_bytes, offset=176)[0]
wfm_info['tstart'] = struct.unpack_from('d', header_bytes, offset=496)[0]
wfm_info['tscale'] = struct.unpack_from('d', header_bytes, offset=488)[0]
# trigger detail
wfm_info['tfrac'] = struct.unpack_from('d', header_bytes, offset=788)[0] # frame index 0
wfm_info['tdatefrac'] = struct.unpack_from('d', header_bytes, offset=796)[0] # frame index 0
wfm_info['tdate'] = struct.unpack_from('I', header_bytes, offset=804)[0] # frame index 0
# data offsets
# frames are same size, only first frame offsets are used
dpre = struct.unpack_from('I', header_bytes, offset=822)[0]
wfm_info['dpre'] = dpre
dpost = struct.unpack_from('I', header_bytes, offset=826)[0]
wfm_info['dpost'] = dpost
readbytes = dpost - dpre
wfm_info['readbytes'] = readbytes
allbytes = struct.unpack_from('I', header_bytes, offset=830)[0]
wfm_info['allbytes'] = allbytes
# sample data type detection
code = struct.unpack_from('i', header_bytes, offset=240)[0]
wfm_info['code'] = code
bps = struct.unpack_from('b', header_bytes, offset=15)[0] # bytes-per-sample
wfm_info['bps'] = bps
if code == 7 and bps == 1:
dformat = 'int8'
samples = readbytes
elif code == 0 and bps == 2:
dformat = 'int16'
samples = readbytes // 2
elif code == 4 and bps == 4:
dformat = 'single'
samples = readbytes // 4
else:
raise WfmReadError('data type code or bytes-per-sample not understood')
wfm_info['dformat'] = dformat
wfm_info['samples'] = samples
wfm_info['avilable_values'] = allbytes // bps
wfm_info['pre_values'] = dpre // bps
wfm_info['post_values'] = (allbytes - dpost) // bps
return wfm_info
def readwfm(path):
volts, tstart, tscale, tfrac, tdatefrac, tdate = read_wfm(path)
toff = tfrac * tscale
samples, frames = volts.shape
tstop = samples * tscale + tstart
volts = volts.reshape(len(volts))
time = np.linspace(tstart+toff, tstop+toff, num=samples, endpoint=False)
time = time.reshape(len(volts))
return time,volts
Get Hi-Res data from MSO58 oscilloscope¶
In [5]:
# load data locally
fnames = ['DAS_raw_data_dir/ch2.wfm',
'DAS_raw_data_dir/ch3.wfm',
'DAS_raw_data_dir/ch4.wfm',
'DAS_raw_data_dir/ch5.wfm',
'DAS_raw_data_dir/ch6.wfm',
'DAS_raw_data_dir/ch7.wfm']
ds = np.DataSource('/tmp/')
if not os.path.isfile('DAS_raw_data_dir/ch6.wfm'):
# or download from web (slow)
for i, ch_no in enumerate(range(2,8)):
try:
file = ds.open(f'http://golem.fjfi.cvut.cz/shots/{shot_no}/Devices/Oscilloscopes/TektrMSO58-a/ch{ch_no}.wfm', mode = 'rb')
fnames[i] = file.name
file.close()
except (ConnectionError, FileNotFoundError):
err = Markdown('### Can not download data')
t_vec, y_ch2 = readwfm(fnames[0])
_, y_ch3 = readwfm(fnames[1])
_, y_ch4 = readwfm(fnames[2])
_, y_ch5 = readwfm(fnames[3])
_, y_ch6 = readwfm(fnames[4])
_, y_ch7 = readwfm(fnames[5])
Check if data has sufficient sample rate
In [6]:
dt = t_vec[1]-t_vec[0]
MSamples = 1e-6/(dt)
downsample = int(np.rint(float(1e-6) / dt))
if MSamples >= 120:
out = Markdown(f'### Data with {MSamples:.1f}MS/s resolution')
low_res_data = False
else:
out = Markdown('### Low resolution for peak recononstruction \n adjust oscilloscope settings')
low_res_data = True
out
Out[6]:
Data with 250.0MS/s resolution¶
In [7]:
# class for storing channels and their parameters and results
@dataclass
class channel:
name : str
time : np.ndarray
y : np.ndarray
noPlot : bool = False
noFit : bool = False
## parameters for peak detection
mean_noise_threshold_mult : float = 1.75
peak_prominence_thr : float = 0.001
# peak rise / fall time
rise_time : float = None
fall_time : float = None
# calibration
calibration : float = None # keV/V
# Flip waveform of PMTs but no SiPMs
flip_waveform : bool = True
## parameters recieved from further analysis
mean_channel_noise : float = 0
# threshold
threshold : float = 0
# arrays for results
simple_peak_loc : np.ndarray = None
lonely_peak_loc : np.ndarray = None
# for debug
lonely_peak_mask : np.ndarray = None
no_corr_peak_height : np.ndarray = None
lonely_peak_height : np.ndarray = None
simple_corr_peak_height : np.ndarray = None
simple_corr_peak_err : np.ndarray = None
simple_corr_peak_filtered_loc : np.ndarray = None
simple_corr_peak_filtered_height : np.ndarray = None
multipass_peak_loc : np.ndarray = None
multipass_corr_peak_height : np.ndarray = None
multipass_corr_peak_err : np.ndarray = None
multipass_corr_peak_filtered_loc : np.ndarray = None
multipass_corr_peak_filtered_height : np.ndarray = None
y_diff : np.ndarray = None
y_gen : np.ndarray = None
y_gen2 : np.ndarray = None
@property
def out_fname(self):
return f'{ch.name}_{shot_no}'
In [8]:
# set channel and their parameters
ch2 = channel('YaP', t_vec, y_ch2, noFit=False) # @ HV:600 V
ch3 = channel('CeBr-a', t_vec, y_ch3, rise_time=1.457e-09, fall_time=2.754e-08, calibration=12578.) # @ HV:600 V
ch4 = channel('CeBr-b', t_vec, y_ch4, rise_time=3.410e-09, fall_time=2.786e-08, calibration=6714., noFit= True) # @ HV:600 V
ch5 = channel('NaITl', t_vec, y_ch5, rise_time=2.258e-08, fall_time=2.202e-07, calibration=71799., noFit = True) # @ HV:600 V
ch6 = channel('LYSO-1', t_vec, y_ch6, rise_time=5.946e-09, fall_time=5.146e-08, calibration=5007., flip_waveform=False) # @ C:300mV
ch7 = channel('GaGG', t_vec, y_ch7, flip_waveform=False, noFit=True)
channels = [ch3,ch4,ch5,ch6,ch7,ch2]
In [9]:
color_cycle = plt.rcParams['axes.prop_cycle'].by_key()['color']
fig, axes = plt.subplots(len(channels), sharex =True)
[ax.plot(ch.time[::downsample],ch.y[::downsample],
label = ch.name, color = c) for ax, ch, c in zip(axes, channels, color_cycle)]
[ax.axhline(1., color = 'red', ls = '--')
for ax, ch in zip(axes, channels) if not ch.flip_waveform and np.any(ch.y > 1.)]
fig.legend()
is_clipping = [np.any(ch.y > 1.) for ch in channels if not ch.flip_waveform]
if np.any(is_clipping):
warning = Markdown("""### SiPM's integrated amplifier saturated
HXR spectrum reconstruction from whole discharge may be unreliable
""")
display(warning)
SiPM's integrated amplifier saturated¶
HXR spectrum reconstruction from whole discharge may be unreliable
Remove offset from OSC waveforms and flip¶
In [10]:
def get_noise_level(t,y_vals,noise_time):
y_noise = y_vals[t < noise_time]
y_noise = np.nan_to_num(y_noise, posinf = 0, neginf = 0)
mean_y = np.mean(y_noise)
# mean noise amplitude
abs_noise = np.abs(y_noise - mean_y)**2
mean_noise_ampl = np.sqrt(np.mean(abs_noise))
return mean_y, mean_noise_ampl
def moving_average(x, w):
return np.convolve(x, np.ones(w), 'save') / w
for ch in channels:
noise, ch.mean_channel_noise = get_noise_level(ch.time, ch.y, 0.000)
ch.y = ch.y - noise
ch.threshold = ch.mean_channel_noise*ch.mean_noise_threshold_mult
if ch.flip_waveform:
ch.y *=-1
## smooth-out noise
ch.y = moving_average(ch.y, 5)
Plot OSC waveforms and peak thresholds¶
In [11]:
#%matplotlib widget
fig, axes = plt.subplots(len(channels), sharex =True)
for ax,ch,color in zip(axes,channels, color_cycle):
ax.plot(ch.time[::downsample]*1e3, ch.y[::downsample], label = ch.name, color=color)
ax.axhline(ch.mean_channel_noise,ls= '-.', c = 'k')
ax.axhline(ch.threshold,ls= '--', c = 'k')
ax.set_ylim(-2*ch.threshold, 2*ch.threshold)
ax.grid()
fig.legend();
Find peaks¶
In [12]:
max_HXR_energy = 10.
for ch in channels:
peaks_loc, _ = find_peaks(ch.y, height=ch.threshold, prominence = ch.mean_channel_noise*2)
ch.simple_peak_loc = peaks_loc
ch.no_corr_peak_height = ch.y[peaks_loc]
if ch.calibration is not None and not ch.noPlot and not ch.noFit:
if ch.name == 'NaITl':
continue
max_HXR_energy = np.max((max_HXR_energy, ch.no_corr_peak_height.max()*ch.calibration))
Define useful functions¶
In [13]:
@vectorize(['float64(float64, boolean)'])
def half_exp(x, where):
'''compute exp function only for selected indices (to avoid infinite/large values)'''
if where:
return np.exp(x)
else:
return 0.0
def get_index(oneDArray, val):
diff = np.abs(oneDArray - val)
return np.argmin(diff)
def exp_decay(t, height, center, rise, fall):
'''smooth peak fucntion from https://arxiv.org/pdf/1706.01211.pdf eq:(2)'''
# bool mask where compute exp functions (without it results in +-inf values)
exp_window_split_index = get_index(t, center-100*rise)
exp_window = np.zeros_like(t, np.bool_)
exp_window[exp_window_split_index:].fill(True)
# scaling and shifting of peak fucntion
A = height / ( (fall-rise)/fall * (fall/rise - 1 )**(-rise/fall) )
peak_center_shift = rise*np.log((fall-rise)/rise)
center = center - peak_center_shift
peak = A * 1/(1+half_exp(-(t-center)/rise, exp_window)) * half_exp(-(t-center)/fall, exp_window)
return peak
def gen_peaks(time, HXR_y, peaks_loc, channel_props, test_diff: bool = True):
corrected_peak_heights = np.zeros_like(peaks_loc, float)
HXR_generated = np.zeros_like(HXR_y)
diff = HXR_y.copy()
discarded_peaks = np.zeros_like(peaks_loc, np.bool_)
# operate only on small slice of waveform (it's faster)
dt = np.mean(np.diff(time))
i_slice = max(int(channel_props.fall_time*100 / dt),1000)
for n, peak in enumerate(peaks_loc):
# use differece between generated and original waveform to obtain peak height
peak_height = diff[peak]
peak_center = time[peak]
slice_idx_left = max(peak-i_slice,0)
slice_idx_right = min(peak+i_slice,len(HXR_y)-1)
gen_slice_time = time[slice_idx_left:slice_idx_right]
if peak_height < 0.:
corrected_peak_heights[n] = 0.
continue
# generate peak
generated_peak = exp_decay(gen_slice_time, peak_height, peak_center,
channel_props.rise_time, channel_props.fall_time)
# if adding this peak increases differance btw original and generated signal
# discard this peak as false detection
diff_test = diff[slice_idx_left:slice_idx_right].copy()
diff_test -= generated_peak
if test_diff and np.sum(np.abs(diff_test)) > np.sum(np.abs(diff[slice_idx_left:slice_idx_right])):
corrected_peak_heights[n] = 0.
discarded_peaks[n] = True
else:
# subtract peak from signal and record peak height
diff[slice_idx_left:slice_idx_right] -= generated_peak
HXR_generated[slice_idx_left:slice_idx_right] += generated_peak
corrected_peak_heights[n] = peak_height
return corrected_peak_heights, HXR_generated, diff, discarded_peaks
@njit
def is_lonely_peak(time, HXR_y, peaks, peak_no , fall_time, mean_noise_level):
peak_fall_time = fall_time * np.log(HXR_y[peaks[peak_no]]/mean_noise_level)
if peak_no != 0:
t_before = time[peaks[peak_no]] - peak_fall_time
t_peak_before = time[peaks[peak_no-1]]
if t_before < t_peak_before:
return False
if peak_no <= peaks.size-2:
t_after = time[peaks[peak_no]] + peak_fall_time
t_peak_after = time[peaks[peak_no+1]]
if t_peak_after < t_after:
return False
return True
def identify_lonely_peaks(time,HXR_y, peaks, channel_props):
def is_lonely_peak_reduced(n):
return is_lonely_peak(time,
HXR_y,
peaks,
n,
channel_props.fall_time,
channel_props.mean_channel_noise)
lonely_peaks_idx = [is_lonely_peak_reduced(n) for n in range(peaks.size)]
lonely_peaks_idx = np.array(lonely_peaks_idx, np.bool_)
return lonely_peaks_idx
def find_nearest_idx(array,value):
idx = np.searchsorted(array, value, side="left")
if idx > 0 and (idx == len(array) or math.fabs(value - array[idx-1]) < math.fabs(value - array[idx])):
return idx-1
else:
return idx
def estimate_peak_err(time, HXR_y, diff, peak_loc, channel_props):
fall_time = channel_props.fall_time
mean_noise_level = channel_props.mean_channel_noise
peak_fall_times = fall_time * np.log(HXR_y[peak_loc]/mean_noise_level)
# get maximum of diff in [peak-peak_fall_time ; peak+peak_fall_time]
left_index = (find_nearest_idx(time, time[p]-peak_fall_time) for p,peak_fall_time in zip(peak_loc,peak_fall_times))
right_index = (find_nearest_idx(time, time[p]+peak_fall_time) for p, peak_fall_time in zip(peak_loc,peak_fall_times))
# how to vectorize?
peak_error_estimate = np.array([np.max(np.abs(diff[l:r])) for l,r in zip(left_index,right_index)])
return peak_error_estimate
In [14]:
for ch in channels:
if ch.noFit or ch.rise_time is None or ch.rise_time is None:
continue
ch.simple_corr_peak_height, ch.y_gen, ch.y_diff ,_ = gen_peaks(ch.time, ch.y,
ch.simple_peak_loc,
ch)
ch.simple_corr_peak_err = estimate_peak_err(ch.time, ch.y, ch.y_diff, ch.simple_peak_loc, ch)
ch.lonely_peak_mask = identify_lonely_peaks(ch.time, ch.y,
ch.simple_peak_loc,
ch)
ch.lonely_peak_loc = ch.simple_peak_loc[ch.lonely_peak_mask].copy()
ch.lonely_peak_height = ch.no_corr_peak_height[ch.lonely_peak_mask].copy()
relative_error = np.zeros_like(ch.simple_corr_peak_height)
np.divide(ch.simple_corr_peak_err, ch.simple_corr_peak_height, out=relative_error, where=ch.simple_corr_peak_height != 0)
filtering_mask = relative_error < 0.1
ch.simple_corr_peak_filtered_loc = ch.simple_peak_loc[filtering_mask]
ch.simple_corr_peak_filtered_height = ch.simple_corr_peak_height[filtering_mask]
In [15]:
# plots = list()
# for ch in channels:
# if ch.noFit or ch.rise_time is None or ch.rise_time is None or ch.noPlot:
# continue
# l_gen = '%s gen' % ch.name
# l_dif = '%s diff' % ch.name
# data_to_plot = {ch.name : ch.y, l_gen : ch.y_gen, l_dif : ch.y_diff}
# df = pd.DataFrame(data_to_plot,index = pd.Index(ch.time*1e3, name = 'time'))
# ddf = dd.from_pandas(df, npartitions=os.cpu_count())
# gen_pl = ddf[l_gen].hvplot.line(rasterize = True, cmap=['green'], colorbar=False)
# org_pl = ddf[ch.name].hvplot.line(rasterize = True, cmap=['blue'], colorbar=False)
# diff_pl = ddf[l_dif].hvplot.line(rasterize = True, cmap=['red'], colorbar=False)
# peaks_pl = hv.ErrorBars((ch.time[ch.simple_peak_loc]*1e3,ch.y[ch.simple_peak_loc],ch.simple_corr_peak_height,0),
# vdims=['U_%s' % ch.name, 'yerrneg', 'yerrpos'],label = 'peaks').opts(upper_head=None, lower_head=None)
# plot = org_pl * gen_pl * diff_pl * peaks_pl
# plot.opts(width=950, height = 200).relabel(ch.name)
# plots.append(plot)
# hv.Layout(plots).cols(1)
In [16]:
fig,(ax,ax2,ax3) = plt.subplots(3, sharex = True, sharey=True)
for ch, color in zip(channels, color_cycle):
ax.hist(ch.no_corr_peak_height, bins = 100, label = '%s' % ch.name, histtype = 'step', color = color)
if ch.noFit or ch.rise_time is None or ch.fall_time is None:
continue
ax2.hist(ch.simple_corr_peak_height[ch.lonely_peak_mask], bins = 100, histtype = 'step', color = color)
ax3.hist(ch.simple_corr_peak_filtered_height, bins = 100, histtype = 'step', color = color)
ax.grid()
ax2.grid()
ax3.grid()
ax.semilogy()
ax2.semilogy()
ax3.semilogy()
fig.legend();
Use differance to find less prominent peaks¶
In [17]:
for ch in channels:
if ch.noFit or ch.rise_time is None or ch.rise_time is None:
continue
peaks_loc = ch.simple_peak_loc
for pass_no in range(3):
peaks_locNew, _ = find_peaks(ch.y_diff, height=ch.threshold, prominence = ch.mean_channel_noise*2)
if peaks_locNew.size == 0:
continue
_, _, ch.y_diff,_ = gen_peaks(ch.time,
ch.y_diff,
peaks_locNew,
ch,
test_diff = False)
peaks_loc = np.concatenate((peaks_loc,peaks_locNew))
peaks_loc = np.sort(peaks_loc)
ch.multipass_peak_loc = peaks_loc
Re-generate waveform¶
In [18]:
plots = list()
for ch in channels:
if ch.noFit or ch.rise_time is None or ch.rise_time is None:
continue
peaks_loc = ch.multipass_peak_loc
corrected_peak_heights, ch.y_gen2, ch.y_diff, discarded = gen_peaks(ch.time,
ch.y,
peaks_loc,
ch,
test_diff = True)
### Discard false positives, peaks below threshold and saturated values
saturated = ch.y[peaks_loc]>0.9
mask_discarded = np.logical_or(discarded, corrected_peak_heights<ch.threshold, saturated)
corrected_peak_heights = corrected_peak_heights[~mask_discarded]
peaks_loc = peaks_loc[~mask_discarded]
ch.multipass_peak_loc = peaks_loc
ch.multipass_corr_peak_height = corrected_peak_heights
ch.multipass_corr_peak_err = estimate_peak_err(ch.time, ch.y, ch.y_diff, ch.multipass_peak_loc, ch)
relative_error = np.zeros_like(ch.multipass_corr_peak_height)
np.divide(ch.multipass_corr_peak_err, ch.multipass_corr_peak_height, out=relative_error, where=ch.multipass_corr_peak_height != 0)
filtering_mask = relative_error < 0.1
ch.multipass_corr_peak_filtered_loc = ch.multipass_peak_loc[filtering_mask]
ch.multipass_corr_peak_filtered_height = ch.multipass_corr_peak_height[filtering_mask]
In [19]:
# plots = list()
# for ch in channels:
# if ch.noFit or ch.rise_time is None or ch.rise_time is None or ch.noPlot:
# continue
# l_gen = '%s gen' % ch.name
# l_gen2 = '%s gen2' % ch.name
# l_dif = '%s diff' % ch.name
# data_to_plot = {ch.name : ch.y, l_gen : ch.y_gen, l_dif : ch.y_diff, l_gen2 : ch.y_gen2}
# df = pd.DataFrame(data_to_plot, index = pd.Index(ch.time*1e3, name = 'time'))
# ddf = dd.from_pandas(df, npartitions=os.cpu_count())
# gen_pl = ddf[l_gen].hvplot.line(rasterize = True, cmap=['olive'], colorbar=False)
# gen_pl2 = ddf[l_gen2].hvplot.line(rasterize = True, cmap=['green'], colorbar=False)
# org_pl = ddf[ch.name].hvplot.line(rasterize = True, cmap=['blue'], colorbar=False)
# diff_pl = ddf[l_dif].hvplot.line(rasterize = True, cmap=['red'], colorbar=False)
# peaks_pl = hv.ErrorBars((ch.time[ch.multipass_peak_loc]*1e3,ch.y[ch.multipass_peak_loc],ch.multipass_corr_peak_height,0),
# vdims=['U_%s' % ch.name, 'yerrneg', 'yerrpos']).opts(upper_head=None, lower_head=None)
# plot = org_pl * gen_pl * gen_pl2 * diff_pl * peaks_pl
# plot.opts(width=950, height = 200).relabel(ch.name)
# plots.append(plot)
# plot = hv.Layout(plots).cols(1)
# plot
Show HXR energy spectrum¶
In [20]:
fig, ax = plt.subplots()
for ch,color in zip(channels,color_cycle):
if ch.noPlot or ch.calibration is None:
continue
_, bins, _ = ax.hist(ch.no_corr_peak_height*ch.calibration, bins = 100, label = ch.name,
range = (0,max_HXR_energy), histtype = 'step', ls = '--', color = color)
if ch.noFit or ch.rise_time is None or ch.rise_time is None:
continue
h1, b1 = np.histogram(ch.simple_corr_peak_filtered_height*ch.calibration, bins = len(bins)-1, range = (bins[0], bins[-1]))
h2, b2 = np.histogram(ch.multipass_corr_peak_filtered_height*ch.calibration, bins = len(bins)-1, range = (bins[0], bins[-1]))
h = (h1+h2)/2
h_p = np.maximum(h1,h2)-h
h_m = h-np.minimum(h1,h2)
ax.errorbar(bins[:-1], h, yerr = (h_m, h_p), color = color, ls = '')
fig.legend(frameon =True, fancybox=True,framealpha=1.)
ax.semilogy()
ax.grid()
ax.set_xlabel('$E$ [keV]')
ax.set_ylabel('N [-]')
Out[20]:
Text(0, 0.5, 'N [-]')
In [21]:
fig,(ax,ax2,ax3,ax4) = plt.subplots(4, sharex = True, sharey=True)
for ch, color in zip(channels, color_cycle):
if ch.noFit or ch.rise_time is None or ch.fall_time is None:
continue
ax.hist(ch.simple_corr_peak_height*ch.calibration, bins = 100, label = '%s' % ch.name, histtype = 'step', color = color)
ax2.hist(ch.simple_corr_peak_filtered_height*ch.calibration, bins = 100, histtype = 'step', color = color)
ax3.hist(ch.multipass_corr_peak_height*ch.calibration, bins = 100, histtype = 'step', color = color)
ax4.hist(ch.multipass_corr_peak_filtered_height*ch.calibration, bins = 100, histtype = 'step', color = color)
ax.grid()
ax2.grid()
ax3.grid()
ax4.grid()
ax.semilogy()
ax2.semilogy()
ax3.semilogy()
ax4.semilogy()
fig.legend();
In [22]:
ip_url = f'http://golem.fjfi.cvut.cz/shots/{shot_no}/Diagnostics/BasicDiagnostics/Results/Ip.csv'
ip = ds.open(ip_url)
ip = np.loadtxt(ip,delimiter=',')
In [23]:
plot_channels = []
for ch,color in zip(channels,color_cycle):
if ch.noPlot or ch.calibration is None or ch.noFit:
continue
plot_channels.append(ch)
fig, axes = plt.subplots(nrows = len(plot_channels), ncols =2, figsize = [1.5*6.4, 1.75*4.8])
prevAx = None
prevrAx = None
for ch,color,ax, rax in zip(plot_channels,color_cycle,axes[:,0],axes[:,1]):
if ch.noPlot or ch.calibration is None:
continue
_, bins, _ = ax.hist(ch.no_corr_peak_height*ch.calibration, bins = 100, label = ch.name,
range = (0,max_HXR_energy), histtype = 'step', ls = '--', color = color)
if ch.noFit or ch.rise_time is None or ch.rise_time is None:
continue
h1, b1 = np.histogram(ch.simple_corr_peak_height*ch.calibration,
bins = len(bins)-1, range = (bins[0], bins[-1]))
h2, b2 = np.histogram(ch.multipass_corr_peak_height*ch.calibration,
bins = len(bins)-1, range = (bins[0], bins[-1]))
h = (h1+h2)/2
h_p = np.maximum(h1,h2)-h
h_m = h-np.minimum(h1,h2)
ax.errorbar(bins[:-1], h, yerr = (h_m, h_p),ds = 'steps', color = color)
h3, b3 = np.histogram(ch.simple_corr_peak_height[ch.lonely_peak_mask]*ch.calibration,
bins = len(bins)-1, range = (bins[0], bins[-1]))
ax.step(bins[:-1], h3, color = color, ls = '--', alpha = .75)
ip_l = rax.plot(ip[:,0],ip[:,1],'k--', label = '$I_p$')
rax.grid(True,axis='x')
rax = rax.twinx()
peak_time = ch.time[ch.simple_peak_loc] * 1e3
peak_height = ch.simple_corr_peak_height*ch.calibration
rax.scatter(peak_time,peak_height, color = color,s = .1 )
rax.grid(True)
#rax.yaxis.tick_right()
if not ch.flip_waveform:
max_reliableE = ch.calibration * .9
if np.any(ch.y[ch.simple_peak_loc]>max_reliableE):
rax.axhline(max_reliableE/2,color = 'r',ls = '-')
ax.semilogy()
ax.grid()
ax.set_ylabel('N [-]')
if prevAx is not None:
ax.sharex(prevAx)
prevAx = ax
rax.yaxis.set_label_position("right")
rax.set_ylabel('$E$ [keV]')
if prevrAx is not None:
#rax.sharex(prevrAx)
rax.sharey(prevrAx)
prevrAx = rax
newHandles = [ip_l[0]]
for ax in axes[:,0]:
handles, labels = ax.get_legend_handles_labels()
newHandles += handles
fig.legend(handles= newHandles,frameon =True, fancybox=True,framealpha=1.,loc = 'lower center')
ax.set_xlabel('$E$ [keV]')
rax.set_xlabel('t [ms]');
rax.set_ylim(-100,max_HXR_energy)
fig.suptitle(f"#{shot_no}");
plt.savefig('icon-fig.png')
