attoworld.data
This module will contain functions for loading the various data formats used across the labs.
1""" 2This module will contain functions for loading the various data formats used across the labs. 3""" 4 5from .file_io import ( 6 read_dwc, 7 load_mean_spectrum_from_scarab, 8 load_waves_from_matfile, 9 load_waveform_from_text, 10 load_spectrum_from_text, 11 read_Trebino_FROG_matrix, 12 read_Trebino_FROG_speck, 13 read_Trebino_FROG_data, 14) 15from .LunaResult import LunaResult 16from .data_structures import ( 17 yaml_io, 18 SpectrometerCalibration, 19 Waveform, 20 ComplexSpectrum, 21 IntensitySpectrum, 22 ComplexEnvelope, 23 FrogData, 24 Spectrogram, 25 CalibrationDataset, 26 CalibrationInput, 27) 28 29__all__ = [ 30 "read_dwc", 31 "load_mean_spectrum_from_scarab", 32 "load_waves_from_matfile", 33 "load_waveform_from_text", 34 "load_spectrum_from_text", 35 "FrogData", 36 "read_Trebino_FROG_matrix", 37 "read_Trebino_FROG_speck", 38 "read_Trebino_FROG_data", 39 "LunaResult", 40 "yaml_io", 41 "SpectrometerCalibration", 42 "Waveform", 43 "ComplexSpectrum", 44 "IntensitySpectrum", 45 "ComplexEnvelope", 46 "Spectrogram", 47 "CalibrationDataset", 48 "CalibrationInput", 49]
13def read_dwc(file_path): 14 """ 15 Reads files in the .dwc format produced by many FROG scanners 16 17 Args: 18 file_path: path to the .dwc file 19 20 Returns: 21 Spectrogram: the loaded data 22 """ 23 with open(file_path, "r") as file: 24 lines = file.readlines() 25 delay_increment = float(lines[2].strip().split("=")[1]) 26 27 wavelength_vector = pd.read_csv( 28 file_path, skiprows=5, nrows=1, delimiter="\t", header=None 29 ).values[0] 30 31 data_array = pd.read_csv( 32 file_path, skiprows=8, delimiter="\t", header=None, dtype=float 33 ).values 34 35 freqs, first_spec = spectrum.wavelength_to_frequency( 36 wavelength_vector, data_array[:, 0] 37 ) 38 if freqs is not None: 39 data_array_freq = np.zeros((len(freqs), data_array.shape[1])) 40 data_array_freq[:, 0] = first_spec 41 for _i in range(data_array.shape[1]): 42 _f, _dat = spectrum.wavelength_to_frequency( 43 wavelength_vector, data_array[:, _i], freqs 44 ) 45 data_array_freq[:, _i] = _dat 46 dt = 1e-15 * delay_increment 47 delays = dt * np.array(range(data_array.shape[1])) 48 delays -= np.mean(delays) 49 return data_structures.Spectrogram( 50 data=data_array_freq, time=delays, freq=freqs 51 ) 52 else: 53 raise Exception("Interpolation failure reading dwc file")
Reads files in the .dwc format produced by many FROG scanners
Arguments:
- file_path: path to the .dwc file
Returns:
Spectrogram: the loaded data
56def load_mean_spectrum_from_scarab(filename: str): 57 data = np.loadtxt(filename) 58 return data_structures.IntensitySpectrum( 59 spectrum=np.mean(data[:, 1::], axis=1), 60 wavelength=1e-9 * data[:, 0], 61 freq=1e9 * constants.speed_of_light / data[:, 0], 62 is_frequency_scaled=False, 63 phase=None, 64 )
128def load_waves_from_matfile(filename: str, phase: Optional[float] = None): 129 """Load the contents of an attolab scanner file in .mat format 130 131 Args: 132 phase (float): phase to use when interpreting the lock-in data 133 filename (str): path to the mat file 134 Returns: 135 time_delay: array of time delay values 136 signal: signals corresponding to the time delays 137 """ 138 139 datablob = sio.loadmat(filename) 140 stage_position = datablob["xdata"][0, :] 141 time_delay = -2e-3 * stage_position / 2.9979e8 142 lia_x = datablob["x0"] 143 lia_y = datablob["y0"] 144 if phase is None: 145 optimized_phase = np.atan2(np.sum(lia_y[:] ** 2), np.sum(lia_x[:] ** 2)) 146 signal = np.fliplr( 147 lia_x * np.cos(optimized_phase) + lia_y * np.sin(optimized_phase) 148 ) 149 return time_delay, signal 150 else: 151 signal = np.fliplr(lia_x * np.cos(phase) - lia_y * np.sin(phase)) 152 return time_delay, signal
Load the contents of an attolab scanner file in .mat format
Arguments:
- phase (float): phase to use when interpreting the lock-in data
- filename (str): path to the mat file
Returns:
time_delay: array of time delay values signal: signals corresponding to the time delays
97def load_waveform_from_text( 98 filename: str, 99 time_multiplier: float = 1e-15, 100 time_field: str = "delay (fs)", 101 wave_field: str = "field (a.u.)", 102 sep="\t", 103) -> data_structures.Waveform: 104 """Loads a waveform from a text file 105 106 Args: 107 filename (str): path to the file 108 time_multiplier (float): multiplier needed to convert the time unit in the file to seconds 109 time_field (str): name (header) of the column containing the times 110 wave_field (str): name (header) of the column containing the waveform 111 sep (str): separator used in the file format (tab is default) 112 113 Returns: 114 Waveform: the waveform 115 """ 116 117 data = pd.read_csv(filename, sep=sep) 118 time = time_multiplier * data[time_field].to_numpy() 119 wave = data[wave_field].to_numpy() 120 dt = time[1] - time[0] 121 diff_time = np.diff(time) 122 uniform = bool(np.all(np.isclose(diff_time, diff_time[0]))) 123 return data_structures.Waveform( 124 wave=wave, time=time, dt=dt, is_uniformly_spaced=uniform 125 )
Loads a waveform from a text file
Arguments:
- filename (str): path to the file
- time_multiplier (float): multiplier needed to convert the time unit in the file to seconds
- time_field (str): name (header) of the column containing the times
- wave_field (str): name (header) of the column containing the waveform
- sep (str): separator used in the file format (tab is default)
Returns:
Waveform: the waveform
67def load_spectrum_from_text( 68 filename: str, 69 wavelength_multiplier: float = 1e-9, 70 wavelength_field: str = "wavelength (nm)", 71 spectrum_field: str = "intensity (a.u.)", 72 sep: str = "\t", 73): 74 """ 75 Load a spectrum contained in a text file. 76 77 Args: 78 filename (str): path to the file 79 wavelength_multiplier (float): multiplier to convert wavelength to m 80 wavelength_field (str): name of the field in the data corresponding to wavelength 81 spectrum_field (str): name of the field in the data corresponding to spectral intensity 82 sep (str): column separator 83 Returns: 84 IntensitySpectrum: the intensity spectrum""" 85 data = pd.read_csv(filename, sep=sep) 86 wavelength = wavelength_multiplier * np.array(data[wavelength_field]) 87 freq = constants.speed_of_light / wavelength 88 spectrum = np.array(data[spectrum_field]) 89 return data_structures.IntensitySpectrum( 90 spectrum=spectrum, 91 wavelength=wavelength, 92 freq=freq, 93 phase=np.zeros(spectrum.shape, dtype=float), 94 )
Load a spectrum contained in a text file.
Arguments:
- filename (str): path to the file
- wavelength_multiplier (float): multiplier to convert wavelength to m
- wavelength_field (str): name of the field in the data corresponding to wavelength
- spectrum_field (str): name of the field in the data corresponding to spectral intensity
- sep (str): column separator
Returns:
IntensitySpectrum: the intensity spectrum
1115@yaml_io 1116@dataclass(slots=True) 1117class FrogData: 1118 """ 1119 Stores data from a FROG measurement 1120 1121 Attributes: 1122 spectrum (ComplexSpectrum): the reconstructed complex spectrum 1123 pulse (Waveform): time-domain reconstructed field 1124 measured_spectrogram (Spectrogram): measured (binned) data 1125 reconstructed_spectrogram (Spectrogram): spectrogram resulting from reconstructed field 1126 f0: the central frequency of the spectrum 1127 dt: the time step 1128 """ 1129 1130 spectrum: ComplexSpectrum 1131 pulse: Waveform 1132 measured_spectrogram: Spectrogram 1133 reconstructed_spectrogram: Spectrogram 1134 raw_reconstruction: np.ndarray 1135 f0: float 1136 dt: float 1137 1138 def lock(self): 1139 """ 1140 Make the data immutable 1141 """ 1142 self.raw_reconstruction.setflags(write=False) 1143 self.spectrum.lock() 1144 self.pulse.lock() 1145 self.measured_spectrogram.lock() 1146 self.reconstructed_spectrogram.lock() 1147 1148 def save(self, base_filename): 1149 """ 1150 Save in the Trebino FROG format 1151 1152 Args: 1153 base_filename: base of the file path; 4 files will be made from it: .A.dat, .Arecon.dat, .Ek.dat, and .Speck.dat 1154 """ 1155 self.measured_spectrogram.save(base_filename + ".A.dat") 1156 self.reconstructed_spectrogram.save(base_filename + ".Arecon.dat") 1157 1158 t = 1e15 * self.dt * np.array(range(len(self.raw_reconstruction))) 1159 t -= np.mean(t) 1160 f = np.fft.fftshift(np.fft.fftfreq(len(t), d=self.dt)) + self.f0 1161 lam = constants.speed_of_light / f 1162 raw_spec = np.fft.fftshift(np.fft.fft(self.raw_reconstruction)) 1163 1164 with open(base_filename + ".Ek.dat", "w") as time_file: 1165 for _i in range(len(self.raw_reconstruction)): 1166 time_file.write( 1167 f"{t[_i]:.15g}\t{np.abs(self.raw_reconstruction[_i]) ** 2:.15g}\t{np.angle(self.raw_reconstruction[_i]):.15g}\t{np.real(self.raw_reconstruction[_i]):.15g}\t{np.imag(self.raw_reconstruction[_i]):.15g}\n" 1168 ) 1169 1170 with open(base_filename + ".Speck.dat", "w") as spec_file: 1171 for _i in range(len(self.raw_reconstruction)): 1172 spec_file.write( 1173 f"{lam[_i]:.15g}\t{np.abs(raw_spec[_i]) ** 2:.15g}\t{np.angle(raw_spec[_i]):.15g}\t{np.real(raw_spec[_i]):.15g}\t{np.imag(raw_spec[_i]):.15g}\n" 1174 ) 1175 1176 def plot_measured_spectrogram(self, ax: Optional[Axes] = None, log: bool = False): 1177 """ 1178 Plot the measured spectrogram. 1179 1180 Args: 1181 ax: optionally plot onto a pre-existing matplotlib Axes 1182 """ 1183 if ax is None: 1184 fig, ax = plt.subplots() 1185 else: 1186 fig = ax.get_figure() 1187 if log: 1188 self.measured_spectrogram.plot_log(ax) 1189 else: 1190 self.measured_spectrogram.plot(ax) 1191 ax.set_title("Measured") 1192 return fig 1193 1194 def plot_reconstructed_spectrogram( 1195 self, ax: Optional[Axes] = None, log: bool = False 1196 ): 1197 """ 1198 Plot the reconstructed spectrogram. 1199 1200 Args: 1201 ax: optionally plot onto a pre-existing matplotlib Axes 1202 """ 1203 if ax is None: 1204 fig, ax = plt.subplots() 1205 else: 1206 fig = ax.get_figure() 1207 if log: 1208 self.reconstructed_spectrogram.plot_log(ax) 1209 else: 1210 self.reconstructed_spectrogram.plot(ax) 1211 ax.set_title( 1212 f"Retrieved (G': {self.get_error():0.2e}; G: {self.get_G_error():0.2e})" 1213 ) 1214 return fig 1215 1216 def plot_pulse( 1217 self, ax: Optional[Axes] = None, phase_blanking: float = 0.05, xlim=None 1218 ): 1219 """ 1220 Plot the reconstructed pulse. 1221 1222 Args: 1223 ax: optionally plot onto a pre-existing matplotlib Axes 1224 phase_blanking: only show phase information (instantaneous frequency) above this level relative to max intensity 1225 xlim: pass arguments to set_xlim() to constrain the x-axis 1226 """ 1227 return self.pulse.to_complex_envelope().plot(ax, phase_blanking, xlim) 1228 1229 def plot_spectrum( 1230 self, ax: Optional[Axes] = None, phase_blanking: float = 0.05, xlim=None 1231 ): 1232 """ 1233 Plot the reconstructed spectrum and group delay curve. 1234 1235 Args: 1236 ax: optionally plot onto a pre-existing matplotlib Axes 1237 phase_blanking: only show phase information (group delay) above this level relative to max intensity 1238 xlim: pass arguments to set_xlim() to constrain the x-axis 1239 """ 1240 return self.spectrum.to_intensity_spectrum().plot_with_group_delay( 1241 ax, phase_blanking=phase_blanking, shift_from_centered=True, xlim=xlim 1242 ) 1243 1244 def plot_all( 1245 self, 1246 phase_blanking=0.05, 1247 time_xlims=None, 1248 wavelength_xlims=None, 1249 figsize=None, 1250 log: bool = False, 1251 ): 1252 """ 1253 Produce a 4-panel plot of the FROG results, combining calls to plot_measured_spectrogram(), 1254 plot_reconstructed_spectrogram(), plot_pulse() and plot_spectrum() as subplots, with letter labels. 1255 1256 Args: 1257 phase_blanking: relative intensity at which to show phase information 1258 time_xlim: x-axis limits to pass to the plot of the pulse 1259 wavelength_xlim: x-axis limits to pass to the plot of the spectrum 1260 figsize: custom figure size 1261 """ 1262 if figsize is None: 1263 default_figsize = plt.rcParams["figure.figsize"] 1264 figsize = (default_figsize[0] * 2, default_figsize[1] * 2) 1265 fig, ax = plt.subplots(2, 2, figsize=figsize) 1266 self.plot_measured_spectrogram(ax[0, 0], log=log) 1267 label_letter("a", ax[0, 0]) 1268 self.plot_reconstructed_spectrogram(ax[1, 0], log=log) 1269 label_letter("b", ax[1, 0]) 1270 self.plot_pulse(ax[0, 1], xlim=time_xlims, phase_blanking=phase_blanking) 1271 label_letter("c", ax[0, 1]) 1272 self.plot_spectrum( 1273 ax[1, 1], xlim=wavelength_xlims, phase_blanking=phase_blanking 1274 ) 1275 label_letter("d", ax[1, 1]) 1276 return fig 1277 1278 def get_error(self) -> float: 1279 """ 1280 Get the G' error of the reconstruction 1281 """ 1282 norm_measured = np.linalg.norm(self.measured_spectrogram.data) 1283 norm_retrieved = np.linalg.norm(self.reconstructed_spectrogram.data) 1284 return np.sqrt( 1285 np.sum( 1286 ( 1287 self.measured_spectrogram.data[:] / norm_measured 1288 - self.reconstructed_spectrogram.data[:] / norm_retrieved 1289 ) 1290 ** 2 1291 ) 1292 / np.sum((self.measured_spectrogram.data[:] / norm_measured) ** 2) 1293 ) 1294 1295 def get_G_error(self) -> float: 1296 """ 1297 Get the G (note: no apostrophe) error. This one doesn't mean much, but is useful 1298 for comparing reconstructions of the same spectrogram between different programs. 1299 """ 1300 return np.sqrt( 1301 (1.0 / float(len(self.measured_spectrogram.data[:]) ** 2)) 1302 * np.sum( 1303 ( 1304 self.measured_spectrogram.data[:] 1305 / np.max(self.measured_spectrogram.data[:]) 1306 - self.reconstructed_spectrogram.data[:] 1307 / np.max(self.reconstructed_spectrogram.data[:]) 1308 ) 1309 ** 2 1310 ) 1311 ) 1312 1313 def get_fwhm(self) -> float: 1314 """ 1315 Get the full-width-at-half-max value of the reconstructed pulse 1316 """ 1317 return self.pulse.get_envelope_fwhm()
Stores data from a FROG measurement
Attributes:
- spectrum (ComplexSpectrum): the reconstructed complex spectrum
- pulse (Waveform): time-domain reconstructed field
- measured_spectrogram (Spectrogram): measured (binned) data
- reconstructed_spectrogram (Spectrogram): spectrogram resulting from reconstructed field
- f0: the central frequency of the spectrum
- dt: the time step
1138 def lock(self): 1139 """ 1140 Make the data immutable 1141 """ 1142 self.raw_reconstruction.setflags(write=False) 1143 self.spectrum.lock() 1144 self.pulse.lock() 1145 self.measured_spectrogram.lock() 1146 self.reconstructed_spectrogram.lock()
Make the data immutable
1148 def save(self, base_filename): 1149 """ 1150 Save in the Trebino FROG format 1151 1152 Args: 1153 base_filename: base of the file path; 4 files will be made from it: .A.dat, .Arecon.dat, .Ek.dat, and .Speck.dat 1154 """ 1155 self.measured_spectrogram.save(base_filename + ".A.dat") 1156 self.reconstructed_spectrogram.save(base_filename + ".Arecon.dat") 1157 1158 t = 1e15 * self.dt * np.array(range(len(self.raw_reconstruction))) 1159 t -= np.mean(t) 1160 f = np.fft.fftshift(np.fft.fftfreq(len(t), d=self.dt)) + self.f0 1161 lam = constants.speed_of_light / f 1162 raw_spec = np.fft.fftshift(np.fft.fft(self.raw_reconstruction)) 1163 1164 with open(base_filename + ".Ek.dat", "w") as time_file: 1165 for _i in range(len(self.raw_reconstruction)): 1166 time_file.write( 1167 f"{t[_i]:.15g}\t{np.abs(self.raw_reconstruction[_i]) ** 2:.15g}\t{np.angle(self.raw_reconstruction[_i]):.15g}\t{np.real(self.raw_reconstruction[_i]):.15g}\t{np.imag(self.raw_reconstruction[_i]):.15g}\n" 1168 ) 1169 1170 with open(base_filename + ".Speck.dat", "w") as spec_file: 1171 for _i in range(len(self.raw_reconstruction)): 1172 spec_file.write( 1173 f"{lam[_i]:.15g}\t{np.abs(raw_spec[_i]) ** 2:.15g}\t{np.angle(raw_spec[_i]):.15g}\t{np.real(raw_spec[_i]):.15g}\t{np.imag(raw_spec[_i]):.15g}\n" 1174 )
Save in the Trebino FROG format
Arguments:
- base_filename: base of the file path; 4 files will be made from it: .A.dat, .Arecon.dat, .Ek.dat, and .Speck.dat
1176 def plot_measured_spectrogram(self, ax: Optional[Axes] = None, log: bool = False): 1177 """ 1178 Plot the measured spectrogram. 1179 1180 Args: 1181 ax: optionally plot onto a pre-existing matplotlib Axes 1182 """ 1183 if ax is None: 1184 fig, ax = plt.subplots() 1185 else: 1186 fig = ax.get_figure() 1187 if log: 1188 self.measured_spectrogram.plot_log(ax) 1189 else: 1190 self.measured_spectrogram.plot(ax) 1191 ax.set_title("Measured") 1192 return fig
Plot the measured spectrogram.
Arguments:
- ax: optionally plot onto a pre-existing matplotlib Axes
1194 def plot_reconstructed_spectrogram( 1195 self, ax: Optional[Axes] = None, log: bool = False 1196 ): 1197 """ 1198 Plot the reconstructed spectrogram. 1199 1200 Args: 1201 ax: optionally plot onto a pre-existing matplotlib Axes 1202 """ 1203 if ax is None: 1204 fig, ax = plt.subplots() 1205 else: 1206 fig = ax.get_figure() 1207 if log: 1208 self.reconstructed_spectrogram.plot_log(ax) 1209 else: 1210 self.reconstructed_spectrogram.plot(ax) 1211 ax.set_title( 1212 f"Retrieved (G': {self.get_error():0.2e}; G: {self.get_G_error():0.2e})" 1213 ) 1214 return fig
Plot the reconstructed spectrogram.
Arguments:
- ax: optionally plot onto a pre-existing matplotlib Axes
1216 def plot_pulse( 1217 self, ax: Optional[Axes] = None, phase_blanking: float = 0.05, xlim=None 1218 ): 1219 """ 1220 Plot the reconstructed pulse. 1221 1222 Args: 1223 ax: optionally plot onto a pre-existing matplotlib Axes 1224 phase_blanking: only show phase information (instantaneous frequency) above this level relative to max intensity 1225 xlim: pass arguments to set_xlim() to constrain the x-axis 1226 """ 1227 return self.pulse.to_complex_envelope().plot(ax, phase_blanking, xlim)
Plot the reconstructed pulse.
Arguments:
- ax: optionally plot onto a pre-existing matplotlib Axes
- phase_blanking: only show phase information (instantaneous frequency) above this level relative to max intensity
- xlim: pass arguments to set_xlim() to constrain the x-axis
1229 def plot_spectrum( 1230 self, ax: Optional[Axes] = None, phase_blanking: float = 0.05, xlim=None 1231 ): 1232 """ 1233 Plot the reconstructed spectrum and group delay curve. 1234 1235 Args: 1236 ax: optionally plot onto a pre-existing matplotlib Axes 1237 phase_blanking: only show phase information (group delay) above this level relative to max intensity 1238 xlim: pass arguments to set_xlim() to constrain the x-axis 1239 """ 1240 return self.spectrum.to_intensity_spectrum().plot_with_group_delay( 1241 ax, phase_blanking=phase_blanking, shift_from_centered=True, xlim=xlim 1242 )
Plot the reconstructed spectrum and group delay curve.
Arguments:
- ax: optionally plot onto a pre-existing matplotlib Axes
- phase_blanking: only show phase information (group delay) above this level relative to max intensity
- xlim: pass arguments to set_xlim() to constrain the x-axis
1244 def plot_all( 1245 self, 1246 phase_blanking=0.05, 1247 time_xlims=None, 1248 wavelength_xlims=None, 1249 figsize=None, 1250 log: bool = False, 1251 ): 1252 """ 1253 Produce a 4-panel plot of the FROG results, combining calls to plot_measured_spectrogram(), 1254 plot_reconstructed_spectrogram(), plot_pulse() and plot_spectrum() as subplots, with letter labels. 1255 1256 Args: 1257 phase_blanking: relative intensity at which to show phase information 1258 time_xlim: x-axis limits to pass to the plot of the pulse 1259 wavelength_xlim: x-axis limits to pass to the plot of the spectrum 1260 figsize: custom figure size 1261 """ 1262 if figsize is None: 1263 default_figsize = plt.rcParams["figure.figsize"] 1264 figsize = (default_figsize[0] * 2, default_figsize[1] * 2) 1265 fig, ax = plt.subplots(2, 2, figsize=figsize) 1266 self.plot_measured_spectrogram(ax[0, 0], log=log) 1267 label_letter("a", ax[0, 0]) 1268 self.plot_reconstructed_spectrogram(ax[1, 0], log=log) 1269 label_letter("b", ax[1, 0]) 1270 self.plot_pulse(ax[0, 1], xlim=time_xlims, phase_blanking=phase_blanking) 1271 label_letter("c", ax[0, 1]) 1272 self.plot_spectrum( 1273 ax[1, 1], xlim=wavelength_xlims, phase_blanking=phase_blanking 1274 ) 1275 label_letter("d", ax[1, 1]) 1276 return fig
Produce a 4-panel plot of the FROG results, combining calls to plot_measured_spectrogram(), plot_reconstructed_spectrogram(), plot_pulse() and plot_spectrum() as subplots, with letter labels.
Arguments:
- phase_blanking: relative intensity at which to show phase information
- time_xlim: x-axis limits to pass to the plot of the pulse
- wavelength_xlim: x-axis limits to pass to the plot of the spectrum
- figsize: custom figure size
1278 def get_error(self) -> float: 1279 """ 1280 Get the G' error of the reconstruction 1281 """ 1282 norm_measured = np.linalg.norm(self.measured_spectrogram.data) 1283 norm_retrieved = np.linalg.norm(self.reconstructed_spectrogram.data) 1284 return np.sqrt( 1285 np.sum( 1286 ( 1287 self.measured_spectrogram.data[:] / norm_measured 1288 - self.reconstructed_spectrogram.data[:] / norm_retrieved 1289 ) 1290 ** 2 1291 ) 1292 / np.sum((self.measured_spectrogram.data[:] / norm_measured) ** 2) 1293 )
Get the G' error of the reconstruction
1295 def get_G_error(self) -> float: 1296 """ 1297 Get the G (note: no apostrophe) error. This one doesn't mean much, but is useful 1298 for comparing reconstructions of the same spectrogram between different programs. 1299 """ 1300 return np.sqrt( 1301 (1.0 / float(len(self.measured_spectrogram.data[:]) ** 2)) 1302 * np.sum( 1303 ( 1304 self.measured_spectrogram.data[:] 1305 / np.max(self.measured_spectrogram.data[:]) 1306 - self.reconstructed_spectrogram.data[:] 1307 / np.max(self.reconstructed_spectrogram.data[:]) 1308 ) 1309 ** 2 1310 ) 1311 )
Get the G (note: no apostrophe) error. This one doesn't mean much, but is useful for comparing reconstructions of the same spectrogram between different programs.
1313 def get_fwhm(self) -> float: 1314 """ 1315 Get the full-width-at-half-max value of the reconstructed pulse 1316 """ 1317 return self.pulse.get_envelope_fwhm()
Get the full-width-at-half-max value of the reconstructed pulse
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file
155def read_Trebino_FROG_matrix(filename: Path | str) -> data_structures.Spectrogram: 156 """ 157 Read a spectrogram file made by the Trebino FROG code 158 159 Args: 160 filename (Path | str): the name (path) of the file 161 """ 162 with open(filename, "r") as f: 163 line = str(f.readline()) 164 line = line.split() 165 n1 = int(line[0]) 166 n2 = int(line[1]) 167 line = str(f.readline()) 168 line = line.split() 169 measured_data = pd.read_csv(filename, sep="\t", header=None, skiprows=2) 170 measure = [] 171 raw_freq = ( 172 1e9 * constants.speed_of_light / np.array(measured_data[0][0:n2]).squeeze() 173 ) 174 df = np.mean(np.diff(raw_freq)) 175 freq = raw_freq[0] + df * np.array(range(raw_freq.shape[0])) 176 time = 1e-15 * np.array(measured_data[0][n2 : (n2 + n1)]).squeeze() 177 for i in range(n1): 178 measure.append(measured_data[0][(i + 2) * n2 : (i + 3) * n2]) 179 data = np.array(measure) 180 return data_structures.Spectrogram(data=data, time=time, freq=freq)
Read a spectrogram file made by the Trebino FROG code
Arguments:
- filename (Path | str): the name (path) of the file
183def read_Trebino_FROG_speck(filename: Path | str) -> data_structures.ComplexSpectrum: 184 """ 185 Read a .Speck file made by the Trebino FROG code 186 187 Args: 188 filename (Path | str): the name (path) of the file 189 """ 190 data = np.array(pd.read_csv(filename, sep="\t", header=None), dtype=float) 191 raw_freq = 1e9 * constants.speed_of_light / data[:, 0] 192 df = np.mean(np.diff(raw_freq)) 193 freq = np.linspace(0.0, raw_freq[-1], int(np.ceil(raw_freq[-1] / df))) 194 spectrum = interpolate(freq, raw_freq, data[:, 3]) + 1j * interpolate( 195 freq, raw_freq, data[:, 4] 196 ) 197 return data_structures.ComplexSpectrum(spectrum=spectrum, freq=freq)
Read a .Speck file made by the Trebino FROG code
Arguments:
- filename (Path | str): the name (path) of the file
200def read_Trebino_FROG_data(filename: str) -> data_structures.FrogData: 201 """ 202 Read a set of data produced by the Trebino FROG reconstruction code. 203 204 Args: 205 filename: Base filename of the .bin file; e.g. if the data are mydata.bin.Speck.dat etc., this will be "mydata.bin" """ 206 spectrum = read_Trebino_FROG_speck(filename + ".Speck.dat") 207 pulse = spectrum.to_centered_waveform() 208 measured_spectrogram = read_Trebino_FROG_matrix(filename + ".A.dat") 209 reconstructed_spectrogram = read_Trebino_FROG_matrix(filename + ".Arecon.dat") 210 raw_speck = np.array( 211 pd.read_csv(filename + ".Speck.dat", sep="\t", header=None), dtype=float 212 ) 213 raw_ek = np.array( 214 pd.read_csv(filename + ".Ek.dat", sep="\t", header=None), dtype=float 215 ) 216 f0 = 1e9 * np.mean(constants.speed_of_light / raw_speck[:, 0]) 217 dt = 1e-15 * (raw_ek[1, 0] - raw_ek[0, 0]) 218 raw_reconstruction = raw_ek[:, 3] + 1.0j * raw_ek[:, 4] 219 return data_structures.FrogData( 220 spectrum=spectrum, 221 pulse=pulse, 222 measured_spectrogram=measured_spectrogram, 223 reconstructed_spectrogram=reconstructed_spectrogram, 224 raw_reconstruction=raw_reconstruction, 225 dt=dt, 226 f0=float(f0), 227 )
Read a set of data produced by the Trebino FROG reconstruction code.
Arguments:
- filename: Base filename of the .bin file; e.g. if the data are mydata.bin.Speck.dat etc., this will be "mydata.bin"
39class LunaResult: 40 """Loads and handles the Luna simulation result. 41 42 The result must be in the HDF5 format using the saving option in the Luna.Interface.prop_capillary() function [filepath="..."]. 43 As opposed to most of the analysis routines here, units are in SI! 44 45 Attributes: 46 z: position along the fiber (for the field data) 47 fieldFT: complex FFT of the field (positive freq axis); shape: (n_z, n_modes, n_freq) [or (n_z, n_freq) after mode selection/averaging] 48 omega: angular frequency axis for the FFT field; shape: (n_z, n_modes, n_freq) [or (n_z, n_freq) after mode selection/averaging] 49 50 stats_z: position along the fiber (for stats data) 51 stats_density: particle density along the fiber (m^-3) 52 stats_electrondensity: electron density along the fiber (m^-3) 53 stats_pressure: gas pressure profile along the fiber (bar) 54 stats_energy: pulse energy along the fiber (J) 55 stats_peakpower: peak power of the pulse along the fiber (W) 56 stats_peakintensity: peak intensity of the pulse along the fiber 57 stats_peak_ionization_rate: peak ionization rate along the fiber 58 """ 59 60 def __init__(self, filename): 61 """Constructor of class LunaResult. 62 63 Args: 64 filename: saved result, file path 65 """ 66 67 self.filename = filename 68 self.fieldFT = None 69 self.omega = None 70 self.z = None 71 72 self.stats_z = None 73 self.stats_energy = None 74 self.stats_electrondensity = None 75 self.stats_density = None 76 self.stats_pressure = None 77 self.stats_peakpower = None 78 self.stats_peakintensity = None 79 self.stats_zdw = None 80 self.stats_peak_ionization_rate = None 81 82 self.open_Luna_result(filename) 83 84 def open_Luna_result(self, filename): 85 """Opens the Luna result file and loads the data""" 86 with h5py.File(filename, "r") as data: 87 # FIELD DATA 88 self.fieldFT = np.array(data["Eω"]) 89 self.omega = np.array(data["grid"]["ω"]) 90 self.z = np.array(data["z"]) 91 92 # STATS 93 self.stats_z = np.array(data["stats"]["z"]) 94 self.stats_energy = np.array(data["stats"]["energy"]) 95 self.stats_electrondensity = np.array(data["stats"]["electrondensity"]) 96 self.stats_density = np.array(data["stats"]["density"]) 97 self.stats_pressure = np.array(data["stats"]["pressure"]) 98 self.stats_peakpower = np.array(data["stats"]["peakpower"]) 99 self.stats_peakintensity = np.array(data["stats"]["peakintensity"]) 100 self.stats_zdw = np.array(data["stats"]["zdw"]) 101 self.stats_peak_ionization_rate = np.array( 102 data["stats"]["peak_ionisation_rate"] 103 ) 104 105 def average_modes(self): 106 """Averages the propagation modes in the Luna result file""" 107 if len(self.fieldFT.shape) == 3: 108 self.fieldFT = np.mean(self.fieldFT, axis=1) 109 self.stats_zdw = None 110 self.stats_peakpower = None 111 self.stats_energy = np.sum(self.stats_energy, axis=1) 112 113 def select_mode(self, mode: int): 114 if len(self.fieldFT.shape) < 3: 115 print("WARNING: No mode to select") 116 elif mode >= self.fieldFT.shape[1] or mode < 0: 117 print("WARNING: mode ", mode, " is out of range") 118 else: 119 self.fieldFT = self.fieldFT[:, mode, :] 120 self.stats_zdw = self.stats_zdw[:, mode] 121 self.stats_peakpower = self.stats_peakpower[:, mode] 122 self.stats_energy = self.stats_energy[:, mode] 123 124 def get_time_field(self, position=None): 125 """Get the electric field in time from the Luna result file. If no mode was previously selected, the method computes the average of all modes. 126 Therefore, after calling get_time_field(), mode selection is not possible any more. 127 128 Args: 129 position (float): position along the fiber in m. If None, the end of the fiber is used. 130 131 Returns: 132 timeV (numpy.ndarray): time axis in seconds 133 fieldV (numpy.ndarray): electric field in V/m 134 """ 135 self.average_modes() 136 if position is None: 137 position = self.z[-1] 138 index = np.argmin(np.abs(self.z - position)) 139 if position > np.max(self.z) or position < np.min(self.z): 140 print("WARNING: position ", position, "m is out of range") 141 check_equal_length(self.fieldFT[index], self.omega) 142 fieldFFT = np.concatenate( 143 (self.fieldFT[index, :], np.conjugate(self.fieldFT[index, :][::-1]) * 0) 144 ) 145 freq = np.concatenate((self.omega, -self.omega[::-1])) / 2 / np.pi 146 timeV, fieldV = inverse_fourier_transform(freq, fieldFFT) 147 return timeV, np.real(fieldV) 148 149 def get_wavelength_spectrum(self, position=None): 150 """Get the spectrum from the Luna result file (|FFT|^2 * (2 * pi * c / λ^2)) 151 152 Args: 153 position (float): position along the fiber in m. If None, the end of the fiber is used. 154 155 Returns: 156 wvl (numpy.ndarray): wavelength axis in m 157 wvlSpectrum (numpy.ndarray): electric field spectrum in V/m 158 """ 159 self.average_modes() 160 if position is None: 161 position = self.z[-1] 162 index = np.argmin(np.abs(self.z - position)) 163 if position > np.max(self.z) or position < np.min(self.z): 164 print("WARNING: position ", position, "m is out of range") 165 wvl = 2 * np.pi * constants.speed_of_light / self.omega[::-1] 166 wvlSpectrum = np.abs(self.fieldFT[index, ::-1]) ** 2 * ( 167 2 * np.pi * constants.speed_of_light / wvl**2 168 ) 169 return wvl, wvlSpectrum 170 171 def get_spectral_phase(self, position=None): 172 """Get the spectral phase from the Luna result file 173 174 Args: 175 position (float): position along the fiber in m. If None, the end of the fiber is used. 176 177 Returns: 178 wvl (numpy.ndarray): wavelength axis in m 179 phase (numpy.ndarray): spectral phase in rad 180 """ 181 self.average_modes() 182 if position is None: 183 position = self.z[-1] 184 index = np.argmin(np.abs(self.z - position)) 185 if position > np.max(self.z) or position < np.min(self.z): 186 print("WARNING: position ", position, "m is out of range") 187 wvl = 2 * np.pi * constants.speed_of_light / self.omega[::-1] 188 phase = np.angle(self.fieldFT[index, ::-1]) 189 return wvl, phase 190 191 def plot_stats(self): 192 """Plots the 'stats_' attribute of the simulation stored in the present object.""" 193 194 fig, axs = plt.subplots(2, 2, figsize=[7, 5]) 195 axs[0, 0].plot(self.stats_z, self.stats_pressure) 196 axs[0, 0].set_xlabel("z (m)") 197 axs[0, 0].set_ylabel("gas pressure (bar)") 198 ax2 = axs[0, 0].twinx() 199 ax2.plot(self.stats_z, self.stats_density, color="r") 200 ax2.set_ylabel("gas particle density ($m^{-3}$)", color="r") 201 ax2.tick_params(axis="y", colors="r") 202 ax2.yaxis.label.set_color("r") 203 axs[0, 1].plot(self.stats_z, self.stats_electrondensity) 204 axs[0, 1].set_xlabel("z (m)") 205 axs[0, 1].set_ylabel("electron density ($m^{-3}$)") 206 ax2 = axs[0, 1].twinx() 207 if len(self.stats_energy.shape) == 2: 208 ax2.plot(self.stats_z, np.sum(self.stats_energy, axis=1) * 1e6, color="r") 209 else: 210 ax2.plot(self.stats_z, self.stats_energy * 1e6, color="r") 211 ax2.set_ylabel(f"pulse energy ({Char.mu}J)", color="r") 212 ax2.tick_params(axis="y", colors="r") 213 ax2.yaxis.label.set_color("r") 214 if self.stats_peakpower is not None: 215 if len(self.stats_peakpower.shape) == 2: 216 axs[1, 0].plot(self.stats_z, self.stats_peakpower[:, 0]) 217 axs[1, 0].set_xlabel("z (m)") 218 axs[1, 0].set_ylabel("peak power (W)") 219 elif len(self.stats_peakpower.shape) == 1: 220 axs[1, 0].plot(self.stats_z, self.stats_peakpower) 221 axs[1, 0].set_xlabel("z (m)") 222 axs[1, 0].set_ylabel("peak power (W)") 223 axs[1, 1].plot(self.stats_z, self.stats_peak_ionization_rate) 224 axs[1, 1].set_xlabel("z (m)") 225 axs[1, 1].set_ylabel("peak ionization rate") 226 plt.tight_layout() 227 228 return fig
Loads and handles the Luna simulation result.
The result must be in the HDF5 format using the saving option in the Luna.Interface.prop_capillary() function [filepath="..."]. As opposed to most of the analysis routines here, units are in SI!
Attributes:
- z: position along the fiber (for the field data)
- fieldFT: complex FFT of the field (positive freq axis); shape: (n_z, n_modes, n_freq) [or (n_z, n_freq) after mode selection/averaging]
- omega: angular frequency axis for the FFT field; shape: (n_z, n_modes, n_freq) [or (n_z, n_freq) after mode selection/averaging]
- stats_z: position along the fiber (for stats data)
- stats_density: particle density along the fiber (m^-3)
- stats_electrondensity: electron density along the fiber (m^-3)
- stats_pressure: gas pressure profile along the fiber (bar)
- stats_energy: pulse energy along the fiber (J)
- stats_peakpower: peak power of the pulse along the fiber (W)
- stats_peakintensity: peak intensity of the pulse along the fiber
- stats_peak_ionization_rate: peak ionization rate along the fiber
60 def __init__(self, filename): 61 """Constructor of class LunaResult. 62 63 Args: 64 filename: saved result, file path 65 """ 66 67 self.filename = filename 68 self.fieldFT = None 69 self.omega = None 70 self.z = None 71 72 self.stats_z = None 73 self.stats_energy = None 74 self.stats_electrondensity = None 75 self.stats_density = None 76 self.stats_pressure = None 77 self.stats_peakpower = None 78 self.stats_peakintensity = None 79 self.stats_zdw = None 80 self.stats_peak_ionization_rate = None 81 82 self.open_Luna_result(filename)
Constructor of class LunaResult.
Arguments:
- filename: saved result, file path
84 def open_Luna_result(self, filename): 85 """Opens the Luna result file and loads the data""" 86 with h5py.File(filename, "r") as data: 87 # FIELD DATA 88 self.fieldFT = np.array(data["Eω"]) 89 self.omega = np.array(data["grid"]["ω"]) 90 self.z = np.array(data["z"]) 91 92 # STATS 93 self.stats_z = np.array(data["stats"]["z"]) 94 self.stats_energy = np.array(data["stats"]["energy"]) 95 self.stats_electrondensity = np.array(data["stats"]["electrondensity"]) 96 self.stats_density = np.array(data["stats"]["density"]) 97 self.stats_pressure = np.array(data["stats"]["pressure"]) 98 self.stats_peakpower = np.array(data["stats"]["peakpower"]) 99 self.stats_peakintensity = np.array(data["stats"]["peakintensity"]) 100 self.stats_zdw = np.array(data["stats"]["zdw"]) 101 self.stats_peak_ionization_rate = np.array( 102 data["stats"]["peak_ionisation_rate"] 103 )
Opens the Luna result file and loads the data
105 def average_modes(self): 106 """Averages the propagation modes in the Luna result file""" 107 if len(self.fieldFT.shape) == 3: 108 self.fieldFT = np.mean(self.fieldFT, axis=1) 109 self.stats_zdw = None 110 self.stats_peakpower = None 111 self.stats_energy = np.sum(self.stats_energy, axis=1)
Averages the propagation modes in the Luna result file
113 def select_mode(self, mode: int): 114 if len(self.fieldFT.shape) < 3: 115 print("WARNING: No mode to select") 116 elif mode >= self.fieldFT.shape[1] or mode < 0: 117 print("WARNING: mode ", mode, " is out of range") 118 else: 119 self.fieldFT = self.fieldFT[:, mode, :] 120 self.stats_zdw = self.stats_zdw[:, mode] 121 self.stats_peakpower = self.stats_peakpower[:, mode] 122 self.stats_energy = self.stats_energy[:, mode]
124 def get_time_field(self, position=None): 125 """Get the electric field in time from the Luna result file. If no mode was previously selected, the method computes the average of all modes. 126 Therefore, after calling get_time_field(), mode selection is not possible any more. 127 128 Args: 129 position (float): position along the fiber in m. If None, the end of the fiber is used. 130 131 Returns: 132 timeV (numpy.ndarray): time axis in seconds 133 fieldV (numpy.ndarray): electric field in V/m 134 """ 135 self.average_modes() 136 if position is None: 137 position = self.z[-1] 138 index = np.argmin(np.abs(self.z - position)) 139 if position > np.max(self.z) or position < np.min(self.z): 140 print("WARNING: position ", position, "m is out of range") 141 check_equal_length(self.fieldFT[index], self.omega) 142 fieldFFT = np.concatenate( 143 (self.fieldFT[index, :], np.conjugate(self.fieldFT[index, :][::-1]) * 0) 144 ) 145 freq = np.concatenate((self.omega, -self.omega[::-1])) / 2 / np.pi 146 timeV, fieldV = inverse_fourier_transform(freq, fieldFFT) 147 return timeV, np.real(fieldV)
Get the electric field in time from the Luna result file. If no mode was previously selected, the method computes the average of all modes. Therefore, after calling get_time_field(), mode selection is not possible any more.
Arguments:
- position (float): position along the fiber in m. If None, the end of the fiber is used.
Returns:
timeV (numpy.ndarray): time axis in seconds fieldV (numpy.ndarray): electric field in V/m
149 def get_wavelength_spectrum(self, position=None): 150 """Get the spectrum from the Luna result file (|FFT|^2 * (2 * pi * c / λ^2)) 151 152 Args: 153 position (float): position along the fiber in m. If None, the end of the fiber is used. 154 155 Returns: 156 wvl (numpy.ndarray): wavelength axis in m 157 wvlSpectrum (numpy.ndarray): electric field spectrum in V/m 158 """ 159 self.average_modes() 160 if position is None: 161 position = self.z[-1] 162 index = np.argmin(np.abs(self.z - position)) 163 if position > np.max(self.z) or position < np.min(self.z): 164 print("WARNING: position ", position, "m is out of range") 165 wvl = 2 * np.pi * constants.speed_of_light / self.omega[::-1] 166 wvlSpectrum = np.abs(self.fieldFT[index, ::-1]) ** 2 * ( 167 2 * np.pi * constants.speed_of_light / wvl**2 168 ) 169 return wvl, wvlSpectrum
Get the spectrum from the Luna result file (|FFT|^2 * (2 * pi * c / λ^2))
Arguments:
- position (float): position along the fiber in m. If None, the end of the fiber is used.
Returns:
wvl (numpy.ndarray): wavelength axis in m wvlSpectrum (numpy.ndarray): electric field spectrum in V/m
171 def get_spectral_phase(self, position=None): 172 """Get the spectral phase from the Luna result file 173 174 Args: 175 position (float): position along the fiber in m. If None, the end of the fiber is used. 176 177 Returns: 178 wvl (numpy.ndarray): wavelength axis in m 179 phase (numpy.ndarray): spectral phase in rad 180 """ 181 self.average_modes() 182 if position is None: 183 position = self.z[-1] 184 index = np.argmin(np.abs(self.z - position)) 185 if position > np.max(self.z) or position < np.min(self.z): 186 print("WARNING: position ", position, "m is out of range") 187 wvl = 2 * np.pi * constants.speed_of_light / self.omega[::-1] 188 phase = np.angle(self.fieldFT[index, ::-1]) 189 return wvl, phase
Get the spectral phase from the Luna result file
Arguments:
- position (float): position along the fiber in m. If None, the end of the fiber is used.
Returns:
wvl (numpy.ndarray): wavelength axis in m phase (numpy.ndarray): spectral phase in rad
191 def plot_stats(self): 192 """Plots the 'stats_' attribute of the simulation stored in the present object.""" 193 194 fig, axs = plt.subplots(2, 2, figsize=[7, 5]) 195 axs[0, 0].plot(self.stats_z, self.stats_pressure) 196 axs[0, 0].set_xlabel("z (m)") 197 axs[0, 0].set_ylabel("gas pressure (bar)") 198 ax2 = axs[0, 0].twinx() 199 ax2.plot(self.stats_z, self.stats_density, color="r") 200 ax2.set_ylabel("gas particle density ($m^{-3}$)", color="r") 201 ax2.tick_params(axis="y", colors="r") 202 ax2.yaxis.label.set_color("r") 203 axs[0, 1].plot(self.stats_z, self.stats_electrondensity) 204 axs[0, 1].set_xlabel("z (m)") 205 axs[0, 1].set_ylabel("electron density ($m^{-3}$)") 206 ax2 = axs[0, 1].twinx() 207 if len(self.stats_energy.shape) == 2: 208 ax2.plot(self.stats_z, np.sum(self.stats_energy, axis=1) * 1e6, color="r") 209 else: 210 ax2.plot(self.stats_z, self.stats_energy * 1e6, color="r") 211 ax2.set_ylabel(f"pulse energy ({Char.mu}J)", color="r") 212 ax2.tick_params(axis="y", colors="r") 213 ax2.yaxis.label.set_color("r") 214 if self.stats_peakpower is not None: 215 if len(self.stats_peakpower.shape) == 2: 216 axs[1, 0].plot(self.stats_z, self.stats_peakpower[:, 0]) 217 axs[1, 0].set_xlabel("z (m)") 218 axs[1, 0].set_ylabel("peak power (W)") 219 elif len(self.stats_peakpower.shape) == 1: 220 axs[1, 0].plot(self.stats_z, self.stats_peakpower) 221 axs[1, 0].set_xlabel("z (m)") 222 axs[1, 0].set_ylabel("peak power (W)") 223 axs[1, 1].plot(self.stats_z, self.stats_peak_ionization_rate) 224 axs[1, 1].set_xlabel("z (m)") 225 axs[1, 1].set_ylabel("peak ionization rate") 226 plt.tight_layout() 227 228 return fig
Plots the 'stats_' attribute of the simulation stored in the present object.
23def yaml_io(cls): 24 """ 25 Adds functions to save and load the dataclass as yaml 26 """ 27 28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data) 57 58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data) 68 69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file) 79 80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict 101 102 cls.from_dict = classmethod(from_dict) 103 cls.load_yaml = classmethod(load_yaml) 104 cls.to_dict = to_dict 105 cls.save_yaml = save_yaml 106 return cls
Adds functions to save and load the dataclass as yaml
109@yaml_io 110@dataclass(slots=True) 111class SpectrometerCalibration: 112 """ 113 Set of data describing a spectrometer calibration 114 115 Attributes: 116 intensity_factors (np.ndarray): the intensity correction factors (weights) 117 corrected_wavelengths (np.ndarray): the corrected wavelength array of the spectrometer 118 corrected_frequencies (np.ndarray): the corrected frequency array of the spectrometer (c/wavelengths) 119 """ 120 121 intensity_factors: np.ndarray 122 original_wavelengths: np.ndarray 123 corrected_wavelengths: np.ndarray 124 corrected_frequencies: np.ndarray 125 126 def apply_to_spectrum(self, spectrum_in): 127 """ 128 Applies itself to an intensity spectrum: 129 130 Args: 131 spectrum_in (IntensitySpectrum): the spectrum to be calibrated 132 133 Returns: 134 IntensitySpectrum: the calibrated spectrum 135 """ 136 intensity_out = spectrum_in.spectrum * self.intensity_factors 137 return IntensitySpectrum( 138 spectrum=intensity_out, 139 phase=spectrum_in.phase, 140 wavelength=self.corrected_wavelengths, 141 freq=self.corrected_frequencies, 142 ) 143 144 def apply_to_spectrogram(self, spectrogram_in): 145 """ 146 Applies itself to an intensity spectrum: 147 148 Args: 149 spectrum_in (Spectrogram): the spectrogram to be calibrated 150 151 Returns: 152 Spectrogram: the calibrated spectrogram 153 """ 154 original_freqs = constants.speed_of_light/self.original_wavelengths 155 original_freq_projection = interpolate(spectrogram_in.freq, original_freqs, self.corrected_frequencies, inputs_are_sorted=False) 156 data_out = np.zeros(spectrogram_in.data.shape,dtype=float) 157 intensity_factors = interpolate(spectrogram_in.freq, self.corrected_frequencies, self.intensity_factors, inputs_are_sorted=False) 158 for i in range(spectrogram_in.data.shape[1]): 159 data_out[:,i] = intensity_factors * interpolate(spectrogram_in.freq, original_freq_projection, np.array(spectrogram_in.data[:,i])) 160 161 return Spectrogram( 162 data=data_out, freq=spectrogram_in.freq, time=spectrogram_in.time 163 ) 164 165 def save_npz(self, filepath): 166 """ 167 Save to an npz file 168 169 Args: 170 filepath: path to save to 171 """ 172 np.savez( 173 filepath, 174 intensity_factors=self.intensity_factors, 175 original_wavelengths=self.original_wavelengths, 176 corrected_wavelengths=self.corrected_wavelengths, 177 corrected_frequencies=self.corrected_frequencies, 178 ) 179 180 @staticmethod 181 def from_npz(filepath): 182 """ 183 Make an instance of the class from an npz file 184 185 Args: 186 filepath: path to the file 187 188 Returns: 189 SpectrometerCalibration: the calibration in the file""" 190 npzfile = np.load(filepath) 191 return SpectrometerCalibration( 192 intensity_factors=npzfile["intensity_factors"], 193 original_wavelengths=npzfile["original_wavelengths"], 194 corrected_wavelengths=npzfile["corrected_wavelengths"], 195 corrected_frequencies=npzfile["corrected_frequencies"], 196 ) 197 198 @staticmethod 199 def from_named(spectrometer: spectrum.CalibrationData): 200 """ 201 Loads a calibration saved in the database 202 203 Args: 204 spectrometer (spectrum.CalibrationData): Value from the CalibrationData enum attoworld.spectrum.CalibrationData 205 206 Returns: 207 SpectrometerCalibration: the calibration associated with the enum value 208 """ 209 return SpectrometerCalibration.from_npz( 210 spectrum.get_calibration_path() / spectrometer.value 211 )
Set of data describing a spectrometer calibration
Attributes:
- intensity_factors (np.ndarray): the intensity correction factors (weights)
- corrected_wavelengths (np.ndarray): the corrected wavelength array of the spectrometer
- corrected_frequencies (np.ndarray): the corrected frequency array of the spectrometer (c/wavelengths)
126 def apply_to_spectrum(self, spectrum_in): 127 """ 128 Applies itself to an intensity spectrum: 129 130 Args: 131 spectrum_in (IntensitySpectrum): the spectrum to be calibrated 132 133 Returns: 134 IntensitySpectrum: the calibrated spectrum 135 """ 136 intensity_out = spectrum_in.spectrum * self.intensity_factors 137 return IntensitySpectrum( 138 spectrum=intensity_out, 139 phase=spectrum_in.phase, 140 wavelength=self.corrected_wavelengths, 141 freq=self.corrected_frequencies, 142 )
Applies itself to an intensity spectrum:
Arguments:
- spectrum_in (IntensitySpectrum): the spectrum to be calibrated
Returns:
IntensitySpectrum: the calibrated spectrum
144 def apply_to_spectrogram(self, spectrogram_in): 145 """ 146 Applies itself to an intensity spectrum: 147 148 Args: 149 spectrum_in (Spectrogram): the spectrogram to be calibrated 150 151 Returns: 152 Spectrogram: the calibrated spectrogram 153 """ 154 original_freqs = constants.speed_of_light/self.original_wavelengths 155 original_freq_projection = interpolate(spectrogram_in.freq, original_freqs, self.corrected_frequencies, inputs_are_sorted=False) 156 data_out = np.zeros(spectrogram_in.data.shape,dtype=float) 157 intensity_factors = interpolate(spectrogram_in.freq, self.corrected_frequencies, self.intensity_factors, inputs_are_sorted=False) 158 for i in range(spectrogram_in.data.shape[1]): 159 data_out[:,i] = intensity_factors * interpolate(spectrogram_in.freq, original_freq_projection, np.array(spectrogram_in.data[:,i])) 160 161 return Spectrogram( 162 data=data_out, freq=spectrogram_in.freq, time=spectrogram_in.time 163 )
Applies itself to an intensity spectrum:
Arguments:
- spectrum_in (Spectrogram): the spectrogram to be calibrated
Returns:
Spectrogram: the calibrated spectrogram
165 def save_npz(self, filepath): 166 """ 167 Save to an npz file 168 169 Args: 170 filepath: path to save to 171 """ 172 np.savez( 173 filepath, 174 intensity_factors=self.intensity_factors, 175 original_wavelengths=self.original_wavelengths, 176 corrected_wavelengths=self.corrected_wavelengths, 177 corrected_frequencies=self.corrected_frequencies, 178 )
Save to an npz file
Arguments:
- filepath: path to save to
180 @staticmethod 181 def from_npz(filepath): 182 """ 183 Make an instance of the class from an npz file 184 185 Args: 186 filepath: path to the file 187 188 Returns: 189 SpectrometerCalibration: the calibration in the file""" 190 npzfile = np.load(filepath) 191 return SpectrometerCalibration( 192 intensity_factors=npzfile["intensity_factors"], 193 original_wavelengths=npzfile["original_wavelengths"], 194 corrected_wavelengths=npzfile["corrected_wavelengths"], 195 corrected_frequencies=npzfile["corrected_frequencies"], 196 )
Make an instance of the class from an npz file
Arguments:
- filepath: path to the file
Returns:
SpectrometerCalibration: the calibration in the file
198 @staticmethod 199 def from_named(spectrometer: spectrum.CalibrationData): 200 """ 201 Loads a calibration saved in the database 202 203 Args: 204 spectrometer (spectrum.CalibrationData): Value from the CalibrationData enum attoworld.spectrum.CalibrationData 205 206 Returns: 207 SpectrometerCalibration: the calibration associated with the enum value 208 """ 209 return SpectrometerCalibration.from_npz( 210 spectrum.get_calibration_path() / spectrometer.value 211 )
Loads a calibration saved in the database
Arguments:
- spectrometer (spectrum.CalibrationData): Value from the CalibrationData enum attoworld.spectrum.CalibrationData
Returns:
SpectrometerCalibration: the calibration associated with the enum value
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file
467@yaml_io 468@dataclass(slots=True) 469class Waveform: 470 """ 471 Contains data describing an electric field waveform. 472 In SI units. 473 474 Attributes: 475 wave (np.ndarray): the waveform data 476 time (np.ndarray): time vector corresponding to the wave array 477 dt (float): the time step of time, if it is uniformly spaced 478 is_uniformly_spaced (bool): says whether or not the time steps are uniform 479 """ 480 481 wave: np.ndarray 482 time: np.ndarray 483 dt: float 484 is_uniformly_spaced: bool = False 485 486 def lock(self): 487 """ 488 Make the data immutable 489 """ 490 self.wave.setflags(write=False) 491 self.time.setflags(write=False) 492 493 def copy(self): 494 """ 495 Returns a deep copy 496 """ 497 return copy.deepcopy(self) 498 499 def time_fs(self): 500 if self.time is not None: 501 return 1e15 * self.time 502 else: 503 raise Exception("No data") 504 505 def to_uniformly_spaced(self): 506 """ 507 Returns a version of the struct with uniformly spaced time 508 """ 509 if self.is_uniformly_spaced: 510 return self 511 else: 512 timesteps = np.abs(np.diff(self.time)) 513 new_dt = np.min(timesteps[timesteps > 0.0]) 514 new_time_length = self.time[-1] - self.time[0] 515 new_time = self.time[0] + new_dt * np.array( 516 range(int(new_time_length / new_dt)) 517 ) 518 return Waveform( 519 wave=interpolate(new_time, self.time, self.wave), 520 time=new_time, 521 dt=new_dt, 522 is_uniformly_spaced=True, 523 ) 524 525 def to_windowed(self, window_desc): 526 """ 527 Create a windowed version of the waveform. Output will be uniformly spaced in time, even if current state isn't. 528 529 Args: 530 window_desc: String or tuple describing the desired window in the same format as scipy.signal.windows.get_window. 531 532 Examples: 533 >>> waveform_with_tukey_window = waveform.to_windowed('tukey') 534 >>> waveform_with_supergaussian_window = waveform.to_windowed(('general_gaussian', 2, 500)) 535 """ 536 uniform_self = self.to_uniformly_spaced() 537 new_wave = uniform_self.wave * sig.windows.get_window( 538 window_desc, Nx=uniform_self.wave.shape[0] 539 ) 540 return Waveform( 541 wave=new_wave, 542 time=uniform_self.time, 543 dt=uniform_self.dt, 544 is_uniformly_spaced=True, 545 ) 546 547 def to_bandpassed(self, frequency: float, sigma: float, order: int = 4): 548 r""" 549 Apply a bandpass filter, with the same spec as to_bandpassed method of ComplexSpectrum: 550 $e^{\frac{(f-f_0)^r}{2\sigma^r}}$ 551 where $f_0$ is the frequency argument, $\sigma$ is the sigma argument, and r is the order argument 552 553 Args: 554 frequency: the central frequency (Hz) of the bandpass 555 sigma: the width of the bandpass (Hz) 556 order: the order of the supergaussian 557 """ 558 return ( 559 self.to_complex_spectrum() 560 .to_bandpassed(frequency, sigma, order) 561 .to_waveform() 562 ) 563 564 def to_complex_spectrum(self, padding_factor: int = 1): 565 """ 566 Converts to a ComplexSpectrum class 567 568 Args: 569 padding_factor (int): factor by which to expand the temporal length in the FFT, giving a smoother spectrum 570 """ 571 if self.is_uniformly_spaced: 572 new_spectrum = np.fft.rfft( 573 self.wave, n=self.wave.shape[0] * padding_factor, axis=0 574 ) 575 new_freq = np.fft.rfftfreq(self.wave.shape[0] * padding_factor, d=self.dt) 576 else: 577 uniform_self = self.to_uniformly_spaced() 578 new_spectrum = np.fft.rfft( 579 uniform_self.wave, n=uniform_self.wave.shape[0] * padding_factor, axis=0 580 ) 581 new_freq = np.fft.rfftfreq( 582 uniform_self.wave.shape[0] * padding_factor, d=uniform_self.dt 583 ) 584 585 return ComplexSpectrum(spectrum=new_spectrum, freq=new_freq) 586 587 def to_intensity_spectrum( 588 self, wavelength_scaled: bool = True, padding_factor: int = 1 589 ): 590 """ 591 Converts to an intensity spectrum 592 593 Args: 594 wavelength_scaled (bool): Correct the spectral intensities for plotting on a wavelength scale 595 """ 596 return self.to_complex_spectrum(padding_factor).to_intensity_spectrum( 597 wavelength_scaled 598 ) 599 600 def to_time_derivative(self): 601 """ 602 Return the time-derivative of the waveform 603 """ 604 return self.to_complex_spectrum().to_time_derivative().to_waveform() 605 606 def to_normalized(self): 607 """ 608 Return a normalized version of the waveform 609 """ 610 max_loc, max_val = find_maximum_location( 611 np.abs(np.array(sig.hilbert(self.wave))) 612 ) 613 return Waveform( 614 wave=self.wave / max_val, 615 time=np.array(self.time), 616 dt=self.dt, 617 is_uniformly_spaced=self.is_uniformly_spaced, 618 ) 619 620 def to_complex_envelope(self, f0: float = 0.0): 621 """ 622 Return a ComplexEnvelope class corresponding to the waveform 623 624 Args: 625 f0 (float): central frequency to use when constructing the envelope. E.g. oscillation at this frequency will be cancelled. 626 """ 627 uniform_self = self.to_uniformly_spaced() 628 analytic = np.array( 629 sig.hilbert(uniform_self.wave) 630 * np.exp(-1j * 2 * np.pi * f0 * uniform_self.time) 631 ) 632 return ComplexEnvelope( 633 envelope=analytic, 634 time=uniform_self.time, 635 dt=uniform_self.dt, 636 carrier_frequency=f0, 637 ) 638 639 def get_envelope_fwhm(self) -> float: 640 """ 641 Get the full-width-at-half-maximum of the intensity envelope. 642 643 Returns: 644 float: the FWHM 645 """ 646 return self.to_complex_envelope().get_fwhm() 647 648 def get_field_squared_fwhm(self): 649 """ 650 Get the full-width-at-half-maximum of the square of the field 651 652 Returns: 653 float: the FWHM 654 """ 655 uniform_self = self.to_uniformly_spaced() 656 return fwhm(uniform_self.wave**2, uniform_self.dt)
Contains data describing an electric field waveform. In SI units.
Attributes:
- wave (np.ndarray): the waveform data
- time (np.ndarray): time vector corresponding to the wave array
- dt (float): the time step of time, if it is uniformly spaced
- is_uniformly_spaced (bool): says whether or not the time steps are uniform
486 def lock(self): 487 """ 488 Make the data immutable 489 """ 490 self.wave.setflags(write=False) 491 self.time.setflags(write=False)
Make the data immutable
505 def to_uniformly_spaced(self): 506 """ 507 Returns a version of the struct with uniformly spaced time 508 """ 509 if self.is_uniformly_spaced: 510 return self 511 else: 512 timesteps = np.abs(np.diff(self.time)) 513 new_dt = np.min(timesteps[timesteps > 0.0]) 514 new_time_length = self.time[-1] - self.time[0] 515 new_time = self.time[0] + new_dt * np.array( 516 range(int(new_time_length / new_dt)) 517 ) 518 return Waveform( 519 wave=interpolate(new_time, self.time, self.wave), 520 time=new_time, 521 dt=new_dt, 522 is_uniformly_spaced=True, 523 )
Returns a version of the struct with uniformly spaced time
525 def to_windowed(self, window_desc): 526 """ 527 Create a windowed version of the waveform. Output will be uniformly spaced in time, even if current state isn't. 528 529 Args: 530 window_desc: String or tuple describing the desired window in the same format as scipy.signal.windows.get_window. 531 532 Examples: 533 >>> waveform_with_tukey_window = waveform.to_windowed('tukey') 534 >>> waveform_with_supergaussian_window = waveform.to_windowed(('general_gaussian', 2, 500)) 535 """ 536 uniform_self = self.to_uniformly_spaced() 537 new_wave = uniform_self.wave * sig.windows.get_window( 538 window_desc, Nx=uniform_self.wave.shape[0] 539 ) 540 return Waveform( 541 wave=new_wave, 542 time=uniform_self.time, 543 dt=uniform_self.dt, 544 is_uniformly_spaced=True, 545 )
Create a windowed version of the waveform. Output will be uniformly spaced in time, even if current state isn't.
Arguments:
- window_desc: String or tuple describing the desired window in the same format as scipy.signal.windows.get_window.
Examples:
>>> waveform_with_tukey_window = waveform.to_windowed('tukey') >>> waveform_with_supergaussian_window = waveform.to_windowed(('general_gaussian', 2, 500))
547 def to_bandpassed(self, frequency: float, sigma: float, order: int = 4): 548 r""" 549 Apply a bandpass filter, with the same spec as to_bandpassed method of ComplexSpectrum: 550 $e^{\frac{(f-f_0)^r}{2\sigma^r}}$ 551 where $f_0$ is the frequency argument, $\sigma$ is the sigma argument, and r is the order argument 552 553 Args: 554 frequency: the central frequency (Hz) of the bandpass 555 sigma: the width of the bandpass (Hz) 556 order: the order of the supergaussian 557 """ 558 return ( 559 self.to_complex_spectrum() 560 .to_bandpassed(frequency, sigma, order) 561 .to_waveform() 562 )
Apply a bandpass filter, with the same spec as to_bandpassed method of ComplexSpectrum: $e^{\frac{(f-f_0)^r}{2\sigma^r}}$ where $f_0$ is the frequency argument, $\sigma$ is the sigma argument, and r is the order argument
Arguments:
- frequency: the central frequency (Hz) of the bandpass
- sigma: the width of the bandpass (Hz)
- order: the order of the supergaussian
564 def to_complex_spectrum(self, padding_factor: int = 1): 565 """ 566 Converts to a ComplexSpectrum class 567 568 Args: 569 padding_factor (int): factor by which to expand the temporal length in the FFT, giving a smoother spectrum 570 """ 571 if self.is_uniformly_spaced: 572 new_spectrum = np.fft.rfft( 573 self.wave, n=self.wave.shape[0] * padding_factor, axis=0 574 ) 575 new_freq = np.fft.rfftfreq(self.wave.shape[0] * padding_factor, d=self.dt) 576 else: 577 uniform_self = self.to_uniformly_spaced() 578 new_spectrum = np.fft.rfft( 579 uniform_self.wave, n=uniform_self.wave.shape[0] * padding_factor, axis=0 580 ) 581 new_freq = np.fft.rfftfreq( 582 uniform_self.wave.shape[0] * padding_factor, d=uniform_self.dt 583 ) 584 585 return ComplexSpectrum(spectrum=new_spectrum, freq=new_freq)
Converts to a ComplexSpectrum class
Arguments:
- padding_factor (int): factor by which to expand the temporal length in the FFT, giving a smoother spectrum
587 def to_intensity_spectrum( 588 self, wavelength_scaled: bool = True, padding_factor: int = 1 589 ): 590 """ 591 Converts to an intensity spectrum 592 593 Args: 594 wavelength_scaled (bool): Correct the spectral intensities for plotting on a wavelength scale 595 """ 596 return self.to_complex_spectrum(padding_factor).to_intensity_spectrum( 597 wavelength_scaled 598 )
Converts to an intensity spectrum
Arguments:
- wavelength_scaled (bool): Correct the spectral intensities for plotting on a wavelength scale
600 def to_time_derivative(self): 601 """ 602 Return the time-derivative of the waveform 603 """ 604 return self.to_complex_spectrum().to_time_derivative().to_waveform()
Return the time-derivative of the waveform
606 def to_normalized(self): 607 """ 608 Return a normalized version of the waveform 609 """ 610 max_loc, max_val = find_maximum_location( 611 np.abs(np.array(sig.hilbert(self.wave))) 612 ) 613 return Waveform( 614 wave=self.wave / max_val, 615 time=np.array(self.time), 616 dt=self.dt, 617 is_uniformly_spaced=self.is_uniformly_spaced, 618 )
Return a normalized version of the waveform
620 def to_complex_envelope(self, f0: float = 0.0): 621 """ 622 Return a ComplexEnvelope class corresponding to the waveform 623 624 Args: 625 f0 (float): central frequency to use when constructing the envelope. E.g. oscillation at this frequency will be cancelled. 626 """ 627 uniform_self = self.to_uniformly_spaced() 628 analytic = np.array( 629 sig.hilbert(uniform_self.wave) 630 * np.exp(-1j * 2 * np.pi * f0 * uniform_self.time) 631 ) 632 return ComplexEnvelope( 633 envelope=analytic, 634 time=uniform_self.time, 635 dt=uniform_self.dt, 636 carrier_frequency=f0, 637 )
Return a ComplexEnvelope class corresponding to the waveform
Arguments:
- f0 (float): central frequency to use when constructing the envelope. E.g. oscillation at this frequency will be cancelled.
639 def get_envelope_fwhm(self) -> float: 640 """ 641 Get the full-width-at-half-maximum of the intensity envelope. 642 643 Returns: 644 float: the FWHM 645 """ 646 return self.to_complex_envelope().get_fwhm()
Get the full-width-at-half-maximum of the intensity envelope.
Returns:
float: the FWHM
648 def get_field_squared_fwhm(self): 649 """ 650 Get the full-width-at-half-maximum of the square of the field 651 652 Returns: 653 float: the FWHM 654 """ 655 uniform_self = self.to_uniformly_spaced() 656 return fwhm(uniform_self.wave**2, uniform_self.dt)
Get the full-width-at-half-maximum of the square of the field
Returns:
float: the FWHM
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file
659@yaml_io 660@dataclass(slots=True) 661class ComplexSpectrum: 662 """ 663 Contains a complex spectrum, with spectral weights on a frequency scale 664 665 Attributes: 666 spectrum (np.ndarray): the spectrum in complex128 format 667 freq (np.ndarray): 668 """ 669 670 spectrum: np.ndarray 671 freq: np.ndarray 672 673 def lock(self): 674 """ 675 Make the data immutable 676 """ 677 self.spectrum.setflags(write=False) 678 self.freq.setflags(write=False) 679 680 def copy(self): 681 return copy.deepcopy(self) 682 683 def to_time_derivative(self): 684 r"""Return a ComplexSpectrum corresponding to the time derivative (multiply by $i\omega$)""" 685 d_dt = 1j * 2 * np.pi * self.freq * self.spectrum 686 return ComplexSpectrum(spectrum=d_dt, freq=np.array(self.freq)) 687 688 def to_bandpassed(self, frequency: float, sigma: float, order: int = 4): 689 r""" 690 Return the complex spectrum after applying a supergaussian bandpass filter to the spectrum, of the form 691 $e^{\frac{(f-f_0)^r}{2\sigma^r}}$ 692 where $f_0$ is the frequency argument, $\sigma$ is the sigma argument, and r is the order argument 693 694 Args: 695 frequency: the central frequency (Hz) of the bandpass 696 sigma: the width of the bandpass (Hz) 697 order: the order of the supergaussian 698 """ 699 new_spectrum = self.spectrum * np.exp( 700 -((self.freq - frequency) ** order) / (2 * sigma**order) 701 ) 702 return ComplexSpectrum(spectrum=new_spectrum, freq=np.array(self.freq)) 703 704 def to_waveform(self): 705 """ 706 Create a Waveform based on this complex spectrum. 707 """ 708 wave = np.fft.irfft(self.spectrum, axis=0) 709 dt = 0.5 / (self.freq[-1] - self.freq[0]) 710 time = dt * np.array(range(wave.shape[0])) 711 return Waveform(wave=wave, time=time, dt=dt, is_uniformly_spaced=True) 712 713 def to_centered_waveform(self): 714 """ 715 Create a Waveform based on this complex spectrum and center it in the time window 716 """ 717 wave = np.fft.irfft(self.spectrum, axis=0) 718 dt = 0.5 / (self.freq[-1] - self.freq[0]) 719 time = dt * np.array(range(wave.shape[0])) 720 max_ind, max_val = find_maximum_location(np.abs(np.array(sig.hilbert(wave)))) 721 max_loc = dt * max_ind - 0.5 * time[-1] 722 wave = np.fft.irfft( 723 np.exp(-1j * self.freq * 2 * np.pi * max_loc) * self.spectrum, axis=0 724 ) 725 return Waveform(wave=wave, time=time, dt=dt, is_uniformly_spaced=True) 726 727 def to_intensity_spectrum(self, wavelength_scaled: bool = True): 728 """Create an IntensitySpectrum based on the current ComplexSpectrum 729 730 Args: 731 wavelength_scaled (bool): Apply the wavelength^-2 Jakobian such to correspond to W/nm spectrum""" 732 new_spectrum = np.array(np.abs(self.spectrum[self.freq > 0.0]) ** 2) 733 new_freq = np.array(self.freq[self.freq > 0.0]) 734 new_wavelength = constants.speed_of_light / new_freq 735 if wavelength_scaled: 736 new_spectrum /= new_wavelength**2 737 return IntensitySpectrum( 738 spectrum=new_spectrum, 739 phase=np.array(np.angle(self.spectrum[self.freq > 0.0])), 740 freq=new_freq, 741 wavelength=new_wavelength, 742 is_frequency_scaled=wavelength_scaled, 743 )
Contains a complex spectrum, with spectral weights on a frequency scale
Attributes:
- spectrum (np.ndarray): the spectrum in complex128 format
- freq (np.ndarray):
673 def lock(self): 674 """ 675 Make the data immutable 676 """ 677 self.spectrum.setflags(write=False) 678 self.freq.setflags(write=False)
Make the data immutable
683 def to_time_derivative(self): 684 r"""Return a ComplexSpectrum corresponding to the time derivative (multiply by $i\omega$)""" 685 d_dt = 1j * 2 * np.pi * self.freq * self.spectrum 686 return ComplexSpectrum(spectrum=d_dt, freq=np.array(self.freq))
Return a ComplexSpectrum corresponding to the time derivative (multiply by $i\omega$)
688 def to_bandpassed(self, frequency: float, sigma: float, order: int = 4): 689 r""" 690 Return the complex spectrum after applying a supergaussian bandpass filter to the spectrum, of the form 691 $e^{\frac{(f-f_0)^r}{2\sigma^r}}$ 692 where $f_0$ is the frequency argument, $\sigma$ is the sigma argument, and r is the order argument 693 694 Args: 695 frequency: the central frequency (Hz) of the bandpass 696 sigma: the width of the bandpass (Hz) 697 order: the order of the supergaussian 698 """ 699 new_spectrum = self.spectrum * np.exp( 700 -((self.freq - frequency) ** order) / (2 * sigma**order) 701 ) 702 return ComplexSpectrum(spectrum=new_spectrum, freq=np.array(self.freq))
Return the complex spectrum after applying a supergaussian bandpass filter to the spectrum, of the form $e^{\frac{(f-f_0)^r}{2\sigma^r}}$ where $f_0$ is the frequency argument, $\sigma$ is the sigma argument, and r is the order argument
Arguments:
- frequency: the central frequency (Hz) of the bandpass
- sigma: the width of the bandpass (Hz)
- order: the order of the supergaussian
704 def to_waveform(self): 705 """ 706 Create a Waveform based on this complex spectrum. 707 """ 708 wave = np.fft.irfft(self.spectrum, axis=0) 709 dt = 0.5 / (self.freq[-1] - self.freq[0]) 710 time = dt * np.array(range(wave.shape[0])) 711 return Waveform(wave=wave, time=time, dt=dt, is_uniformly_spaced=True)
Create a Waveform based on this complex spectrum.
713 def to_centered_waveform(self): 714 """ 715 Create a Waveform based on this complex spectrum and center it in the time window 716 """ 717 wave = np.fft.irfft(self.spectrum, axis=0) 718 dt = 0.5 / (self.freq[-1] - self.freq[0]) 719 time = dt * np.array(range(wave.shape[0])) 720 max_ind, max_val = find_maximum_location(np.abs(np.array(sig.hilbert(wave)))) 721 max_loc = dt * max_ind - 0.5 * time[-1] 722 wave = np.fft.irfft( 723 np.exp(-1j * self.freq * 2 * np.pi * max_loc) * self.spectrum, axis=0 724 ) 725 return Waveform(wave=wave, time=time, dt=dt, is_uniformly_spaced=True)
Create a Waveform based on this complex spectrum and center it in the time window
727 def to_intensity_spectrum(self, wavelength_scaled: bool = True): 728 """Create an IntensitySpectrum based on the current ComplexSpectrum 729 730 Args: 731 wavelength_scaled (bool): Apply the wavelength^-2 Jakobian such to correspond to W/nm spectrum""" 732 new_spectrum = np.array(np.abs(self.spectrum[self.freq > 0.0]) ** 2) 733 new_freq = np.array(self.freq[self.freq > 0.0]) 734 new_wavelength = constants.speed_of_light / new_freq 735 if wavelength_scaled: 736 new_spectrum /= new_wavelength**2 737 return IntensitySpectrum( 738 spectrum=new_spectrum, 739 phase=np.array(np.angle(self.spectrum[self.freq > 0.0])), 740 freq=new_freq, 741 wavelength=new_wavelength, 742 is_frequency_scaled=wavelength_scaled, 743 )
Create an IntensitySpectrum based on the current ComplexSpectrum
Arguments:
- wavelength_scaled (bool): Apply the wavelength^-2 Jakobian such to correspond to W/nm spectrum
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file
746@yaml_io 747@dataclass(slots=True) 748class IntensitySpectrum: 749 """ 750 Contains an intensity spectrum - real valued. SI units 751 752 Attributes: 753 spectrum (Optional[np.ndarray]): the spectral intensity 754 phase (Optional[np.ndarray]): the spectral phase 755 freq (Optional[np.ndarray]): the frequencies corresponding to the spectrum 756 wavelength (Optional[np.ndarray]): the wavelengths 757 is_frequency_scaled (bool): has the lamda^-2 Jakobian been applied to Fourier transformed data? 758 """ 759 760 spectrum: np.ndarray 761 phase: Optional[np.ndarray] 762 freq: np.ndarray 763 wavelength: np.ndarray 764 is_frequency_scaled: bool = False 765 766 def lock(self): 767 """ 768 Make the data immutable 769 """ 770 self.spectrum.setflags(write=False) 771 self.freq.setflags(write=False) 772 self.wavelength.setflags(write=False) 773 if self.phase is not None: 774 self.phase.setflags(write=False) 775 776 def copy(self): 777 return copy.deepcopy(self) 778 779 def wavelength_nm(self): 780 """ 781 Returns: 782 Optional[np.ndarray]: the wavelengths in nm 783 """ 784 if self.wavelength is not None: 785 return 1e9 * self.wavelength 786 else: 787 return None 788 789 def wavelength_micron(self): 790 """ 791 Returns: 792 Optional[np.ndarray]: the wavelengths in microns 793 """ 794 if self.wavelength is not None: 795 return 1e6 * self.wavelength 796 else: 797 return None 798 799 @staticmethod 800 def from_spectrometer_spectrum_nanometers( 801 wavelengths_nanometers: np.ndarray, spectrum: np.ndarray 802 ): 803 """ 804 Generate an instance based on a spepctrum array and wavelength array in nm (typical spectrometer data) 805 806 Args: 807 wavelengths_nanometers: the wavelengths in nanometers 808 spectrum: the spectral intensity 809 """ 810 wavelengths = 1e-9 * wavelengths_nanometers 811 freqs = constants.speed_of_light / wavelengths 812 phase = np.zeros(freqs.shape, dtype=float) 813 return IntensitySpectrum( 814 spectrum=spectrum, wavelength=wavelengths, freq=freqs, phase=phase 815 ) 816 817 def get_transform_limited_pulse(self, gate_level: Optional[float] = None): 818 """ 819 Returns the transform-limited pulse corresponding to the spectrum. 820 821 Args: 822 gate_level (float): Apply a gate such that only values above gate_level*max(spectrum) are included 823 824 Returns: 825 ComplexEnvelope: the transform-limited pulse 826 """ 827 if self.spectrum is not None: 828 from ..spectrum import transform_limited_pulse_from_spectrometer 829 830 spec = np.array(self.spectrum) 831 lam = None 832 if self.wavelength is None and self.freq is not None: 833 lam = constants.speed_of_light / self.freq[self.freq > 0.0] 834 spec = spec[self.freq > 0.0] 835 elif self.wavelength is not None: 836 lam = self.wavelength 837 if lam is not None: 838 if self.is_frequency_scaled: 839 spec /= lam**2 840 t, intensity = transform_limited_pulse_from_spectrometer( 841 1e9 * lam, spec, gate_level 842 ) 843 return ComplexEnvelope( 844 envelope=np.sqrt(intensity), time=t, dt=t[1] - t[0] 845 ) 846 raise Exception("Missing data") 847 848 def get_wavelength_spectrum(self): 849 """ 850 Return a wavelength-scaled spectrum, independent of the state of the current instance 851 852 Returns: 853 np.ndarray, np.ndarray: wavelength and spectrum 854 """ 855 if self.is_frequency_scaled: 856 if self.freq is not None and self.spectrum is not None: 857 from ..spectrum import frequency_to_wavelength 858 859 return frequency_to_wavelength(self.freq, self.spectrum) 860 else: 861 raise Exception("Missing data") 862 else: 863 if self.wavelength is not None and self.spectrum is not None: 864 return self.wavelength, self.spectrum 865 else: 866 raise Exception("Missing data") 867 868 def get_frequency_spectrum(self): 869 """ 870 Return a frequency-scaled spectrum, independent of the state of the current instance 871 872 Returns: 873 np.ndarray, np.ndarray: wavelength and spectrum 874 """ 875 if self.is_frequency_scaled: 876 if self.freq is not None and self.spectrum is not None: 877 return self.freq, self.spectrum 878 else: 879 raise Exception("Missing data") 880 else: 881 if self.wavelength is not None and self.spectrum is not None: 882 from ..spectrum import wavelength_to_frequency 883 884 return wavelength_to_frequency(1e9 * self.wavelength, self.spectrum) 885 else: 886 raise Exception("Missing data") 887 888 def to_normalized(self): 889 """ 890 Returns a normalized version of the current instance. 891 """ 892 893 normalized_spectrum = self.spectrum / np.max(self.spectrum) 894 895 return IntensitySpectrum( 896 spectrum=normalized_spectrum, 897 phase=self.phase, 898 freq=np.array(self.freq), 899 wavelength=np.array(self.wavelength), 900 is_frequency_scaled=self.is_frequency_scaled, 901 ) 902 903 def to_interpolated_wavelength(self, new_wavelengths: np.ndarray): 904 new_spectrum = interpolate(new_wavelengths, self.wavelength, self.spectrum) 905 if self.phase is not None: 906 new_phase = interpolate(new_wavelengths, self.wavelength, self.phase) 907 else: 908 new_phase = None 909 910 return IntensitySpectrum( 911 spectrum=new_spectrum, 912 wavelength=new_wavelengths, 913 freq=constants.speed_of_light / new_wavelengths, 914 phase=new_phase, 915 is_frequency_scaled=self.is_frequency_scaled, 916 ) 917 918 def to_corrected_wavelength(self, new_wavelengths: np.ndarray): 919 assert len(new_wavelengths) == len(self.wavelength) 920 return IntensitySpectrum( 921 spectrum=self.spectrum, 922 wavelength=new_wavelengths, 923 freq=constants.speed_of_light / new_wavelengths, 924 phase=self.phase, 925 is_frequency_scaled=self.is_frequency_scaled, 926 ) 927 928 def plot_with_group_delay( 929 self, 930 ax: Optional[Axes] = None, 931 phase_blanking: float = 0.05, 932 shift_from_centered=True, 933 xlim=None, 934 ): 935 """ 936 Plot the spectrum and group delay curve. 937 938 Args: 939 ax: optionally plot onto a pre-existing matplotlib Axes 940 phase_blanking: only show phase information (group delay) above this level relative to max intensity 941 xlim: pass arguments to set_xlim() to constrain the x-axis 942 """ 943 if ax is None: 944 fig, ax = plt.subplots() 945 else: 946 fig = ax.get_figure() 947 948 start_index = np.argmax(self.spectrum > 0) 949 intensity = self.spectrum[start_index::] 950 freq = self.freq[start_index::] 951 wl = constants.speed_of_light / freq 952 if self.phase is not None: 953 if shift_from_centered: 954 shift = -0.5 / (freq[1] - freq[0]) 955 phase_shift = np.angle(np.exp(1j * 2 * np.pi * freq * shift)) 956 phase = np.unwrap(phase_shift + self.phase[start_index::]) 957 else: 958 phase = np.unwrap(self.phase[start_index::]) 959 960 else: 961 phase = np.zeros(intensity.shape, dtype=float) 962 963 intensity /= np.max(intensity) 964 intensity_line = ax.plot(1e9 * wl, intensity, label="Intensity") 965 ax.set_xlabel("Wavelength (nm)") 966 ax.set_ylabel("Intensity (Arb. unit)") 967 ax_phase = plt.twinx(ax) 968 group_delay = (1e15 / (2 * np.pi)) * derivative(phase, 1) / (freq[1] - freq[2]) 969 assert isinstance(ax_phase, Axes) 970 ax_phase.plot([], []) 971 phase_line = ax_phase.plot( 972 1e9 * wl[intensity > phase_blanking], 973 group_delay[intensity > phase_blanking], 974 "--", 975 label="Group delay", 976 ) 977 ax_phase.set_ylabel("Group delay (fs)") 978 if xlim is not None: 979 ax.set_xlim(xlim) 980 ax_phase.set_xlim(xlim) 981 lines = lines = intensity_line + phase_line 982 ax.legend(lines, [str(line.get_label()) for line in lines]) 983 return fig
Contains an intensity spectrum - real valued. SI units
Attributes:
- spectrum (Optional[np.ndarray]): the spectral intensity
- phase (Optional[np.ndarray]): the spectral phase
- freq (Optional[np.ndarray]): the frequencies corresponding to the spectrum
- wavelength (Optional[np.ndarray]): the wavelengths
- is_frequency_scaled (bool): has the lamda^-2 Jakobian been applied to Fourier transformed data?
766 def lock(self): 767 """ 768 Make the data immutable 769 """ 770 self.spectrum.setflags(write=False) 771 self.freq.setflags(write=False) 772 self.wavelength.setflags(write=False) 773 if self.phase is not None: 774 self.phase.setflags(write=False)
Make the data immutable
779 def wavelength_nm(self): 780 """ 781 Returns: 782 Optional[np.ndarray]: the wavelengths in nm 783 """ 784 if self.wavelength is not None: 785 return 1e9 * self.wavelength 786 else: 787 return None
Returns:
Optional[np.ndarray]: the wavelengths in nm
789 def wavelength_micron(self): 790 """ 791 Returns: 792 Optional[np.ndarray]: the wavelengths in microns 793 """ 794 if self.wavelength is not None: 795 return 1e6 * self.wavelength 796 else: 797 return None
Returns:
Optional[np.ndarray]: the wavelengths in microns
799 @staticmethod 800 def from_spectrometer_spectrum_nanometers( 801 wavelengths_nanometers: np.ndarray, spectrum: np.ndarray 802 ): 803 """ 804 Generate an instance based on a spepctrum array and wavelength array in nm (typical spectrometer data) 805 806 Args: 807 wavelengths_nanometers: the wavelengths in nanometers 808 spectrum: the spectral intensity 809 """ 810 wavelengths = 1e-9 * wavelengths_nanometers 811 freqs = constants.speed_of_light / wavelengths 812 phase = np.zeros(freqs.shape, dtype=float) 813 return IntensitySpectrum( 814 spectrum=spectrum, wavelength=wavelengths, freq=freqs, phase=phase 815 )
Generate an instance based on a spepctrum array and wavelength array in nm (typical spectrometer data)
Arguments:
- wavelengths_nanometers: the wavelengths in nanometers
- spectrum: the spectral intensity
817 def get_transform_limited_pulse(self, gate_level: Optional[float] = None): 818 """ 819 Returns the transform-limited pulse corresponding to the spectrum. 820 821 Args: 822 gate_level (float): Apply a gate such that only values above gate_level*max(spectrum) are included 823 824 Returns: 825 ComplexEnvelope: the transform-limited pulse 826 """ 827 if self.spectrum is not None: 828 from ..spectrum import transform_limited_pulse_from_spectrometer 829 830 spec = np.array(self.spectrum) 831 lam = None 832 if self.wavelength is None and self.freq is not None: 833 lam = constants.speed_of_light / self.freq[self.freq > 0.0] 834 spec = spec[self.freq > 0.0] 835 elif self.wavelength is not None: 836 lam = self.wavelength 837 if lam is not None: 838 if self.is_frequency_scaled: 839 spec /= lam**2 840 t, intensity = transform_limited_pulse_from_spectrometer( 841 1e9 * lam, spec, gate_level 842 ) 843 return ComplexEnvelope( 844 envelope=np.sqrt(intensity), time=t, dt=t[1] - t[0] 845 ) 846 raise Exception("Missing data")
Returns the transform-limited pulse corresponding to the spectrum.
Arguments:
- gate_level (float): Apply a gate such that only values above gate_level*max(spectrum) are included
Returns:
ComplexEnvelope: the transform-limited pulse
848 def get_wavelength_spectrum(self): 849 """ 850 Return a wavelength-scaled spectrum, independent of the state of the current instance 851 852 Returns: 853 np.ndarray, np.ndarray: wavelength and spectrum 854 """ 855 if self.is_frequency_scaled: 856 if self.freq is not None and self.spectrum is not None: 857 from ..spectrum import frequency_to_wavelength 858 859 return frequency_to_wavelength(self.freq, self.spectrum) 860 else: 861 raise Exception("Missing data") 862 else: 863 if self.wavelength is not None and self.spectrum is not None: 864 return self.wavelength, self.spectrum 865 else: 866 raise Exception("Missing data")
Return a wavelength-scaled spectrum, independent of the state of the current instance
Returns:
np.ndarray, np.ndarray: wavelength and spectrum
868 def get_frequency_spectrum(self): 869 """ 870 Return a frequency-scaled spectrum, independent of the state of the current instance 871 872 Returns: 873 np.ndarray, np.ndarray: wavelength and spectrum 874 """ 875 if self.is_frequency_scaled: 876 if self.freq is not None and self.spectrum is not None: 877 return self.freq, self.spectrum 878 else: 879 raise Exception("Missing data") 880 else: 881 if self.wavelength is not None and self.spectrum is not None: 882 from ..spectrum import wavelength_to_frequency 883 884 return wavelength_to_frequency(1e9 * self.wavelength, self.spectrum) 885 else: 886 raise Exception("Missing data")
Return a frequency-scaled spectrum, independent of the state of the current instance
Returns:
np.ndarray, np.ndarray: wavelength and spectrum
888 def to_normalized(self): 889 """ 890 Returns a normalized version of the current instance. 891 """ 892 893 normalized_spectrum = self.spectrum / np.max(self.spectrum) 894 895 return IntensitySpectrum( 896 spectrum=normalized_spectrum, 897 phase=self.phase, 898 freq=np.array(self.freq), 899 wavelength=np.array(self.wavelength), 900 is_frequency_scaled=self.is_frequency_scaled, 901 )
Returns a normalized version of the current instance.
903 def to_interpolated_wavelength(self, new_wavelengths: np.ndarray): 904 new_spectrum = interpolate(new_wavelengths, self.wavelength, self.spectrum) 905 if self.phase is not None: 906 new_phase = interpolate(new_wavelengths, self.wavelength, self.phase) 907 else: 908 new_phase = None 909 910 return IntensitySpectrum( 911 spectrum=new_spectrum, 912 wavelength=new_wavelengths, 913 freq=constants.speed_of_light / new_wavelengths, 914 phase=new_phase, 915 is_frequency_scaled=self.is_frequency_scaled, 916 )
918 def to_corrected_wavelength(self, new_wavelengths: np.ndarray): 919 assert len(new_wavelengths) == len(self.wavelength) 920 return IntensitySpectrum( 921 spectrum=self.spectrum, 922 wavelength=new_wavelengths, 923 freq=constants.speed_of_light / new_wavelengths, 924 phase=self.phase, 925 is_frequency_scaled=self.is_frequency_scaled, 926 )
928 def plot_with_group_delay( 929 self, 930 ax: Optional[Axes] = None, 931 phase_blanking: float = 0.05, 932 shift_from_centered=True, 933 xlim=None, 934 ): 935 """ 936 Plot the spectrum and group delay curve. 937 938 Args: 939 ax: optionally plot onto a pre-existing matplotlib Axes 940 phase_blanking: only show phase information (group delay) above this level relative to max intensity 941 xlim: pass arguments to set_xlim() to constrain the x-axis 942 """ 943 if ax is None: 944 fig, ax = plt.subplots() 945 else: 946 fig = ax.get_figure() 947 948 start_index = np.argmax(self.spectrum > 0) 949 intensity = self.spectrum[start_index::] 950 freq = self.freq[start_index::] 951 wl = constants.speed_of_light / freq 952 if self.phase is not None: 953 if shift_from_centered: 954 shift = -0.5 / (freq[1] - freq[0]) 955 phase_shift = np.angle(np.exp(1j * 2 * np.pi * freq * shift)) 956 phase = np.unwrap(phase_shift + self.phase[start_index::]) 957 else: 958 phase = np.unwrap(self.phase[start_index::]) 959 960 else: 961 phase = np.zeros(intensity.shape, dtype=float) 962 963 intensity /= np.max(intensity) 964 intensity_line = ax.plot(1e9 * wl, intensity, label="Intensity") 965 ax.set_xlabel("Wavelength (nm)") 966 ax.set_ylabel("Intensity (Arb. unit)") 967 ax_phase = plt.twinx(ax) 968 group_delay = (1e15 / (2 * np.pi)) * derivative(phase, 1) / (freq[1] - freq[2]) 969 assert isinstance(ax_phase, Axes) 970 ax_phase.plot([], []) 971 phase_line = ax_phase.plot( 972 1e9 * wl[intensity > phase_blanking], 973 group_delay[intensity > phase_blanking], 974 "--", 975 label="Group delay", 976 ) 977 ax_phase.set_ylabel("Group delay (fs)") 978 if xlim is not None: 979 ax.set_xlim(xlim) 980 ax_phase.set_xlim(xlim) 981 lines = lines = intensity_line + phase_line 982 ax.legend(lines, [str(line.get_label()) for line in lines]) 983 return fig
Plot the spectrum and group delay curve.
Arguments:
- ax: optionally plot onto a pre-existing matplotlib Axes
- phase_blanking: only show phase information (group delay) above this level relative to max intensity
- xlim: pass arguments to set_xlim() to constrain the x-axis
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file
986@yaml_io 987@dataclass(slots=True) 988class ComplexEnvelope: 989 """ 990 Data corresponding to a complex envelope of a pulse, e.g. from a FROG measurement. 991 992 Attributes: 993 envelope (np.ndarray): the complex envelope 994 time: (np.ndarray): the time array 995 dt (float): the time step 996 carrier_frequency (float): the carrier frequency of the envelope 997 """ 998 999 envelope: np.ndarray 1000 time: np.ndarray 1001 dt: float 1002 carrier_frequency: float = 0.0 1003 1004 def lock(self): 1005 """ 1006 Make the data immutable 1007 """ 1008 self.envelope.setflags(write=False) 1009 self.time.setflags(write=False) 1010 1011 def time_fs(self): 1012 return 1e15 * self.time 1013 1014 def copy(self): 1015 return copy.deepcopy(self) 1016 1017 def get_fwhm(self) -> float: 1018 """ 1019 Full-width-at-half-maximum value of the envelope 1020 Returns: 1021 float: the fwhm 1022 """ 1023 return fwhm(np.abs(self.envelope) ** 2, self.dt) 1024 1025 def to_complex_spectrum(self, padding_factor: int = 1) -> ComplexSpectrum: 1026 """ 1027 Returns a ComplexSpectrum based on the data 1028 """ 1029 1030 return ComplexSpectrum( 1031 spectrum=np.fft.rfft( 1032 self.envelope, self.envelope.shape[0] * padding_factor 1033 ), 1034 freq=np.fft.rfftfreq(self.envelope.shape[0] * padding_factor, self.dt) 1035 + self.carrier_frequency, 1036 ) 1037 1038 def to_waveform( 1039 self, interpolation_factor: int = 1, CEP_shift: float = 0.0 1040 ) -> Waveform: 1041 """ 1042 Returns a Waveform based on the data 1043 """ 1044 output_dt = self.dt / interpolation_factor 1045 output_time = self.time[0] + output_dt * np.array( 1046 range(self.time.shape[0] * interpolation_factor) 1047 ) 1048 if interpolation_factor != 1: 1049 output_envelope = sig.resample( 1050 np.real(self.envelope), output_time.shape[0] 1051 ) + 1j * np.array( 1052 sig.resample(np.imag(self.envelope), output_time.shape[0]) 1053 ) 1054 else: 1055 output_envelope = self.envelope 1056 return Waveform( 1057 wave=np.real( 1058 np.exp( 1059 1j * self.carrier_frequency * 2 * np.pi * output_time 1060 - 1j * CEP_shift 1061 ) 1062 * output_envelope 1063 ), 1064 time=output_time, 1065 dt=output_dt, 1066 is_uniformly_spaced=True, 1067 ) 1068 1069 def plot(self, ax: Optional[Axes] = None, phase_blanking: float = 0.05, xlim=None): 1070 """ 1071 Plot the pulse. 1072 1073 Args: 1074 ax: optionally plot onto a pre-existing matplotlib Axes 1075 phase_blanking: only show phase information (instantaneous frequency) above this level relative to max intensity 1076 xlim: pass arguments to set_xlim() to constrain the x-axis 1077 """ 1078 if ax is None: 1079 fig, ax = plt.subplots() 1080 else: 1081 fig = ax.get_figure() 1082 time_ax = self.time_fs() - np.mean(self.time_fs()) 1083 intensity = np.abs(self.envelope) ** 2 1084 intensity /= np.max(intensity) 1085 intensity_line = ax.plot( 1086 time_ax, 1087 intensity, 1088 label=f"Intensity, fwhm {1e15 * self.get_fwhm():0.1f} fs", 1089 ) 1090 ax.set_xlabel("Time (fs)") 1091 ax.set_ylabel("Intensity (Arb. unit)") 1092 inst_freq = ( 1093 (1e-12 / (2 * np.pi)) 1094 * derivative(np.unwrap(np.angle(self.envelope)), 1) 1095 / self.dt 1096 ) 1097 ax_phase = plt.twinx(ax) 1098 assert isinstance(ax_phase, Axes) 1099 ax_phase.plot([], []) 1100 phase_line = ax_phase.plot( 1101 time_ax[intensity > phase_blanking], 1102 inst_freq[intensity > phase_blanking], 1103 "--", 1104 label="Inst. frequency", 1105 ) 1106 ax_phase.set_ylabel("Inst. frequency (THz)") 1107 if xlim is not None: 1108 ax.set_xlim(xlim) 1109 ax_phase.set_xlim(xlim) 1110 lines = lines = intensity_line + phase_line 1111 ax.legend(lines, [str(line.get_label()) for line in lines]) 1112 return fig
Data corresponding to a complex envelope of a pulse, e.g. from a FROG measurement.
Attributes:
- envelope (np.ndarray): the complex envelope
- time: (np.ndarray): the time array
- dt (float): the time step
- carrier_frequency (float): the carrier frequency of the envelope
1004 def lock(self): 1005 """ 1006 Make the data immutable 1007 """ 1008 self.envelope.setflags(write=False) 1009 self.time.setflags(write=False)
Make the data immutable
1017 def get_fwhm(self) -> float: 1018 """ 1019 Full-width-at-half-maximum value of the envelope 1020 Returns: 1021 float: the fwhm 1022 """ 1023 return fwhm(np.abs(self.envelope) ** 2, self.dt)
Full-width-at-half-maximum value of the envelope
Returns:
float: the fwhm
1025 def to_complex_spectrum(self, padding_factor: int = 1) -> ComplexSpectrum: 1026 """ 1027 Returns a ComplexSpectrum based on the data 1028 """ 1029 1030 return ComplexSpectrum( 1031 spectrum=np.fft.rfft( 1032 self.envelope, self.envelope.shape[0] * padding_factor 1033 ), 1034 freq=np.fft.rfftfreq(self.envelope.shape[0] * padding_factor, self.dt) 1035 + self.carrier_frequency, 1036 )
Returns a ComplexSpectrum based on the data
1038 def to_waveform( 1039 self, interpolation_factor: int = 1, CEP_shift: float = 0.0 1040 ) -> Waveform: 1041 """ 1042 Returns a Waveform based on the data 1043 """ 1044 output_dt = self.dt / interpolation_factor 1045 output_time = self.time[0] + output_dt * np.array( 1046 range(self.time.shape[0] * interpolation_factor) 1047 ) 1048 if interpolation_factor != 1: 1049 output_envelope = sig.resample( 1050 np.real(self.envelope), output_time.shape[0] 1051 ) + 1j * np.array( 1052 sig.resample(np.imag(self.envelope), output_time.shape[0]) 1053 ) 1054 else: 1055 output_envelope = self.envelope 1056 return Waveform( 1057 wave=np.real( 1058 np.exp( 1059 1j * self.carrier_frequency * 2 * np.pi * output_time 1060 - 1j * CEP_shift 1061 ) 1062 * output_envelope 1063 ), 1064 time=output_time, 1065 dt=output_dt, 1066 is_uniformly_spaced=True, 1067 )
Returns a Waveform based on the data
1069 def plot(self, ax: Optional[Axes] = None, phase_blanking: float = 0.05, xlim=None): 1070 """ 1071 Plot the pulse. 1072 1073 Args: 1074 ax: optionally plot onto a pre-existing matplotlib Axes 1075 phase_blanking: only show phase information (instantaneous frequency) above this level relative to max intensity 1076 xlim: pass arguments to set_xlim() to constrain the x-axis 1077 """ 1078 if ax is None: 1079 fig, ax = plt.subplots() 1080 else: 1081 fig = ax.get_figure() 1082 time_ax = self.time_fs() - np.mean(self.time_fs()) 1083 intensity = np.abs(self.envelope) ** 2 1084 intensity /= np.max(intensity) 1085 intensity_line = ax.plot( 1086 time_ax, 1087 intensity, 1088 label=f"Intensity, fwhm {1e15 * self.get_fwhm():0.1f} fs", 1089 ) 1090 ax.set_xlabel("Time (fs)") 1091 ax.set_ylabel("Intensity (Arb. unit)") 1092 inst_freq = ( 1093 (1e-12 / (2 * np.pi)) 1094 * derivative(np.unwrap(np.angle(self.envelope)), 1) 1095 / self.dt 1096 ) 1097 ax_phase = plt.twinx(ax) 1098 assert isinstance(ax_phase, Axes) 1099 ax_phase.plot([], []) 1100 phase_line = ax_phase.plot( 1101 time_ax[intensity > phase_blanking], 1102 inst_freq[intensity > phase_blanking], 1103 "--", 1104 label="Inst. frequency", 1105 ) 1106 ax_phase.set_ylabel("Inst. frequency (THz)") 1107 if xlim is not None: 1108 ax.set_xlim(xlim) 1109 ax_phase.set_xlim(xlim) 1110 lines = lines = intensity_line + phase_line 1111 ax.legend(lines, [str(line.get_label()) for line in lines]) 1112 return fig
Plot the pulse.
Arguments:
- ax: optionally plot onto a pre-existing matplotlib Axes
- phase_blanking: only show phase information (instantaneous frequency) above this level relative to max intensity
- xlim: pass arguments to set_xlim() to constrain the x-axis
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file
214@yaml_io 215@dataclass(slots=True) 216class Spectrogram: 217 """ 218 Contains the data describing a spectrogram 219 220 Attributes: 221 data (np.ndarray): 2d spectrogram 222 time (np.ndarray): time vector 223 freq (np.ndarray): frequency vector""" 224 225 data: np.ndarray 226 time: np.ndarray 227 freq: np.ndarray 228 229 def lock(self): 230 """ 231 Make the data immutable 232 """ 233 self.data.setflags(write=False) 234 self.time.setflags(write=False) 235 self.freq.setflags(write=False) 236 237 def save(self, filename): 238 """ 239 Save in the .A.dat file format used by FROG .etc 240 Args: 241 filename: the file to be saved 242 243 The file is structured like this: 244 [number of wavelengths] [number of times] 245 [minimum value of the trace] [maximum value of the trace] 246 [array of wavelengths] 247 [array of times] 248 [data array as single column] 249 """ 250 with open(filename, "w") as file: 251 file.write(f"{len(self.freq)}\t{len(self.time)}\n") 252 file.write(f"{np.min(self.data[:])}\t{np.max(self.data[:])}\n") 253 for freq in self.freq: 254 wavelength_nm = 1e9 * constants.speed_of_light / freq 255 file.write(f"{wavelength_nm:15.15g}\n") 256 for time in self.time: 257 time_fs = 1e15 * time 258 file.write(f"{time_fs:15.15g}\n") 259 for x in self.data: 260 for y in x: 261 file.write(f"{y:15.15g}\n") 262 263 def to_block_binned(self, freq_bin: int, time_bin: int, method: str = "mean"): 264 """ 265 Apply block-binning to the spectrogram. 266 267 Args: 268 freq_bin (int): block size for averaging in the frequency direction 269 time_bin (int): block size for averaging in the time-direction 270 method (str): can be ```mean``` or ```median``` 271 """ 272 return Spectrogram( 273 data=block_binning_2d(self.data, time_bin, freq_bin, method), 274 freq=block_binning_1d(self.freq, freq_bin, "mean"), 275 time=block_binning_1d(self.time, time_bin, "mean"), 276 ) 277 278 def to_per_frequency_dc_removed(self, extra_offset: float = 0.0): 279 """Perform DC offset removal on a measured spectrogram, on a per-frequency basis 280 281 Args: 282 extra_offset (float): subtract a value from the entire array (negative values are always set to zero) 283 284 Returns: 285 Spectrogram: the spectrogram with offset removed.""" 286 new_data = np.array(self.data) 287 new_data -= extra_offset 288 new_data[new_data < 0.0] = 0.0 289 for _i in range(new_data.shape[0]): 290 new_data[_i, :] -= np.min(new_data[_i, :]) 291 292 return Spectrogram(data=new_data, time=self.time, freq=self.freq) 293 294 def to_symmetrized(self): 295 """ 296 Average the trace with a time-reversed copy. This might be useful for getting a reconstruction of difficult data, but keep in mind that the resulting measured trace will no longer represent the real measurement and should not be published as such. 297 """ 298 return Spectrogram( 299 data=0.5 * (self.data + np.fliplr(self.data)), 300 time=self.time, 301 freq=self.freq, 302 ) 303 304 def to_removed_spatial_chirp(self): 305 """ 306 Remove the effects of spatial chirp on an SHG-FROG trace by centering all single-frequency autocorrelations to the same time-zero 307 """ 308 new_data = np.array(self.data) 309 for i in range(len(self.freq)): 310 total = np.sum(self.data[i, :]) 311 if total > 0.0: 312 t0 = np.sum(self.time * self.data[i, :]) / total 313 new_data[i, :] = interpolate(self.time + t0, self.time, self.data[i, :]) 314 315 return Spectrogram(data=new_data, time=self.time, freq=self.freq) 316 317 def to_combined_and_binned( 318 self, 319 other, 320 stitching_band: Tuple[float, float], 321 dim: int = 64, 322 dt: float = 5e-15, 323 t0: Optional[Tuple[float, float]] = None, 324 f0: float = 750e12, 325 ): 326 """ 327 Bin two different spectrograms, e.g. from different spectrometers, onto the time time/frequency grid 328 329 Args: 330 other: the other spectrogram 331 stitching_band (Tuple[float, float]): the lower and upper frequency of the band where the two spectrometers should have equivalent response (hopefully there is one) 332 dim (int): size of each size of the resulting square data 333 dt (float): time step of the data 334 t0: (Optional[Tuple[float, float]): time-zero of the data (this, and other). If not specified, will be calculated by the first moment of the time-distribution of the signal 335 f0: (float): central frequency of the binned array 336 """ 337 t0_self = None 338 t0_other = None 339 340 if t0 is not None: 341 t0_self = t0[0] 342 t0_other = t0[1] 343 344 binned_self = self.to_binned(dim, dt, t0_self, f0) 345 binned_other = other.to_binned(dim, dt, t0_other, f0) 346 freq = binned_self.freq 347 348 # add more logic here to combine the spectrograms 349 stitching_band_integral_self = np.sum( 350 binned_self.data[ 351 ((freq > stitching_band[0]) & (freq < stitching_band[1])), : 352 ][:] 353 ) 354 stitching_band_integral_other = np.sum( 355 binned_other.data[ 356 ((freq > stitching_band[0]) & (freq < stitching_band[1])), : 357 ][:] 358 ) 359 weights_self = np.zeros(binned_self.freq.shape, dtype=float) 360 weights_other = np.zeros(binned_other.freq.shape, dtype=float) 361 other_multiplier = stitching_band_integral_self / stitching_band_integral_other 362 for i in range(len(freq)): 363 sum_self = np.sum(binned_self.data[i, :]) 364 sum_other = other_multiplier * np.sum(binned_other.data[i, :]) 365 total = sum_self + sum_other 366 if total > 0.0: 367 weight_self = sum_self / total 368 weight_other = other_multiplier * sum_other / total 369 weights_self[i] = weight_self 370 weights_other[i] = weight_other 371 372 return Spectrogram( 373 data=weights_self[:, np.newaxis] * binned_self.data 374 + weights_other[:, np.newaxis] * binned_other.data, 375 time=binned_self.time, 376 freq=binned_self.freq, 377 ) 378 379 def to_binned( 380 self, 381 dim: int = 64, 382 dt: float = 5e-15, 383 t0: Optional[float] = None, 384 f0: float = 750e12, 385 ): 386 """Bin a spectrogram to a FFT-appropriate shape 387 388 Args: 389 dim (int): size of each size of the resulting square data 390 dt (float): time step of the data 391 t0: (Optional[float]): time-zero of the data. If not specified, will be calculated by the first moment of the time-distribution of the signal 392 f0: (float): central frequency of the binned array 393 394 Returns: 395 Spectrogram: the binned spectrogram 396 """ 397 _t = np.array(range(dim)) * dt 398 _t -= np.mean(_t) 399 _f = np.fft.fftshift(np.fft.fftfreq(dim, d=dt) + f0) 400 binned_data = np.zeros((dim, self.time.shape[0]), dtype=float) 401 for _i in range(self.time.shape[0]): 402 binned_data[:, _i] = interpolate( 403 _f, self.freq, np.array(self.data[:, _i]), neighbors=2 404 ) 405 binned_data /= np.max(binned_data[:]) 406 if t0 is None: 407 ac = np.sum(binned_data, axis=0) 408 t0 = np.sum(ac * self.time) / np.sum(ac) 409 binned_data_square = np.zeros((dim, dim), dtype=float) 410 for _i in range(dim): 411 binned_data_square[_i, :] = interpolate( 412 _t, self.time - t0, np.array(binned_data[_i, :]), neighbors=2 413 ) 414 return Spectrogram(data=binned_data_square, time=_t, freq=_f) 415 416 def plot(self, ax: Optional[Axes] = None): 417 """ 418 Plot the spectrogram. 419 420 Args: 421 ax: optionally plot onto a pre-existing matplotlib Axes 422 """ 423 if ax is None: 424 fig, ax = plt.subplots() 425 else: 426 fig = ax.get_figure() 427 a = ax.pcolormesh( 428 1e15 * self.time, 429 1e-12 * self.freq, 430 self.data / np.max(self.data[:]), 431 rasterized=True, 432 ) 433 ax.set_xlabel("Time (fs)") 434 ax.set_ylabel("Frequency (THz)") 435 plt.colorbar(a) 436 return fig 437 438 def plot_log(self, ax: Optional[Axes] = None): 439 """ 440 Plot the spectrogram. 441 442 Args: 443 ax: optionally plot onto a pre-existing matplotlib Axes 444 """ 445 if ax is None: 446 fig, ax = plt.subplots() 447 else: 448 fig = ax.get_figure() 449 logdata = np.array(self.data) 450 logdata[self.data > 0.0] = np.log(self.data[self.data > 0.0]) 451 logdata[self.data < 0.0] = 0.0 452 a = ax.pcolormesh( 453 1e15 * self.time, 454 1e-12 * self.freq, 455 logdata, 456 rasterized=True, 457 vmin=-11, 458 vmax=0, 459 ) 460 ax.set_xlabel("Time (fs)") 461 ax.set_ylabel("Frequency (THz)") 462 ax.grid(True, lw=1) 463 plt.colorbar(a) 464 return fig
Contains the data describing a spectrogram
Attributes:
- data (np.ndarray): 2d spectrogram
- time (np.ndarray): time vector
- freq (np.ndarray): frequency vector
229 def lock(self): 230 """ 231 Make the data immutable 232 """ 233 self.data.setflags(write=False) 234 self.time.setflags(write=False) 235 self.freq.setflags(write=False)
Make the data immutable
237 def save(self, filename): 238 """ 239 Save in the .A.dat file format used by FROG .etc 240 Args: 241 filename: the file to be saved 242 243 The file is structured like this: 244 [number of wavelengths] [number of times] 245 [minimum value of the trace] [maximum value of the trace] 246 [array of wavelengths] 247 [array of times] 248 [data array as single column] 249 """ 250 with open(filename, "w") as file: 251 file.write(f"{len(self.freq)}\t{len(self.time)}\n") 252 file.write(f"{np.min(self.data[:])}\t{np.max(self.data[:])}\n") 253 for freq in self.freq: 254 wavelength_nm = 1e9 * constants.speed_of_light / freq 255 file.write(f"{wavelength_nm:15.15g}\n") 256 for time in self.time: 257 time_fs = 1e15 * time 258 file.write(f"{time_fs:15.15g}\n") 259 for x in self.data: 260 for y in x: 261 file.write(f"{y:15.15g}\n")
Save in the .A.dat file format used by FROG .etc
Arguments:
- filename: the file to be saved
The file is structured like this: [number of wavelengths] [number of times] [minimum value of the trace] [maximum value of the trace] [array of wavelengths] [array of times] [data array as single column]
263 def to_block_binned(self, freq_bin: int, time_bin: int, method: str = "mean"): 264 """ 265 Apply block-binning to the spectrogram. 266 267 Args: 268 freq_bin (int): block size for averaging in the frequency direction 269 time_bin (int): block size for averaging in the time-direction 270 method (str): can be ```mean``` or ```median``` 271 """ 272 return Spectrogram( 273 data=block_binning_2d(self.data, time_bin, freq_bin, method), 274 freq=block_binning_1d(self.freq, freq_bin, "mean"), 275 time=block_binning_1d(self.time, time_bin, "mean"), 276 )
Apply block-binning to the spectrogram.
Arguments:
- freq_bin (int): block size for averaging in the frequency direction
- time_bin (int): block size for averaging in the time-direction
- method (str): can be
mean
ormedian
278 def to_per_frequency_dc_removed(self, extra_offset: float = 0.0): 279 """Perform DC offset removal on a measured spectrogram, on a per-frequency basis 280 281 Args: 282 extra_offset (float): subtract a value from the entire array (negative values are always set to zero) 283 284 Returns: 285 Spectrogram: the spectrogram with offset removed.""" 286 new_data = np.array(self.data) 287 new_data -= extra_offset 288 new_data[new_data < 0.0] = 0.0 289 for _i in range(new_data.shape[0]): 290 new_data[_i, :] -= np.min(new_data[_i, :]) 291 292 return Spectrogram(data=new_data, time=self.time, freq=self.freq)
Perform DC offset removal on a measured spectrogram, on a per-frequency basis
Arguments:
- extra_offset (float): subtract a value from the entire array (negative values are always set to zero)
Returns:
Spectrogram: the spectrogram with offset removed.
294 def to_symmetrized(self): 295 """ 296 Average the trace with a time-reversed copy. This might be useful for getting a reconstruction of difficult data, but keep in mind that the resulting measured trace will no longer represent the real measurement and should not be published as such. 297 """ 298 return Spectrogram( 299 data=0.5 * (self.data + np.fliplr(self.data)), 300 time=self.time, 301 freq=self.freq, 302 )
Average the trace with a time-reversed copy. This might be useful for getting a reconstruction of difficult data, but keep in mind that the resulting measured trace will no longer represent the real measurement and should not be published as such.
304 def to_removed_spatial_chirp(self): 305 """ 306 Remove the effects of spatial chirp on an SHG-FROG trace by centering all single-frequency autocorrelations to the same time-zero 307 """ 308 new_data = np.array(self.data) 309 for i in range(len(self.freq)): 310 total = np.sum(self.data[i, :]) 311 if total > 0.0: 312 t0 = np.sum(self.time * self.data[i, :]) / total 313 new_data[i, :] = interpolate(self.time + t0, self.time, self.data[i, :]) 314 315 return Spectrogram(data=new_data, time=self.time, freq=self.freq)
Remove the effects of spatial chirp on an SHG-FROG trace by centering all single-frequency autocorrelations to the same time-zero
317 def to_combined_and_binned( 318 self, 319 other, 320 stitching_band: Tuple[float, float], 321 dim: int = 64, 322 dt: float = 5e-15, 323 t0: Optional[Tuple[float, float]] = None, 324 f0: float = 750e12, 325 ): 326 """ 327 Bin two different spectrograms, e.g. from different spectrometers, onto the time time/frequency grid 328 329 Args: 330 other: the other spectrogram 331 stitching_band (Tuple[float, float]): the lower and upper frequency of the band where the two spectrometers should have equivalent response (hopefully there is one) 332 dim (int): size of each size of the resulting square data 333 dt (float): time step of the data 334 t0: (Optional[Tuple[float, float]): time-zero of the data (this, and other). If not specified, will be calculated by the first moment of the time-distribution of the signal 335 f0: (float): central frequency of the binned array 336 """ 337 t0_self = None 338 t0_other = None 339 340 if t0 is not None: 341 t0_self = t0[0] 342 t0_other = t0[1] 343 344 binned_self = self.to_binned(dim, dt, t0_self, f0) 345 binned_other = other.to_binned(dim, dt, t0_other, f0) 346 freq = binned_self.freq 347 348 # add more logic here to combine the spectrograms 349 stitching_band_integral_self = np.sum( 350 binned_self.data[ 351 ((freq > stitching_band[0]) & (freq < stitching_band[1])), : 352 ][:] 353 ) 354 stitching_band_integral_other = np.sum( 355 binned_other.data[ 356 ((freq > stitching_band[0]) & (freq < stitching_band[1])), : 357 ][:] 358 ) 359 weights_self = np.zeros(binned_self.freq.shape, dtype=float) 360 weights_other = np.zeros(binned_other.freq.shape, dtype=float) 361 other_multiplier = stitching_band_integral_self / stitching_band_integral_other 362 for i in range(len(freq)): 363 sum_self = np.sum(binned_self.data[i, :]) 364 sum_other = other_multiplier * np.sum(binned_other.data[i, :]) 365 total = sum_self + sum_other 366 if total > 0.0: 367 weight_self = sum_self / total 368 weight_other = other_multiplier * sum_other / total 369 weights_self[i] = weight_self 370 weights_other[i] = weight_other 371 372 return Spectrogram( 373 data=weights_self[:, np.newaxis] * binned_self.data 374 + weights_other[:, np.newaxis] * binned_other.data, 375 time=binned_self.time, 376 freq=binned_self.freq, 377 )
Bin two different spectrograms, e.g. from different spectrometers, onto the time time/frequency grid
Arguments:
- other: the other spectrogram
- stitching_band (Tuple[float, float]): the lower and upper frequency of the band where the two spectrometers should have equivalent response (hopefully there is one)
- dim (int): size of each size of the resulting square data
- dt (float): time step of the data
- t0: (Optional[Tuple[float, float]): time-zero of the data (this, and other). If not specified, will be calculated by the first moment of the time-distribution of the signal
- f0: (float): central frequency of the binned array
379 def to_binned( 380 self, 381 dim: int = 64, 382 dt: float = 5e-15, 383 t0: Optional[float] = None, 384 f0: float = 750e12, 385 ): 386 """Bin a spectrogram to a FFT-appropriate shape 387 388 Args: 389 dim (int): size of each size of the resulting square data 390 dt (float): time step of the data 391 t0: (Optional[float]): time-zero of the data. If not specified, will be calculated by the first moment of the time-distribution of the signal 392 f0: (float): central frequency of the binned array 393 394 Returns: 395 Spectrogram: the binned spectrogram 396 """ 397 _t = np.array(range(dim)) * dt 398 _t -= np.mean(_t) 399 _f = np.fft.fftshift(np.fft.fftfreq(dim, d=dt) + f0) 400 binned_data = np.zeros((dim, self.time.shape[0]), dtype=float) 401 for _i in range(self.time.shape[0]): 402 binned_data[:, _i] = interpolate( 403 _f, self.freq, np.array(self.data[:, _i]), neighbors=2 404 ) 405 binned_data /= np.max(binned_data[:]) 406 if t0 is None: 407 ac = np.sum(binned_data, axis=0) 408 t0 = np.sum(ac * self.time) / np.sum(ac) 409 binned_data_square = np.zeros((dim, dim), dtype=float) 410 for _i in range(dim): 411 binned_data_square[_i, :] = interpolate( 412 _t, self.time - t0, np.array(binned_data[_i, :]), neighbors=2 413 ) 414 return Spectrogram(data=binned_data_square, time=_t, freq=_f)
Bin a spectrogram to a FFT-appropriate shape
Arguments:
- dim (int): size of each size of the resulting square data
- dt (float): time step of the data
- t0: (Optional[float]): time-zero of the data. If not specified, will be calculated by the first moment of the time-distribution of the signal
- f0: (float): central frequency of the binned array
Returns:
Spectrogram: the binned spectrogram
416 def plot(self, ax: Optional[Axes] = None): 417 """ 418 Plot the spectrogram. 419 420 Args: 421 ax: optionally plot onto a pre-existing matplotlib Axes 422 """ 423 if ax is None: 424 fig, ax = plt.subplots() 425 else: 426 fig = ax.get_figure() 427 a = ax.pcolormesh( 428 1e15 * self.time, 429 1e-12 * self.freq, 430 self.data / np.max(self.data[:]), 431 rasterized=True, 432 ) 433 ax.set_xlabel("Time (fs)") 434 ax.set_ylabel("Frequency (THz)") 435 plt.colorbar(a) 436 return fig
Plot the spectrogram.
Arguments:
- ax: optionally plot onto a pre-existing matplotlib Axes
438 def plot_log(self, ax: Optional[Axes] = None): 439 """ 440 Plot the spectrogram. 441 442 Args: 443 ax: optionally plot onto a pre-existing matplotlib Axes 444 """ 445 if ax is None: 446 fig, ax = plt.subplots() 447 else: 448 fig = ax.get_figure() 449 logdata = np.array(self.data) 450 logdata[self.data > 0.0] = np.log(self.data[self.data > 0.0]) 451 logdata[self.data < 0.0] = 0.0 452 a = ax.pcolormesh( 453 1e15 * self.time, 454 1e-12 * self.freq, 455 logdata, 456 rasterized=True, 457 vmin=-11, 458 vmax=0, 459 ) 460 ax.set_xlabel("Time (fs)") 461 ax.set_ylabel("Frequency (THz)") 462 ax.grid(True, lw=1) 463 plt.colorbar(a) 464 return fig
Plot the spectrogram.
Arguments:
- ax: optionally plot onto a pre-existing matplotlib Axes
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file
1456@yaml_io 1457@dataclass 1458class CalibrationDataset: 1459 measurement: IntensitySpectrum 1460 reference: IntensitySpectrum 1461 input_parameters: CalibrationInput 1462 parameterized_calibration: SpectrometerCalibration 1463 measurement_with_parameterized_calibration: IntensitySpectrum 1464 residuals: np.ndarray 1465 final_calibration: SpectrometerCalibration 1466 1467 def plot(self, plot_xmin=None, plot_xmax=None): 1468 fig, ax = plt.subplots(2, 3, figsize=(16, 9)) 1469 ax[0, 0].plot( 1470 self.measurement.wavelength_nm(), 1471 self.measurement.spectrum, 1472 label="Measurement", 1473 ) 1474 ax[0, 0].plot( 1475 self.reference.wavelength_nm(), self.reference.spectrum, label="Reference" 1476 ) 1477 ax[0, 0].set_xlabel("Wavelength (nm)") 1478 ax[0, 0].set_ylabel("Intensity (Arb. unit)") 1479 ax[0, 0].legend() 1480 initial_guess_calibration = generate_calibration_from_coeffs( 1481 self.input_parameters.get_amplitude_array(), 1482 self.input_parameters.get_wavelength_array(), 1483 self.measurement.wavelength, 1484 ) 1485 initial_guess_spectrum = initial_guess_calibration.apply_to_spectrum( 1486 self.measurement 1487 ) 1488 ax[0, 1].plot( 1489 initial_guess_spectrum.wavelength_nm(), 1490 initial_guess_spectrum.spectrum, 1491 label="Initial guess", 1492 ) 1493 ax[0, 1].plot( 1494 self.reference.wavelength_nm(), self.reference.spectrum, label="Reference" 1495 ) 1496 ax[0, 1].set_ylabel("Intensity (Arb. unit)") 1497 ax[0, 1].set_xlabel("Wavelength (nm)") 1498 ax[0, 1].legend() 1499 1500 ax[0, 2].plot( 1501 self.measurement_with_parameterized_calibration.wavelength_nm(), 1502 self.measurement_with_parameterized_calibration.spectrum, 1503 label="Model calibrated", 1504 ) 1505 ax[0, 2].plot( 1506 self.reference.wavelength_nm(), self.reference.spectrum, label="Reference" 1507 ) 1508 ax[0, 2].set_ylabel("Intensity (Arb. unit)") 1509 ax[0, 2].set_xlabel("Wavelength (nm)") 1510 ax[0, 2].legend() 1511 1512 ax[1, 0].plot( 1513 self.measurement_with_parameterized_calibration.wavelength_nm(), 1514 self.residuals, 1515 label="Residual", 1516 ) 1517 ax[1, 0].set_ylim(-0.1, 0.1) 1518 ax[1, 0].set_ylabel("Intensity (Arb. unit)") 1519 ax[1, 0].set_xlabel("Wavelength (nm)") 1520 ax[1, 0].legend() 1521 1522 second_calibration = self.final_calibration.apply_to_spectrum(self.measurement) 1523 ax[1, 1].plot( 1524 second_calibration.wavelength_nm(), 1525 second_calibration.spectrum, 1526 label="Calibrated", 1527 ) 1528 ax[1, 1].plot( 1529 self.reference.wavelength_nm(), self.reference.spectrum, label="Reference" 1530 ) 1531 ax[1, 1].set_ylabel("Intensity (Arb. unit)") 1532 ax[1, 1].set_xlabel("Wavelength (nm)") 1533 ax[1, 1].legend() 1534 ax[1, 2].plot( 1535 self.final_calibration.corrected_wavelengths * 1e9, 1536 self.final_calibration.intensity_factors, 1537 label="Final weights", 1538 ) 1539 ax[1, 2].set_ylabel("Final weights") 1540 ax[1, 2].set_xlabel("Wavelength (nm)") 1541 ax[1, 2].legend() 1542 1543 if plot_xmin is not None and plot_xmax is not None: 1544 for a in ax: 1545 for b in a: 1546 b.set_xlim(plot_xmin, plot_xmax) 1547 1548 return fig 1549 1550 @staticmethod 1551 def generate( 1552 measurement: IntensitySpectrum, 1553 reference: IntensitySpectrum, 1554 input_parameters: CalibrationInput, 1555 ): 1556 new_guess = fit_calibration_amplitude_model( 1557 measurement=measurement, 1558 reference=reference, 1559 wavelength_coeffs=input_parameters.get_wavelength_array(), 1560 amplitude_guess=input_parameters.get_amplitude_array(), 1561 roi=np.array([input_parameters.roi_lowest, input_parameters.roi_highest]), 1562 ) 1563 1564 parameterized_calibration = generate_calibration_from_coeffs( 1565 new_guess, input_parameters.get_wavelength_array(), measurement.wavelength 1566 ) 1567 measurement_with_parameterized_calibration = ( 1568 parameterized_calibration.apply_to_spectrum(measurement) 1569 ) 1570 1571 ref_on_new_axis = reference.to_interpolated_wavelength( 1572 measurement_with_parameterized_calibration.wavelength 1573 ) 1574 measurement_shifted = measurement.to_corrected_wavelength( 1575 measurement_with_parameterized_calibration.wavelength 1576 ) 1577 residuals = ( 1578 measurement_with_parameterized_calibration.spectrum 1579 - ref_on_new_axis.spectrum 1580 ) 1581 1582 new_weights = ( 1583 parameterized_calibration.intensity_factors 1584 - measurement_shifted.spectrum 1585 * residuals 1586 / (measurement_shifted.spectrum**2 + input_parameters.noise_level) 1587 ) 1588 1589 final_calibration = SpectrometerCalibration( 1590 intensity_factors=new_weights, 1591 original_wavelengths=measurement.wavelength, 1592 corrected_frequencies=parameterized_calibration.corrected_frequencies, 1593 corrected_wavelengths=parameterized_calibration.corrected_wavelengths, 1594 ) 1595 1596 return CalibrationDataset( 1597 measurement=measurement, 1598 reference=reference, 1599 input_parameters=input_parameters, 1600 parameterized_calibration=parameterized_calibration, 1601 measurement_with_parameterized_calibration=measurement_with_parameterized_calibration, 1602 residuals=residuals, 1603 final_calibration=final_calibration, 1604 )
1467 def plot(self, plot_xmin=None, plot_xmax=None): 1468 fig, ax = plt.subplots(2, 3, figsize=(16, 9)) 1469 ax[0, 0].plot( 1470 self.measurement.wavelength_nm(), 1471 self.measurement.spectrum, 1472 label="Measurement", 1473 ) 1474 ax[0, 0].plot( 1475 self.reference.wavelength_nm(), self.reference.spectrum, label="Reference" 1476 ) 1477 ax[0, 0].set_xlabel("Wavelength (nm)") 1478 ax[0, 0].set_ylabel("Intensity (Arb. unit)") 1479 ax[0, 0].legend() 1480 initial_guess_calibration = generate_calibration_from_coeffs( 1481 self.input_parameters.get_amplitude_array(), 1482 self.input_parameters.get_wavelength_array(), 1483 self.measurement.wavelength, 1484 ) 1485 initial_guess_spectrum = initial_guess_calibration.apply_to_spectrum( 1486 self.measurement 1487 ) 1488 ax[0, 1].plot( 1489 initial_guess_spectrum.wavelength_nm(), 1490 initial_guess_spectrum.spectrum, 1491 label="Initial guess", 1492 ) 1493 ax[0, 1].plot( 1494 self.reference.wavelength_nm(), self.reference.spectrum, label="Reference" 1495 ) 1496 ax[0, 1].set_ylabel("Intensity (Arb. unit)") 1497 ax[0, 1].set_xlabel("Wavelength (nm)") 1498 ax[0, 1].legend() 1499 1500 ax[0, 2].plot( 1501 self.measurement_with_parameterized_calibration.wavelength_nm(), 1502 self.measurement_with_parameterized_calibration.spectrum, 1503 label="Model calibrated", 1504 ) 1505 ax[0, 2].plot( 1506 self.reference.wavelength_nm(), self.reference.spectrum, label="Reference" 1507 ) 1508 ax[0, 2].set_ylabel("Intensity (Arb. unit)") 1509 ax[0, 2].set_xlabel("Wavelength (nm)") 1510 ax[0, 2].legend() 1511 1512 ax[1, 0].plot( 1513 self.measurement_with_parameterized_calibration.wavelength_nm(), 1514 self.residuals, 1515 label="Residual", 1516 ) 1517 ax[1, 0].set_ylim(-0.1, 0.1) 1518 ax[1, 0].set_ylabel("Intensity (Arb. unit)") 1519 ax[1, 0].set_xlabel("Wavelength (nm)") 1520 ax[1, 0].legend() 1521 1522 second_calibration = self.final_calibration.apply_to_spectrum(self.measurement) 1523 ax[1, 1].plot( 1524 second_calibration.wavelength_nm(), 1525 second_calibration.spectrum, 1526 label="Calibrated", 1527 ) 1528 ax[1, 1].plot( 1529 self.reference.wavelength_nm(), self.reference.spectrum, label="Reference" 1530 ) 1531 ax[1, 1].set_ylabel("Intensity (Arb. unit)") 1532 ax[1, 1].set_xlabel("Wavelength (nm)") 1533 ax[1, 1].legend() 1534 ax[1, 2].plot( 1535 self.final_calibration.corrected_wavelengths * 1e9, 1536 self.final_calibration.intensity_factors, 1537 label="Final weights", 1538 ) 1539 ax[1, 2].set_ylabel("Final weights") 1540 ax[1, 2].set_xlabel("Wavelength (nm)") 1541 ax[1, 2].legend() 1542 1543 if plot_xmin is not None and plot_xmax is not None: 1544 for a in ax: 1545 for b in a: 1546 b.set_xlim(plot_xmin, plot_xmax) 1547 1548 return fig
1550 @staticmethod 1551 def generate( 1552 measurement: IntensitySpectrum, 1553 reference: IntensitySpectrum, 1554 input_parameters: CalibrationInput, 1555 ): 1556 new_guess = fit_calibration_amplitude_model( 1557 measurement=measurement, 1558 reference=reference, 1559 wavelength_coeffs=input_parameters.get_wavelength_array(), 1560 amplitude_guess=input_parameters.get_amplitude_array(), 1561 roi=np.array([input_parameters.roi_lowest, input_parameters.roi_highest]), 1562 ) 1563 1564 parameterized_calibration = generate_calibration_from_coeffs( 1565 new_guess, input_parameters.get_wavelength_array(), measurement.wavelength 1566 ) 1567 measurement_with_parameterized_calibration = ( 1568 parameterized_calibration.apply_to_spectrum(measurement) 1569 ) 1570 1571 ref_on_new_axis = reference.to_interpolated_wavelength( 1572 measurement_with_parameterized_calibration.wavelength 1573 ) 1574 measurement_shifted = measurement.to_corrected_wavelength( 1575 measurement_with_parameterized_calibration.wavelength 1576 ) 1577 residuals = ( 1578 measurement_with_parameterized_calibration.spectrum 1579 - ref_on_new_axis.spectrum 1580 ) 1581 1582 new_weights = ( 1583 parameterized_calibration.intensity_factors 1584 - measurement_shifted.spectrum 1585 * residuals 1586 / (measurement_shifted.spectrum**2 + input_parameters.noise_level) 1587 ) 1588 1589 final_calibration = SpectrometerCalibration( 1590 intensity_factors=new_weights, 1591 original_wavelengths=measurement.wavelength, 1592 corrected_frequencies=parameterized_calibration.corrected_frequencies, 1593 corrected_wavelengths=parameterized_calibration.corrected_wavelengths, 1594 ) 1595 1596 return CalibrationDataset( 1597 measurement=measurement, 1598 reference=reference, 1599 input_parameters=input_parameters, 1600 parameterized_calibration=parameterized_calibration, 1601 measurement_with_parameterized_calibration=measurement_with_parameterized_calibration, 1602 residuals=residuals, 1603 final_calibration=final_calibration, 1604 )
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file
1320@yaml_io 1321@dataclass 1322class CalibrationInput: 1323 wavelength_center: float 1324 wavelength_offset: float 1325 wavelength_slope: float 1326 amplitude_center: float 1327 amplitude_multiplier: float 1328 amplitude_slope: float 1329 amplitude_width: float 1330 amplitude_order: float 1331 roi_lowest: float 1332 roi_highest: float 1333 noise_level: float 1334 1335 def get_wavelength_array(self): 1336 return np.array( 1337 [self.wavelength_center, self.wavelength_offset, self.wavelength_slope] 1338 ) 1339 1340 def get_amplitude_array(self): 1341 return np.array( 1342 [ 1343 self.amplitude_center, 1344 self.amplitude_multiplier, 1345 self.amplitude_slope, 1346 self.amplitude_width, 1347 self.amplitude_order, 1348 ] 1349 ) 1350 1351 def plot( 1352 self, 1353 measurement: IntensitySpectrum, 1354 reference: IntensitySpectrum, 1355 plot_xmin=None, 1356 plot_xmax=None, 1357 ): 1358 fig, ax = plt.subplots(1, 2, figsize=(8, 4)) 1359 ax[0].plot( 1360 measurement.wavelength_nm(), measurement.spectrum, label="Measurement" 1361 ) 1362 ax[0].plot(reference.wavelength_nm(), reference.spectrum, label="Reference") 1363 ax[0].set_xlabel("Wavelength (nm)") 1364 ax[0].set_ylabel("Intensity (Arb. unit)") 1365 ax[0].legend() 1366 initial_guess_calibration = generate_calibration_from_coeffs( 1367 self.get_amplitude_array(), 1368 self.get_wavelength_array(), 1369 measurement.wavelength, 1370 ) 1371 initial_guess_spectrum = initial_guess_calibration.apply_to_spectrum( 1372 measurement 1373 ) 1374 ax[1].plot( 1375 initial_guess_spectrum.wavelength_nm(), 1376 initial_guess_spectrum.spectrum, 1377 label="Initial guess", 1378 ) 1379 ax[1].plot(reference.wavelength_nm(), reference.spectrum, label="Reference") 1380 ax[1].set_ylabel("Intensity (Arb. unit)") 1381 ax[1].set_xlabel("Wavelength (nm)") 1382 ax[1].legend() 1383 1384 if plot_xmin is not None and plot_xmax is not None: 1385 for a in ax: 1386 a.set_xlim(plot_xmin, plot_xmax) 1387 1388 return fig
1351 def plot( 1352 self, 1353 measurement: IntensitySpectrum, 1354 reference: IntensitySpectrum, 1355 plot_xmin=None, 1356 plot_xmax=None, 1357 ): 1358 fig, ax = plt.subplots(1, 2, figsize=(8, 4)) 1359 ax[0].plot( 1360 measurement.wavelength_nm(), measurement.spectrum, label="Measurement" 1361 ) 1362 ax[0].plot(reference.wavelength_nm(), reference.spectrum, label="Reference") 1363 ax[0].set_xlabel("Wavelength (nm)") 1364 ax[0].set_ylabel("Intensity (Arb. unit)") 1365 ax[0].legend() 1366 initial_guess_calibration = generate_calibration_from_coeffs( 1367 self.get_amplitude_array(), 1368 self.get_wavelength_array(), 1369 measurement.wavelength, 1370 ) 1371 initial_guess_spectrum = initial_guess_calibration.apply_to_spectrum( 1372 measurement 1373 ) 1374 ax[1].plot( 1375 initial_guess_spectrum.wavelength_nm(), 1376 initial_guess_spectrum.spectrum, 1377 label="Initial guess", 1378 ) 1379 ax[1].plot(reference.wavelength_nm(), reference.spectrum, label="Reference") 1380 ax[1].set_ylabel("Intensity (Arb. unit)") 1381 ax[1].set_xlabel("Wavelength (nm)") 1382 ax[1].legend() 1383 1384 if plot_xmin is not None and plot_xmax is not None: 1385 for a in ax: 1386 a.set_xlim(plot_xmin, plot_xmax) 1387 1388 return fig
28 def from_dict(cls, data: dict): 29 """ 30 Takes a dict and makes an instance of the class 31 32 Args: 33 data (dict): the result of a call of .to_dict on the class 34 """ 35 36 def handle_complex_array(serialized_array) -> np.ndarray: 37 """Helper function to deserialize numpy arrays, handling complex types""" 38 if isinstance(serialized_array, list) and all( 39 isinstance(item, dict) and "re" in item and "im" in item 40 for item in serialized_array 41 ): 42 return np.array( 43 [complex(item["re"], item["im"]) for item in serialized_array], 44 dtype=np.complex128, 45 ) 46 return np.array(serialized_array) 47 48 loaded_data = {} 49 for field_name, field_type in cls.__annotations__.items(): 50 if field_type is np.ndarray: 51 loaded_data[field_name] = handle_complex_array(data[field_name]) 52 elif is_dataclass(field_type): 53 loaded_data[field_name] = field_type.from_dict(data[field_name]) 54 else: 55 loaded_data[field_name] = data[field_name] 56 return cls(**loaded_data)
Takes a dict and makes an instance of the class
Arguments:
- data (dict): the result of a call of .to_dict on the class
58 def load_yaml(cls, filename: str): 59 """ 60 load from a yaml file 61 62 Args: 63 filename (str): path to the file 64 """ 65 with open(filename, "r") as file: 66 data = yaml.load(file, yaml.SafeLoader) 67 return cls.from_dict(data)
load from a yaml file
Arguments:
- filename (str): path to the file
80 def to_dict(instance): 81 """ 82 serialize the class into a dict 83 """ 84 data_dict = {} 85 for field_name, field_type in instance.__annotations__.items(): 86 field_value = getattr(instance, field_name) 87 if field_type is np.ndarray: 88 if field_value.dtype == np.complex128: 89 data_dict[field_name] = [ 90 {"re": num.real, "im": num.imag} for num in field_value.tolist() 91 ] 92 else: 93 data_dict[field_name] = field_value.tolist() 94 elif is_dataclass(field_type): 95 data_dict[field_name] = field_value.to_dict() 96 elif field_type is np.float64 or field_type is float: 97 data_dict[field_name] = float(field_value) 98 else: 99 data_dict[field_name] = field_value 100 return data_dict
serialize the class into a dict
69 def save_yaml(instance, filename: str): 70 """ 71 save to a yaml file 72 73 Args: 74 filename (str): path to the file 75 """ 76 data_dict = instance.to_dict() 77 with open(filename, "w") as file: 78 yaml.dump(data_dict, file)
save to a yaml file
Arguments:
- filename (str): path to the file