Coverage for polars_analysis / plotting / pulse_plotting.py: 88%

547 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-03-13 13:37 -0400

1import logging 

2import pathlib 

3from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, cast 

4 

5import matplotlib.pyplot as plt 

6import numpy as np 

7import polars as pl 

8import scipy 

9from diptest import diptest # type: ignore 

10from scipy.optimize import curve_fit 

11 

12from polars_analysis.analysis import constants 

13from polars_analysis.plotting import helper 

14 

15if TYPE_CHECKING: 

16 from matplotlib.container import BarContainer 

17 from matplotlib.patches import Polygon 

18 

19# Instantiate logger 

20log = logging.getLogger(__name__) 

21 

22""" 

23Functions for plotting pulse run data, which has already been processed. 

24""" 

25 

26 

27def plot_pulse_overlay_all( 

28 df: pl.DataFrame, 

29 channel: int, 

30 plot_dir: pathlib.Path, 

31) -> None: 

32 """ 

33 Plots all pulse overlays for a given channel. 

34 

35 Args: 

36 df: Dataframe. Needs columns 'samples_interleaved, 

37 'gain', 'board_id', 'run_number', 'att_val', and 'channel'. 

38 channel: Channel to plot 

39 plot_dir: Where to save plots to 

40 

41 Returns: 

42 None 

43 """ 

44 plot_pulse_overlay(df, channel, "lo", plot_dir) 

45 plot_pulse_overlay(df, channel, "hi", plot_dir) 

46 plot_pulse_overlay(df, channel, "lo", plot_dir, norm=True) 

47 plot_pulse_overlay(df, channel, "hi", plot_dir, norm=True) 

48 

49 plot_pulse_overlay(df, channel, "lo", plot_dir, deriv=True) 

50 plot_pulse_overlay(df, channel, "hi", plot_dir, deriv=True) 

51 plot_pulse_overlay(df, channel, "lo", plot_dir, deriv=True, norm=True) 

52 plot_pulse_overlay(df, channel, "hi", plot_dir, deriv=True, norm=True) 

53 

54 plot_pulse_overlay(df, channel, "lo", plot_dir, grad=True) 

55 plot_pulse_overlay(df, channel, "hi", plot_dir, grad=True) 

56 plot_pulse_overlay(df, channel, "lo", plot_dir, grad=True, norm=True) 

57 plot_pulse_overlay(df, channel, "hi", plot_dir, grad=True, norm=True) 

58 

59 

60def plot_pulse_overlay_bs(df: pl.DataFrame, channel: int, plot_dir: pathlib.Path, gain: Literal["hi", "lo"]) -> None: 

61 """ 

62 Plots all pulse overlays for a given channel. 

63 

64 Args: 

65 df: Dataframe. Needs columns 'samples_interleaved, 

66 'gain', 'board_id', 'run_number', 'att_val', and 'channel'. 

67 channel: Channel to plot 

68 plot_dir: Where to save plots to 

69 

70 Returns: 

71 None 

72 """ 

73 plot_pulse_overlay(df, channel, gain, plot_dir) 

74 plot_pulse_overlay(df, channel, gain, plot_dir, norm=True) 

75 

76 plot_pulse_overlay(df, channel, gain, plot_dir, deriv=True, norm=True) 

77 plot_pulse_overlay(df, channel, gain, plot_dir, grad=True, norm=True) 

78 

79 

80def plot_pulse_overlay( 

81 df: pl.DataFrame, 

82 channel: int, 

83 gain: Literal["hi", "lo"], 

84 plot_dir: pathlib.Path, 

85 norm: bool = False, 

86 deriv: bool = False, 

87 grad: bool = False, 

88) -> None: 

89 """ 

90 Plot the interleaved signals for a given channel and gain. 

91 

92 Args: 

93 df: Dataframe. Needs columns 'samples_interleaved, 

94 'gain', 'board_id', 'run_number', 'att_val', and 'channel'. 

95 channel: Channel to plot 

96 plot_dir: Where to save plots to 

97 

98 Returns: 

99 None 

100 """ 

101 filtered_df: pl.DataFrame = df.filter(pl.col("gain") == gain, pl.col("channel") == channel) 

102 board_id: str = filtered_df[0]["board_id"][0] 

103 run_num: int = int(filtered_df[0]["run_number"][0]) 

104 atten_val: List[float] = filtered_df["att_val"].unique().sort().to_list() 

105 pas_mode = filtered_df[0]["pas_mode"][0] 

106 

107 rows: List[Dict[str, Any]] = [row for row in filtered_df.iter_rows(named=True)] 

108 rows.reverse() 

109 index = 0 

110 for row in rows: 

111 fine_pulse: np.ndarray = np.mean(row["samples_interleaved"], axis=0) 

112 if norm: 

113 fine_pulse /= max(fine_pulse) 

114 if deriv: 

115 fine_pulse = np.diff(fine_pulse, prepend=0) 

116 if grad: 

117 fine_pulse = np.gradient(fine_pulse) 

118 

119 # Make sure plotting window falls within bounds of array 

120 argmax_minus200 = np.argmax(fine_pulse) - 200 

121 if argmax_minus200 < 0: 

122 fine_pulse = np.concatenate([fine_pulse[argmax_minus200:], fine_pulse[:argmax_minus200]]) 

123 argmax_plus1000 = np.argmax(fine_pulse) + 1000 

124 if argmax_plus1000 >= np.size(fine_pulse): 

125 idx = argmax_plus1000 % (constants.PULSES_PER_TRAIN * constants.SAMPLES_PER_PULSE) 

126 fine_pulse = np.concatenate([fine_pulse[idx:], fine_pulse[:idx]]) 

127 

128 plt.plot( 

129 fine_pulse[np.argmax(fine_pulse) - 200 : np.argmax(fine_pulse) + 1000], 

130 label=f"{row['amp']:.3g} mA", 

131 color=helper.jet(index / len(filtered_df)), 

132 ) 

133 index += 1 

134 

135 plt.grid() 

136 plt.xlabel("Time [ns]") 

137 plt.ylabel("Normalized Amplitude" if norm else "Amplitude (ADC counts)") 

138 

139 name = "Pulse" + (" derivative" if deriv else "") + (" gradient" if grad else "") + (" normalized" if norm else "") 

140 plt.title( 

141 helper.plot_summary_string( 

142 name=name, 

143 board_id=board_id, 

144 run_numbers=run_num, 

145 channels=channel, 

146 pas_mode=pas_mode, 

147 gain=gain, 

148 attenuation=atten_val, 

149 ) 

150 ) 

151 saveName = f"pulse_overlay_{gain}_ch{channel}" # +summary_plot_string.split(" ")[-1] 

152 if deriv: 

153 saveName += "_deriv" 

154 if grad: 

155 saveName += "_grad" 

156 if norm: 

157 saveName += "_norm" 

158 plt.legend(loc="upper right", ncol=2) 

159 plt.tight_layout() 

160 plt.savefig(f"{plot_dir}/{saveName}.png") 

161 plt.cla() 

162 plt.clf() 

163 plt.close() 

164 return 

165 

166 

167def power_of_10_range(numbers: np.ndarray) -> tuple[float, float]: 

168 numbers = numbers[~np.isnan(numbers)] # drop NaNs 

169 powers = np.log10(np.abs(numbers[numbers != 0])) 

170 min_power = np.floor(powers.min()) 

171 max_power = np.ceil(powers.max()) 

172 

173 if min_power == max_power: 

174 min_power -= 1 

175 max_power += 1 

176 

177 return (10**min_power, 10**max_power) 

178 

179 

180def plot_energy_resolution( 

181 df: pl.DataFrame, 

182 channel: int, 

183 plot_dir: pathlib.Path, 

184 plot_log_scale: bool = False, 

185) -> None: 

186 """ 

187 Args: 

188 df: Dataframe. Needs columns 'energy_mean, 

189 'energy_std', 'samples_interleaved', 'amp', 

190 'gain', 'board_id', 'run_number', 'att_val', and 'channel'. 

191 plot_dir: Where to save plots to 

192 plot_log_scale: Plot using log scale. Default is False 

193 

194 Returns: 

195 None 

196 """ 

197 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

198 gain: Literal["hi", "lo"] = df["gain"][0] 

199 pas_mode = df[0]["pas_mode"][0] 

200 

201 energies_lf = df.lazy().select("energies", "amp", "awg_amp").explode("energies") 

202 if df["energies"][0].dtype.is_nested(): 

203 energies_lf = energies_lf.explode("energies") 

204 energies_df = ( 

205 energies_lf.group_by("amp", "awg_amp") 

206 .agg( 

207 energy_mean=pl.col("energies").mean(), 

208 energy_std=pl.col("energies").std(), 

209 n_energies=pl.col("energies").len(), 

210 ) 

211 .sort(by="amp") 

212 .collect() 

213 ) 

214 

215 e_arr: np.ndarray = energies_df["energy_mean"].to_numpy(writable=True) 

216 if (e_arr < 0).any(): 

217 # TODO proper solution upstream? 

218 log.warning("Negative energies! Setting to 0.01") 

219 e_arr[e_arr < 0] = 0.01 

220 

221 dE_arr: np.ndarray = energies_df["energy_std"].to_numpy() 

222 amps_arr: np.ndarray = energies_df["amp"].to_numpy() 

223 n_energies: np.ndarray = energies_df["n_energies"].to_numpy() 

224 

225 _, ax = plt.subplots() 

226 ax.grid(True) 

227 

228 d_dE_arr: np.ndarray = dE_arr / np.sqrt(n_energies) 

229 y_err: np.ndarray = (dE_arr / e_arr) * np.sqrt((d_dE_arr / dE_arr) ** 2 + (dE_arr / e_arr) ** 2) 

230 if (y_err < 0).any(): 

231 log.warning("Negative y_err values! Setting to abs(y_err)") 

232 y_err = np.abs(y_err) 

233 

234 ax.errorbar( 

235 amps_arr, 

236 dE_arr / e_arr, 

237 yerr=y_err, 

238 fmt="{c}o".format(c="r"), 

239 markersize=4, 

240 capsize=2, 

241 label=f"{gain} ch{channel} {atten_val}dB", 

242 ) 

243 

244 try: 

245 A, dA, tau, dTau, C, dC = helper.fit_exp_decay(amps_arr, dE_arr / e_arr) 

246 exp_x = np.arange(min(amps_arr), max(amps_arr), (max(amps_arr) - min(amps_arr)) / 100) 

247 exp_fit = helper.exp_decay(exp_x, A, tau, C) 

248 

249 ax.plot( 

250 exp_x, 

251 exp_fit, 

252 label=rf"""Fit: A = {A:.03f}$\pm${dA:.03f} 

253$\tau$ = {tau:.03f}$\pm${dTau:.03f} 

254C = {C:.03f}$\pm${dC:.03f},""", 

255 ) 

256 except ValueError: 

257 log.warning("Cannot calculate exponentional fit for energy resolution, skipping plotting fit line") 

258 

259 ax.set_xlabel("Input Current [mA]") 

260 ax.set_ylabel(r"$\sigma_{E} / E$") 

261 

262 if plot_log_scale: 

263 filename = f"energy_res_log_{gain}_ch{channel}.png" 

264 ax.set_yscale("log") 

265 ax.set_xscale("log") 

266 ax.set_xlim(power_of_10_range(amps_arr)) 

267 ax.set_ylim(power_of_10_range(dE_arr / e_arr)) 

268 else: 

269 filename = f"energy_res_{gain}_ch{channel}.png" 

270 

271 ax.set_title( 

272 helper.plot_summary_string( 

273 name="Energy Resolution", 

274 board_id=df["board_id"][0], 

275 run_numbers=df["run_number"][0], 

276 channels=channel, 

277 attenuation=atten_val, 

278 pas_mode=pas_mode, 

279 gain=gain, 

280 ) 

281 ) 

282 plt.legend(loc="upper right") 

283 

284 plt.savefig(f"{plot_dir}/{filename}") 

285 

286 plt.close() 

287 plt.cla() 

288 plt.clf() 

289 return 

290 

291 

292def plot_sigma_e( 

293 df: pl.DataFrame, 

294 channel: int, 

295 plot_dir: pathlib.Path, 

296) -> None: 

297 """ 

298 Args: 

299 df: Dataframe. Needs columns 'energy_std, 

300 'amp', 'gain', 'board_id', 

301 'run_number', and 'att_val'. 

302 plot_dir: Where to save plots to 

303 

304 Returns: 

305 None 

306 """ 

307 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

308 gain: Literal["hi", "lo"] = df["gain"][0] 

309 run_num: int = df["run_number"][0] 

310 pas_mode = df[0]["pas_mode"][0] 

311 

312 energies_lf = df.lazy().select("energies", "amp").explode("energies") 

313 if df["energies"][0].dtype.is_nested(): 

314 energies_lf = energies_lf.explode("energies") 

315 energies_df = ( 

316 energies_lf.group_by("amp") 

317 .agg( 

318 energy_std=pl.col("energies").std(), 

319 n_energies=pl.col("energies").len(), 

320 ) 

321 .sort(by="amp") 

322 .collect() 

323 ) 

324 dE_arr: np.ndarray = energies_df["energy_std"].to_numpy() 

325 n_energies: np.ndarray = energies_df["n_energies"].to_numpy() 

326 d_dE_arr: np.ndarray = dE_arr / np.sqrt(2 * n_energies - 2) 

327 amps_arr: np.ndarray = energies_df["amp"].to_numpy() 

328 

329 _, ax = plt.subplots() 

330 ax.grid(True) 

331 ax.errorbar( 

332 amps_arr, 

333 dE_arr, 

334 yerr=d_dE_arr, 

335 fmt="{c}o".format(c="r"), 

336 markersize=4, 

337 capsize=2, 

338 ) 

339 ax.set_title( 

340 helper.plot_summary_string( 

341 name="Energy Resolution", 

342 board_id=df["board_id"][0], 

343 run_numbers=run_num, 

344 channels=channel, 

345 attenuation=atten_val, 

346 pas_mode=pas_mode, 

347 gain=gain, 

348 ) 

349 ) 

350 ax.set_xlabel("Input Current [mA]") 

351 ax.set_ylabel(r"$\sigma_{E}$") 

352 

353 plt.savefig(plot_dir / f"sigma_e_{gain}_ch{channel}.png") 

354 plt.close() 

355 plt.cla() 

356 plt.clf() 

357 return 

358 

359 

360def plot_sigma_T( 

361 df: pl.DataFrame, 

362 channel: int, 

363 plot_dir: pathlib.Path, 

364 plot_log_scale: bool = False, 

365) -> None: 

366 """ 

367 Args: 

368 df: Dataframe. Needs columns 'time_std, 

369 'amp', 'gain', 'board_id', 

370 'run_number', 'att_val', and 'channel'. 

371 plot_dir: Where to save plots to 

372 plot_log_scale: Plot using log scale. Default is False 

373 

374 Returns: 

375 None 

376 """ 

377 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

378 gain: Literal["hi", "lo"] = df["gain"][0] 

379 pas_mode = df[0]["pas_mode"][0] 

380 run_num: int = df["run_number"][0] 

381 

382 times_lf = df.lazy().select("times", "amp", "awg_amp") 

383 if df["times"][0].dtype.is_nested(): 

384 times_lf = times_lf.explode("times") 

385 times_df = ( 

386 # Mean subtraction is off by default, but can be used if pulses have different means 

387 # times_df.with_columns(pl.col("times") - pl.col("times").list.mean()) 

388 times_lf.explode("times") 

389 .group_by(["amp", "awg_amp"]) 

390 .agg( 

391 time_std=pl.col("times").std(), 

392 n_times=pl.col("times").len(), 

393 ) 

394 .sort(by="amp") 

395 .collect() 

396 ) 

397 

398 dT_arr: np.ndarray = times_df["time_std"].to_numpy() 

399 amps_arr: np.ndarray = times_df["amp"].to_numpy() 

400 n_times: np.ndarray = times_df["n_times"].to_numpy() 

401 

402 _, ax = plt.subplots() 

403 ax.grid(True) 

404 

405 d_dT_arr: np.ndarray = dT_arr / np.sqrt(2 * n_times - 2) 

406 

407 ax.errorbar( 

408 amps_arr, 

409 dT_arr, 

410 yerr=d_dT_arr, 

411 fmt="{c}o".format(c="r"), 

412 markersize=4, 

413 capsize=2, 

414 label=f"{gain} ch{channel} {atten_val}dB", 

415 ) 

416 

417 try: 

418 A, dA, tau, dTau, C, dC = helper.fit_exp_decay(amps_arr, dT_arr) 

419 

420 exp_x = np.arange(min(amps_arr), max(amps_arr), (max(amps_arr) - min(amps_arr)) / 100) 

421 exp_fit = helper.exp_decay(exp_x, A, tau, C) 

422 

423 ax.plot( 

424 exp_x, 

425 exp_fit, 

426 label=rf"""Fit: A = {A:.03g}$\pm${dA:.03g} 

427$\tau$ = {tau:.03g}$\pm${dTau:.03g} 

428C = {C:.03g}$\pm${dC:.03g}""", 

429 ) 

430 except ValueError: 

431 log.warning("Cannot calculate exponentional fit for timing resolution, skipping plotting fit line") 

432 

433 ax.set_xlabel("Input Current [mA]") 

434 ax.set_ylabel(r"$\sigma_{t}$ [ns]") 

435 

436 if plot_log_scale: 

437 filename = f"timing_res_log_{gain}_ch{channel}.png" 

438 ax.set_yscale("log") 

439 ax.set_xscale("log") 

440 ax.set_xlim(power_of_10_range(amps_arr)) 

441 ax.set_ylim(power_of_10_range(dT_arr)) 

442 else: 

443 filename = f"timing_res_{gain}_ch{channel}.png" 

444 

445 ax.set_title( 

446 helper.plot_summary_string( 

447 name="Timing Resolution", 

448 board_id=df["board_id"][0], 

449 run_numbers=run_num, 

450 channels=channel, 

451 attenuation=atten_val, 

452 pas_mode=pas_mode, 

453 gain=gain, 

454 ) 

455 ) 

456 plt.legend(loc="upper right") 

457 

458 plt.savefig(f"{plot_dir}/{filename}") 

459 

460 plt.close() 

461 plt.cla() 

462 plt.clf() 

463 return 

464 

465 

466def plot_timing_mean( 

467 df: pl.DataFrame, 

468 channel: int, 

469 plot_dir: pathlib.Path, 

470) -> None: 

471 """ 

472 Args: 

473 df: Dataframe. Needs columns 'time_mean', 

474 'time_std', 'amp', 'gain', 'board_id', 

475 'run_number', 'att_val', and 'channel'. 

476 plot_dir: Where to save plots to 

477 

478 Returns: 

479 None 

480 """ 

481 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

482 gain: Literal["hi", "lo"] = df["gain"][0] 

483 run_num: int = int(df[0]["run_number"][0]) 

484 pas_mode = df[0]["pas_mode"][0] 

485 

486 times_lf = df.lazy().select("times", "amp", "awg_amp").explode("times") 

487 if df["times"][0].dtype.is_nested(): 

488 times_lf = times_lf.explode("times") 

489 times_df = ( 

490 times_lf.group_by(["amp", "awg_amp"]) 

491 .agg( 

492 time_mean=pl.col("times").mean(), 

493 time_std=pl.col("times").std(), 

494 n_times=pl.col("times").len(), 

495 ) 

496 .sort(by="amp") 

497 .collect() 

498 ) 

499 

500 t_arr: np.ndarray = times_df["time_mean"].to_numpy() 

501 dT_arr: np.ndarray = times_df["time_std"].to_numpy() 

502 amps_arr: np.ndarray = times_df["amp"].to_numpy() 

503 n_times: np.ndarray = times_df["n_times"].to_numpy() 

504 

505 _, ax = plt.subplots() 

506 ax.grid(True) 

507 

508 ax.errorbar( 

509 amps_arr, 

510 t_arr, 

511 yerr=dT_arr / np.sqrt(n_times), 

512 fmt="{c}o".format(c="r"), 

513 markersize=4, 

514 capsize=2, 

515 label=f"{gain} ch{channel} {atten_val}dB", 

516 ) 

517 

518 ax.set_xlabel("Input Current [mA]") 

519 ax.set_ylabel(r"$t$ [ns]") 

520 

521 ax.set_title( 

522 helper.plot_summary_string( 

523 name="Timing Mean", 

524 board_id=df["board_id"][0], 

525 run_numbers=run_num, 

526 channels=channel, 

527 attenuation=atten_val, 

528 pas_mode=pas_mode, 

529 gain=gain, 

530 ) 

531 ) 

532 plt.legend(loc="upper right") 

533 

534 filename: str = f"timing_mean_{gain}_ch{channel}.png" 

535 plt.savefig(plot_dir / filename) 

536 

537 plt.close() 

538 plt.cla() 

539 plt.clf() 

540 

541 return 

542 

543 

544def plot_risetime( 

545 df: pl.DataFrame, 

546 channel: int, 

547 plot_dir: pathlib.Path, 

548) -> None: 

549 """ 

550 Args: 

551 df: Dataframe. Needs columns 'rise_time', 

552 'amp', 'gain', 'board_id', 

553 'run_number', 'att_val', and 'channel'. 

554 plot_dir: Where to save plots to 

555 

556 Returns: 

557 None 

558 """ 

559 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

560 gain: Literal["hi", "lo"] = df["gain"][0] 

561 run_num: int = int(df[0]["run_number"][0]) 

562 pas_mode = df[0]["pas_mode"][0] 

563 

564 risetimes_df = ( 

565 df.lazy() 

566 .select("rise_time", "rise_time_error", "amp", "awg_amp") 

567 .group_by(["amp", "awg_amp"]) 

568 .agg([pl.col("rise_time").mean().alias("rise_time"), pl.col("rise_time_error").mean().alias("rise_time_error")]) 

569 .sort(by="amp") 

570 .collect() 

571 ) 

572 risetimes: np.ndarray = risetimes_df["rise_time"].to_numpy() 

573 amps_arr: np.ndarray = risetimes_df["amp"].to_numpy() 

574 d_risetimes: np.ndarray = risetimes_df["rise_time_error"].to_numpy() 

575 

576 _, ax = plt.subplots() 

577 

578 ax.grid(zorder=0) 

579 ax.set_title( 

580 helper.plot_summary_string( 

581 name="Rise times", 

582 board_id=df["board_id"][0], 

583 run_numbers=run_num, 

584 channels=channel, 

585 attenuation=atten_val, 

586 pas_mode=pas_mode, 

587 gain=gain, 

588 ) 

589 ) 

590 ax.set_xlabel("Amplitudes [mA]") 

591 ax.set_ylabel("Risetime [ns]") 

592 # ax.set_ylim(0, 100) 

593 

594 ax.errorbar(amps_arr, risetimes, yerr=d_risetimes, color="black", fmt="o") 

595 plt.savefig(plot_dir / f"risetime_{gain}_ch{channel}_summary.png") 

596 plt.cla() 

597 plt.clf() 

598 return 

599 

600 

601def plot_zero_crossing( 

602 df: pl.DataFrame, 

603 channel: int, 

604 plot_dir: pathlib.Path, 

605) -> None: 

606 """ 

607 Args: 

608 df: Dataframe. Needs columns 'zero_crossing_time', 

609 'amp', 'gain', 'board_id', 

610 'run_number', 'att_val', and 'channel'. 

611 plot_dir: Where to save plots to 

612 

613 Returns: 

614 None 

615 """ 

616 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

617 gain: Literal["hi", "lo"] = df["gain"][0] 

618 run_num: int = int(df[0]["run_number"][0]) 

619 pas_mode = df[0]["pas_mode"][0] 

620 zero_crossing_df = ( 

621 df.lazy() 

622 .select("zero_crossing_time", "zero_crossing_error", "amp", "awg_amp", "att_val") 

623 .group_by("amp", "awg_amp") 

624 .agg( 

625 [ 

626 pl.col("zero_crossing_time").mean().alias("zero_crossing"), 

627 pl.col("zero_crossing_error").mean().alias("zero_crossing_error"), 

628 ] 

629 ) 

630 .sort(by="amp") 

631 .collect() 

632 ) 

633 

634 zero_crossings: np.ndarray = zero_crossing_df["zero_crossing"].to_numpy() 

635 zero_crossing_errors: np.ndarray = zero_crossing_df["zero_crossing_error"].to_numpy() 

636 amps_arr: np.ndarray = zero_crossing_df["amp"].to_numpy() 

637 

638 _, ax = plt.subplots() 

639 

640 ax.grid(zorder=0) 

641 ax.set_title( 

642 helper.plot_summary_string( 

643 name="Zero Crossing Time from Peak", 

644 board_id=df["board_id"][0], 

645 run_numbers=run_num, 

646 channels=channel, 

647 attenuation=atten_val, 

648 pas_mode=pas_mode, 

649 gain=gain, 

650 ) 

651 ) 

652 ax.set_xlabel("Amplitudes [mA]") 

653 ax.set_ylabel("Zero Crossing Time [ns]") 

654 # ax.set_ylim(80, 120) 

655 

656 ax.errorbar(amps_arr, zero_crossings, yerr=zero_crossing_errors, color="black", fmt="o") 

657 

658 plt.savefig(plot_dir / f"zerocrossing_{gain}_ch{channel}_summary.png") 

659 plt.cla() 

660 plt.clf() 

661 return 

662 

663 

664def plot_gain_ratios( 

665 df: pl.DataFrame, 

666 channel: int, 

667 plot_dir: pathlib.Path, 

668) -> None: 

669 """ 

670 Args: 

671 df: Dataframe. Needs columns 'energy_mean', 

672 'amp', 'gain', 'board_id', 

673 'run_number', 'att_val', and 'channel'. 

674 plot_dir: Where to save plots to 

675 

676 Returns: 

677 None 

678 """ 

679 gain_ratios = ( 

680 df.filter(pl.col("gain") == "lo") 

681 .select(["amp", "energy_mean", "awg_amp", "att_val"]) 

682 .join(df.filter(gain="hi")["amp", "energy_mean"], on="amp", suffix="_hi") 

683 .select("amp", (pl.col("energy_mean_hi") / pl.col("energy_mean")).alias("gain_ratio"), "awg_amp", "att_val") 

684 ) 

685 

686 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

687 run_num: int = int(df[0]["run_number"][0]) 

688 pas_mode = df[0]["pas_mode"][0] 

689 

690 plt.scatter(gain_ratios["amp"], gain_ratios["gain_ratio"]) 

691 plt.title( 

692 helper.plot_summary_string( 

693 name="Gain Ratio Summary", 

694 board_id=df["board_id"][0], 

695 run_numbers=run_num, 

696 channels=channel, 

697 attenuation=atten_val, 

698 pas_mode=pas_mode, 

699 ) 

700 ) 

701 plt.grid() 

702 plt.ylabel("Gain Energy Ratio") 

703 plt.xlabel("Amplitudes [mA]") 

704 

705 plt.savefig(plot_dir / f"gain_ch{channel}_overlay_summary.png") 

706 plt.cla() 

707 plt.clf() 

708 return 

709 

710 

711def plot_INL( 

712 df: pl.DataFrame, 

713 channel: int, 

714 plot_dir: pathlib.Path, 

715 skip_last_n: Optional[int] = None, 

716) -> None: 

717 """ 

718 Args: 

719 df: Dataframe. Needs columns 'energy_mean', 

720 'energy_std', 'amp', 'gain', 'board_id', 

721 'run_number', 'att_val', and 'channel'. 

722 plot_dir: Where to save plots to 

723 

724 skip_last_n: skip last n points 

725 

726 Returns: 

727 None 

728 """ 

729 

730 if skip_last_n is not None: 

731 skip_last_n = -skip_last_n 

732 

733 energies_lf = df.lazy().select("energies", "amp").explode("energies") 

734 if df["energies"][0].dtype.is_nested(): 

735 energies_lf = energies_lf.explode("energies") 

736 energies_df = ( 

737 energies_lf.group_by("amp") 

738 .agg( 

739 energy_mean=pl.col("energies").mean(), 

740 energy_std=pl.col("energies").std(), 

741 n_energies=pl.col("energies").len(), 

742 ) 

743 .sort(by="amp") 

744 .collect() 

745 ) 

746 amps_arr: np.ndarray = energies_df["amp"].to_numpy() 

747 n_energies: np.ndarray = energies_df["n_energies"].to_numpy() 

748 e_arr: np.ndarray = energies_df["energy_mean"].to_numpy() 

749 dE_arr: np.ndarray = energies_df["energy_std"].to_numpy() / np.sqrt(n_energies) 

750 

751 if len(e_arr[:skip_last_n]) == 1 or len(dE_arr[:skip_last_n]) == 1 or len(amps_arr[:skip_last_n]) == 1: 

752 # You can't fit a _unique_ line to a single point 

753 return 

754 

755 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

756 gain: Literal["hi", "lo"] = df["gain"][0] 

757 run_num: int = int(df[0]["run_number"][0]) 

758 pas_mode = df[0]["pas_mode"][0] 

759 

760 fig, ax = plt.subplots(1, 2, figsize=(10, 5)) 

761 ax1, ax2 = ax[0], ax[1] 

762 

763 title: Literal["LG", "HG"] = "LG" if gain == "lo" else "HG" 

764 fig.suptitle( 

765 helper.plot_summary_string( 

766 name="Linearity", 

767 board_id=df["board_id"][0], 

768 run_numbers=run_num, 

769 channels=df["channel"][0], 

770 attenuation=atten_val, 

771 gain=gain, 

772 pas_mode=pas_mode, 

773 ) 

774 ) 

775 

776 ax1.set(ylabel="Pulse Height [ADC Counts]", xlabel="Input Current [mA]") 

777 ax1.set_xlim(0, max(amps_arr) + 1) 

778 ax1.set_ylim(0, max(e_arr) + 0.3 * max(e_arr)) 

779 

780 ax2.set(ylabel="INL [%]", xlabel="Input Current [mA]") 

781 ax2.set_xlim(min(amps_arr) / 3.0, max(amps_arr) * 3.0) 

782 ax2.set_xscale("log") 

783 plt.axhline(y=0, color="r", linestyle="-") 

784 

785 ax1.grid() 

786 ax2.grid() 

787 

788 ax1.errorbar(amps_arr, e_arr, yerr=dE_arr, fmt="ko", markersize=4, capsize=2) 

789 

790 popt, pcov = curve_fit( 

791 helper.lin, 

792 amps_arr[:skip_last_n], 

793 e_arr[:skip_last_n], 

794 # amps_arr[amps_arr <= plateau_amp], 

795 # e_arr[amps_arr <= plateau_amp], 

796 p0=[e_arr[1] / amps_arr[1], 0], 

797 # sigma=dE_arr[amps_arr <= plateau_amp], 

798 sigma=dE_arr[:skip_last_n], 

799 absolute_sigma=True, 

800 ) 

801 

802 m, b, dm, db = popt[0], popt[1], pcov[0][0], pcov[1][1] 

803 

804 popt, pcov = curve_fit( 

805 helper.lin, 

806 amps_arr[:skip_last_n], 

807 e_arr[:skip_last_n], 

808 # amps_arr[amps_arr <= plateau_amp], 

809 # e_arr[amps_arr <= plateau_amp], 

810 p0=[e_arr[1] / amps_arr[1], 0], 

811 # sigma=dE_arr[amps_arr <= plateau_amp], 

812 sigma=dE_arr[:skip_last_n], 

813 absolute_sigma=True, 

814 ) 

815 xspace: np.ndarray = np.linspace(0, max(amps_arr) + 0.5 * max(amps_arr), 500) 

816 

817 ax1.plot(xspace, m * xspace + b, "r-", label="Fit") 

818 

819 text: str = f"m = {m:1f} $\\pm$ {dm:.1f}\nb = {b:.1f} $\\pm$ {db:.1f}" 

820 ax1.text( 

821 0.1, 

822 0.9, 

823 text, 

824 horizontalalignment="left", 

825 verticalalignment="top", 

826 transform=ax1.transAxes, 

827 ) 

828 

829 y_pred: np.ndarray = helper.lin(amps_arr, *popt) 

830 

831 INL: np.ndarray = 100 * (e_arr - y_pred) / max(e_arr) 

832 

833 error: np.ndarray = 100 * (max(e_arr)) ** -1 * dE_arr 

834 ax2.errorbar(amps_arr, INL, yerr=error, fmt="ko", markersize=4, capsize=2) 

835 

836 if title == "LG": 

837 ax1.set_xlim(0, max(amps_arr) + 0.5) 

838 ax2.set_ylim(-3, 3) 

839 elif title == "HG": 

840 ax1.set_xlim(0, max(amps_arr) + 0.5) 

841 ax2.set_ylim(-3, 3) 

842 

843 plt.tight_layout() 

844 plt.savefig(plot_dir / f"{title}_ch{channel}_Linearity_plot.png") 

845 plt.close() 

846 plt.cla() 

847 plt.clf() 

848 

849 return 

850 

851 

852def plot_autocorrelation( 

853 df: pl.DataFrame, 

854 channel: int, 

855 plot_dir: pathlib.Path, 

856) -> None: 

857 """ 

858 Args: 

859 df: Dataframe. Needs columns 'autocorr', 'gain', 'board_id', 'run_number', 'att_val', and 'channel'. 

860 plot_dir: Where to save plots to 

861 

862 Returns: 

863 None 

864 """ 

865 autocorr: np.ndarray = df["autocorr"][0].to_numpy() 

866 

867 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

868 gain: Literal["hi", "lo"] = df["gain"][0] 

869 run_num: int = int(df[0]["run_number"][0]) 

870 pas_mode = df[0]["pas_mode"][0] 

871 

872 _, ax = plt.subplots() 

873 ax.grid(True) 

874 norm_ac = autocorr / autocorr[0] 

875 ax.set_title( 

876 helper.plot_summary_string( 

877 name="Autocorrelcation", 

878 board_id=df["board_id"][0], 

879 run_numbers=run_num, 

880 channels=channel, 

881 attenuation=atten_val, 

882 pas_mode=pas_mode, 

883 gain=gain, 

884 ) 

885 ) 

886 ax.set_xlabel("Lag") 

887 ax.set_ylabel("ACF") 

888 

889 ax.plot(norm_ac, "k-") 

890 ax.set_xlim(0, len(norm_ac) - 1) 

891 ax.set_ylim(min(norm_ac) - 0.05, 1.05) 

892 

893 plt.savefig(plot_dir / f"autocorrelation_{gain}_ch{channel}.png") 

894 plt.cla() 

895 plt.clf() 

896 plt.close() 

897 return 

898 

899 

900""" 

901========================================================= 

902 Plots for the low and hi gain webpages 

903========================================================= 

904""" 

905 

906 

907def plot_energy_hist( 

908 df: pl.DataFrame, 

909 channel: int, 

910 plot_dir: pathlib.Path, 

911) -> None: 

912 """ 

913 Args: 

914 df: Dataframe 

915 plot_dir: Where to save plots to 

916 

917 Returns: 

918 None 

919 """ 

920 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

921 gain: Literal["hi", "lo"] = df["gain"][0] 

922 run_num: int = int(df[0]["run_number"][0]) 

923 pas_mode = df[0]["pas_mode"][0] 

924 

925 energies = df.select(pl.col("energies").list.eval(pl.element().explode())).to_numpy()[0][0] 

926 

927 if gain == "hi": 

928 hist_bins = np.linspace( 

929 5 * int((min(energies) - 2.5) / 5), 

930 5 * int((max(energies) + 7.5) / 5), 

931 -int((min(energies) - 2.5) / 5) + int((max(energies) + 7.5) / 5) + 1, 

932 ) 

933 elif gain == "lo": 

934 hist_bins = np.linspace( 

935 1 * int((min(energies) - 0.5) / 1), 

936 1 * int((max(energies) + 1.5) / 1), 

937 -int((min(energies) - 0.5) / 1) + int((max(energies) + 1.5) / 1) + 1, 

938 ) 

939 

940 y, bins, h = plt.hist( 

941 energies, 

942 bins=hist_bins.tolist(), 

943 ) 

944 skew = scipy.stats.skew(energies) 

945 _, dip_pval = diptest(y) # type: ignore 

946 

947 if TYPE_CHECKING: 

948 h = cast(list[BarContainer | Polygon], h) 

949 h[0].set_label( 

950 f"Samples ({len(energies)}):\nMean = {np.round(np.mean(energies), 3)}\ 

951 \nRMS = {np.round(np.std(energies), 3)}, \nγ={skew:.01f}, dip={dip_pval:.02f}" 

952 ) 

953 

954 xaxis: np.ndarray = np.linspace(np.min(energies), np.max(energies), 1000) 

955 fit_pars = helper.calc_gaussian(energies, bins) 

956 fit_mu, fit_sigma, fit_N = fit_pars[0], fit_pars[2], fit_pars[4] 

957 d_mu, d_sigma = fit_pars[1], fit_pars[3] 

958 gauss_fit = helper.gauss(xaxis, fit_mu, fit_sigma, fit_N) 

959 plt.plot( 

960 xaxis, 

961 gauss_fit, 

962 label=rf"""Gaussian Fit 

963$\mu$ = {fit_mu:.03f} $\pm$ {d_mu:.03g} 

964$\sigma$ = {fit_sigma:.03f} $\pm$ {d_sigma:.03g}""", 

965 ) 

966 

967 plt.title( 

968 helper.plot_summary_string( 

969 name="Energy", 

970 board_id=df["board_id"][0], 

971 run_numbers=run_num, 

972 channels=channel, 

973 attenuation=atten_val, 

974 pas_mode=pas_mode, 

975 gain=gain, 

976 ) 

977 ) 

978 plt.ylabel("Entries") 

979 plt.xlabel("Energy [ADC Counts]") 

980 

981 plt.grid() 

982 plt.legend(loc="upper right", frameon=False) 

983 

984 plt.savefig(plot_dir / f"""energy_{gain}_ch{channel}_{f"{df['amp'][0]:.4g}".replace(".", "p")}.png""") 

985 plt.cla() 

986 plt.clf() 

987 plt.close() 

988 return 

989 

990 

991def plot_timing_hist( 

992 df: pl.DataFrame, 

993 channel: int, 

994 plot_dir: pathlib.Path, 

995) -> None: 

996 """ 

997 Args: 

998 df: Dataframe 

999 plot_dir: Where to save plots to 

1000 

1001 Returns: 

1002 None 

1003 """ 

1004 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

1005 gain: Literal["hi", "lo"] = df["gain"][0] 

1006 run_num: int = int(df[0]["run_number"][0]) 

1007 pas_mode = df[0]["pas_mode"][0] 

1008 

1009 times = df.select(pl.col("times").list.eval(pl.element().explode())).to_numpy()[0][0] 

1010 times = times[~np.isnan(times)] 

1011 

1012 hist_bins = np.linspace(min(times), max(times), 25) 

1013 

1014 y, bins, h = plt.hist( 

1015 times, 

1016 bins=hist_bins.tolist(), 

1017 ) 

1018 skew = scipy.stats.skew(times) 

1019 _, dip_pval = diptest(y) # type: ignore 

1020 

1021 if TYPE_CHECKING: 

1022 h = cast(BarContainer, h) 

1023 h.set_label(f"""Samples ({len(times)}):\nMean = {np.round(np.mean(times), 3)} 

1024RMS = {np.round(np.std(times), 3)} 

1025γ={skew:.02f} 

1026dip={dip_pval:.02f}""") 

1027 

1028 xaxis: np.ndarray = np.linspace(np.min(times), np.max(times), 1000) 

1029 fit_pars = helper.calc_gaussian(times, bins) 

1030 fit_mu, fit_sigma, fit_N = fit_pars[0], fit_pars[2], fit_pars[4] 

1031 d_mu, d_sigma = fit_pars[1], fit_pars[3] 

1032 gauss_fit = helper.gauss(xaxis, fit_mu, fit_sigma, fit_N) 

1033 plt.plot( 

1034 xaxis, 

1035 gauss_fit, 

1036 label=rf"""Gaussian Fit 

1037$\mu$ = {fit_mu:.03f} $\pm$ {d_mu:.03g} 

1038$\sigma$ = {fit_sigma:.03f} $\pm$ {d_sigma:.03g}""", 

1039 ) 

1040 

1041 plt.title( 

1042 helper.plot_summary_string( 

1043 name="Timing", 

1044 board_id=df["board_id"][0], 

1045 run_numbers=run_num, 

1046 channels=channel, 

1047 attenuation=atten_val, 

1048 pas_mode=pas_mode, 

1049 gain=gain, 

1050 ) 

1051 ) 

1052 plt.ylabel("Entries") 

1053 plt.xlabel("time [ns]") 

1054 

1055 plt.grid() 

1056 plt.legend(loc="upper right", frameon=False) 

1057 

1058 plt.savefig(plot_dir / f"""timing_{gain}_ch{channel}_{f"{df['amp'][0]:.4g}".replace(".", "p")}.png""") 

1059 plt.cla() 

1060 plt.clf() 

1061 plt.close() 

1062 return 

1063 

1064 

1065def plot_ofc_samples( 

1066 df: pl.DataFrame, 

1067 channel: int, 

1068 plot_dir: pathlib.Path, 

1069) -> None: 

1070 """ 

1071 Args: 

1072 df: Dataframe 

1073 plot_dir: Where to save plots to 

1074 

1075 Returns: 

1076 None 

1077 """ 

1078 interleaved_samples: np.ndarray = df["samples_interleaved"].to_numpy() 

1079 mean_interleaved_pulse: np.ndarray = df["mean_interleaved_pulse"].to_numpy()[0] 

1080 max_phase_indices: np.ndarray = df["max_phase_indices"].to_numpy()[0] 

1081 OFC_amp: np.ndarray = df["OFC_amp"].to_numpy()[0] 

1082 amp: float = df["amp"].unique()[0] 

1083 

1084 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

1085 gain: Literal["hi", "lo"] = df["gain"][0] 

1086 run_num: int = int(df[0]["run_number"][0]) 

1087 pas_mode = df[0]["pas_mode"][0] 

1088 

1089 for i in range(10): 

1090 plt.plot(constants.INTERLEAVED_TIMES - helper.t_align, interleaved_samples[0][i], "b.") 

1091 

1092 plt.plot( 

1093 constants.INTERLEAVED_TIMES - helper.t_align, 

1094 mean_interleaved_pulse, 

1095 "k-", 

1096 label=f"Average interleaved pulse {amp:.3f}mA", 

1097 ) 

1098 

1099 plt.plot( 

1100 constants.INTERLEAVED_TIMES[max_phase_indices] - helper.t_align, 

1101 mean_interleaved_pulse[max_phase_indices], 

1102 "ro", 

1103 label=f"max phase: {max_phase_indices[2]:.3f} (OFC amp={OFC_amp:.3f})", 

1104 ) 

1105 

1106 plt.grid() 

1107 plt.legend(loc="upper right") 

1108 plt.xlabel("Time [ns]") 

1109 plt.ylabel("Amplitude [ADC Counts]") 

1110 plt.xlim((-200, 1000)) 

1111 plt.title( 

1112 helper.plot_summary_string( 

1113 name="OFC samples", 

1114 board_id=df["board_id"][0], 

1115 run_numbers=run_num, 

1116 channels=channel, 

1117 attenuation=atten_val, 

1118 pas_mode=pas_mode, 

1119 gain=gain, 

1120 ) 

1121 ) 

1122 

1123 plt.savefig(plot_dir / f"""ofc_samples_{gain}_ch{channel}_{f"{df['amp'][0]:.4g}".replace(".", "p")}.png""") 

1124 plt.cla() 

1125 plt.clf() 

1126 plt.close() 

1127 return 

1128 

1129 

1130def plot_pulse_gain_overlay( 

1131 df: pl.DataFrame, 

1132 channel: int, 

1133 plot_dir: pathlib.Path, 

1134) -> None: 

1135 """ 

1136 Args: 

1137 df: Dataframe 

1138 plot_dir: Where to save plots to 

1139 

1140 Returns: 

1141 None 

1142 """ 

1143 # Check that both gains contain data for the given amplitude 

1144 if df.filter(pl.col("gain") == "hi").is_empty() or df.filter(pl.col("gain") == "lo").is_empty(): 

1145 return 

1146 

1147 mean_interleaved_pulse_hi: np.ndarray = ( 

1148 df.filter(pl.col("gain") == "hi")["mean_interleaved_pulse"].to_numpy().copy()[0] 

1149 ) 

1150 mean_interleaved_pulse_lo: np.ndarray = ( 

1151 df.filter(pl.col("gain") == "lo")["mean_interleaved_pulse"].to_numpy().copy()[0] 

1152 ) 

1153 

1154 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

1155 run_num: int = int(df[0]["run_number"][0]) 

1156 pas_mode = df[0]["pas_mode"][0] 

1157 

1158 gainRatio: float = np.max(mean_interleaved_pulse_hi) / np.max(mean_interleaved_pulse_lo) 

1159 

1160 mean_interleaved_pulse_hi = mean_interleaved_pulse_hi / max(mean_interleaved_pulse_hi) 

1161 mean_interleaved_pulse_lo = mean_interleaved_pulse_lo / max(mean_interleaved_pulse_lo) 

1162 

1163 # Make sure plotting window falls within bounds of array 

1164 argmax_minus200 = np.argmax(mean_interleaved_pulse_hi) - 200 

1165 if argmax_minus200 < 0: 

1166 mean_interleaved_pulse_lo = np.concatenate( 

1167 [mean_interleaved_pulse_lo[argmax_minus200:], mean_interleaved_pulse_lo[:argmax_minus200]] 

1168 ) 

1169 mean_interleaved_pulse_hi = np.concatenate( 

1170 [mean_interleaved_pulse_hi[argmax_minus200:], mean_interleaved_pulse_hi[:argmax_minus200]] 

1171 ) 

1172 argmax_plus1000 = np.argmax(mean_interleaved_pulse_hi) + 1000 

1173 if argmax_plus1000 >= np.size(mean_interleaved_pulse_hi): 

1174 idx = argmax_plus1000 % (constants.PULSES_PER_TRAIN * constants.SAMPLES_PER_PULSE) 

1175 mean_interleaved_pulse_lo = np.concatenate([mean_interleaved_pulse_lo[idx:], mean_interleaved_pulse_lo[:idx]]) 

1176 mean_interleaved_pulse_hi = np.concatenate([mean_interleaved_pulse_hi[idx:], mean_interleaved_pulse_hi[:idx]]) 

1177 

1178 plt.plot( 

1179 mean_interleaved_pulse_lo[ 

1180 np.argmax(mean_interleaved_pulse_hi) - 200 : np.argmax(mean_interleaved_pulse_hi) + 1000 

1181 ], 

1182 label="lo", 

1183 ) 

1184 plt.plot( 

1185 mean_interleaved_pulse_hi[ 

1186 np.argmax(mean_interleaved_pulse_hi) - 200 : np.argmax(mean_interleaved_pulse_hi) + 1000 

1187 ], 

1188 label="hi", 

1189 ) 

1190 plt.plot([], label=f"Gain Ratio {gainRatio:3f}") 

1191 

1192 plt.title( 

1193 helper.plot_summary_string( 

1194 name="Gain Ratio", 

1195 board_id=df["board_id"][0], 

1196 run_numbers=run_num, 

1197 channels=channel, 

1198 attenuation=atten_val, 

1199 pas_mode=pas_mode, 

1200 ) 

1201 ) 

1202 plt.xlabel("Time [ns]") 

1203 plt.ylabel("Normalized Amplitude (A.U.)") 

1204 plt.grid() 

1205 

1206 plt.savefig(plot_dir / f"""gain_ch{channel}_overlay_{f"{df['amp'][0]:.4g}".replace(".", "p")}.png""") 

1207 plt.cla() 

1208 plt.clf() 

1209 plt.close() 

1210 return 

1211 

1212 

1213def plot_pulse_means_rms( 

1214 df: pl.DataFrame, 

1215 plot_dir: pathlib.Path, 

1216 channels: List[int], 

1217 pas_mode=None, 

1218 board_id=None, 

1219 atten_val=None, 

1220) -> None: 

1221 """ 

1222 Plot the mean and RMS of the pulse values for a given measurement and channels for high and low gain. 

1223 

1224 :param mean_dict: Dictionary containing means for hi and lo gain 

1225 :type mean_dict: dict[str, np.ndarray] 

1226 :param std_dict: Dictionary containing standard deviations for hi and lo gain 

1227 :type std_dict: dict[str, np.ndarray] 

1228 :param plot_dir: The directory to save the plot. 

1229 :type plot_dir: pathlib.Path 

1230 :param channels: The list of channels to plot. 

1231 :type channels: list[int] 

1232 """ 

1233 

1234 names = [f"ch{channel:03}" for channel in channels] 

1235 color_dict = {"lo": "b", "hi": "r"} 

1236 title_dict = {"lo": "LG", "hi": "HG"} 

1237 ylabel_dict = {"time": "time [ns]", "energy": "ADC Counts"} 

1238 plot_title_dict = {"time": "OFC Timing", "energy": "OFC Amplitude"} 

1239 run_num: int = int(df[0]["run_number"][0]) 

1240 

1241 for varn in ["energy", "time"]: 

1242 fig, ax = plt.subplots() 

1243 # plt.xticks(np.arange(0, n_channels, 4), rotation=70) 

1244 ax.xaxis.set_tick_params(pad=0.1) 

1245 fig2, ax2 = plt.subplots(1) 

1246 # plt.xticks(np.arange(0, n_channels, 4), rotation=70) 

1247 ax2.xaxis.set_tick_params(pad=0.1) 

1248 max_means = 0 

1249 max_rms = 0 

1250 min_means = 0 

1251 min_rms = 0 

1252 

1253 amp_mask = pl.col("amp") == pl.col("amp").max().over("gain") 

1254 for gain in ["lo", "hi"]: 

1255 gain_mask = pl.col("gain") == gain 

1256 stds = df.filter(amp_mask, gain_mask)[f"{varn}_std"].to_numpy() 

1257 means = df.filter(amp_mask, gain_mask)[f"{varn}_mean"].to_numpy() 

1258 color = color_dict[gain] 

1259 title = title_dict[gain] 

1260 

1261 ax.grid(visible=True, zorder=0) 

1262 ax.bar(names, means, fill=False, ec=color, label=title, zorder=3) 

1263 max_means = max(max_means, max(means)) 

1264 min_means = min(min_means, min(means)) 

1265 

1266 ax2.grid(visible=True, zorder=0) 

1267 if len(means) > 1: 

1268 mean = np.mean(stds) 

1269 std = np.std(stds) 

1270 else: 

1271 mean = stds[0] 

1272 std = 0 

1273 ax2.bar(names, stds, fill=False, ec=color, label=f"{title} mean = {mean:.2f}±{std:.2f}", zorder=3) 

1274 max_rms = max(max_rms, max(stds)) 

1275 min_rms = min(min_rms, min(stds)) 

1276 

1277 ax.set_title( 

1278 helper.plot_summary_string( 

1279 name=f"Pulse {plot_title_dict[varn]} Mean", 

1280 board_id=board_id, 

1281 run_numbers=run_num, 

1282 channels=helper.list_to_text_ranges(channels), 

1283 attenuation=atten_val, 

1284 pas_mode=pas_mode, 

1285 ) 

1286 ) 

1287 ax.set_ylabel(ylabel_dict[varn]) 

1288 ax.set_ylim(min_means - 0.25 * abs(min_means), 1.33 * max_means) 

1289 ax.legend() 

1290 fig.savefig(f"{plot_dir}/{varn}_mu_summary.png") 

1291 fig.clf() 

1292 

1293 ax2.set_title( 

1294 helper.plot_summary_string( 

1295 name=f"Pulse {plot_title_dict[varn]} RMS", 

1296 board_id=board_id, 

1297 run_numbers=run_num, 

1298 channels=helper.list_to_text_ranges(channels), 

1299 attenuation=atten_val, 

1300 pas_mode=pas_mode, 

1301 ) 

1302 ) 

1303 ax2.set_ylabel(ylabel_dict[varn]) 

1304 ax2.set_ylim(min_rms - 0.25 * abs(min_rms), 1.33 * max_rms) 

1305 ax2.legend() 

1306 fig2.savefig(f"{plot_dir}/{varn}_rms_summary.png") 

1307 fig2.clf() 

1308 

1309 plt.cla() 

1310 plt.clf() 

1311 plt.close() 

1312 

1313 

1314def plot_all_phases_energy( 

1315 df: pl.DataFrame, 

1316 channel: int, 

1317 plot_dir: pathlib.Path, 

1318) -> None: 

1319 """ 

1320 Args: 

1321 df: Dataframe 

1322 plot_dir: Where to save plots to 

1323 

1324 Returns: 

1325 None 

1326 """ 

1327 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

1328 gain: Literal["hi", "lo"] = df["gain"][0] 

1329 run_num: int = int(df[0]["run_number"][0]) 

1330 pas_mode = df[0]["pas_mode"][0] 

1331 

1332 energies: np.ndarray = df.select(pl.col("energies").list.eval(pl.element().list.to_array(30)))["energies"][ 

1333 0 

1334 ].to_numpy() 

1335 

1336 plt.boxplot( 

1337 energies, 

1338 positions=range(30), 

1339 tick_labels=[f"{i}" if i % 5 == 0 else "" for i in range(30)], 

1340 medianprops={"color": "red"}, 

1341 ) 

1342 

1343 plt.title( 

1344 helper.plot_summary_string( 

1345 name="Energy vs Phase", 

1346 board_id=df["board_id"][0], 

1347 run_numbers=run_num, 

1348 channels=channel, 

1349 attenuation=atten_val, 

1350 pas_mode=pas_mode, 

1351 gain=gain, 

1352 ) 

1353 ) 

1354 plt.ylabel("Energy [ADC Counts]") 

1355 plt.xlabel("Phase") 

1356 

1357 plt.grid(linewidth=0.5) 

1358 

1359 amp_string = f"{df['amp'][0]:.4g}".replace(".", "p") 

1360 plt.savefig(plot_dir / f"""all_phases_energy_{gain}_ch{channel}_{amp_string}.png""") 

1361 plt.cla() 

1362 plt.clf() 

1363 plt.close() 

1364 return 

1365 

1366 

1367def plot_all_phases_timing( 

1368 df: pl.DataFrame, 

1369 channel: int, 

1370 plot_dir: pathlib.Path, 

1371) -> None: 

1372 """ 

1373 Args: 

1374 df: Dataframe 

1375 plot_dir: Where to save plots to 

1376 

1377 Returns: 

1378 None 

1379 """ 

1380 atten_val: List[float] = df["att_val"].unique().sort().to_list() 

1381 gain: Literal["hi", "lo"] = df["gain"][0] 

1382 run_num: int = int(df[0]["run_number"][0]) 

1383 pas_mode = df[0]["pas_mode"][0] 

1384 

1385 times: np.ndarray = df.select(pl.col("times").list.eval(pl.element().list.to_array(30)))["times"][0].to_numpy() 

1386 

1387 plt.boxplot( 

1388 [phase[~np.isnan(phase)] for phase in times.T], 

1389 positions=range(30), 

1390 tick_labels=[f"{i}" if i % 5 == 0 else "" for i in range(30)], 

1391 medianprops={"color": "red"}, 

1392 ) 

1393 

1394 plt.title( 

1395 helper.plot_summary_string( 

1396 name="Timing vs Phase", 

1397 board_id=df["board_id"][0], 

1398 run_numbers=run_num, 

1399 channels=channel, 

1400 attenuation=atten_val, 

1401 pas_mode=pas_mode, 

1402 gain=gain, 

1403 ) 

1404 ) 

1405 plt.ylabel("Time [ns]") 

1406 plt.xlabel("Phase") 

1407 

1408 plt.grid(linewidth=0.5) 

1409 

1410 amp_string = f"{df['amp'][0]:.4g}".replace(".", "p") 

1411 plt.savefig(plot_dir / f"""all_phases_timing_{gain}_ch{channel}_{amp_string}.png""") 

1412 plt.cla() 

1413 plt.clf() 

1414 plt.close() 

1415 return