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

547 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-04-30 10:50 -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 constants.INTERLEAVED_TIMES[np.argmax(fine_pulse) - 200 : np.argmax(fine_pulse) + 1000], 

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

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

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

133 ) 

134 index += 1 

135 

136 plt.grid() 

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

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

139 

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

141 plt.title( 

142 helper.plot_summary_string( 

143 name=name, 

144 board_id=board_id, 

145 run_numbers=run_num, 

146 channels=channel, 

147 pas_mode=pas_mode, 

148 gain=gain, 

149 attenuation=atten_val, 

150 ) 

151 ) 

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

153 if deriv: 

154 saveName += "_deriv" 

155 if grad: 

156 saveName += "_grad" 

157 if norm: 

158 saveName += "_norm" 

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

160 plt.tight_layout() 

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

162 plt.cla() 

163 plt.clf() 

164 plt.close() 

165 return 

166 

167 

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

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

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

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

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

173 

174 if min_power == max_power: 

175 min_power -= 1 

176 max_power += 1 

177 

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

179 

180 

181def plot_energy_resolution( 

182 df: pl.DataFrame, 

183 channel: int, 

184 plot_dir: pathlib.Path, 

185 plot_log_scale: bool = False, 

186) -> None: 

187 """ 

188 Args: 

189 df: Dataframe. Needs columns 'energy_mean, 

190 'energy_std', 'samples_interleaved', 'amp', 

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

192 plot_dir: Where to save plots to 

193 plot_log_scale: Plot using log scale. Default is False 

194 

195 Returns: 

196 None 

197 """ 

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

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

200 pas_mode = df[0]["pas_mode"][0] 

201 

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

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

204 energies_lf = energies_lf.explode("energies") 

205 energies_df = ( 

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

207 .agg( 

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

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

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

211 ) 

212 .sort(by="amp") 

213 .collect() 

214 ) 

215 

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

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

218 # TODO proper solution upstream? 

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

220 e_arr[e_arr < 0] = 0.01 

221 

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

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

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

225 

226 _, ax = plt.subplots() 

227 ax.grid(True) 

228 

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

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

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

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

233 y_err = np.abs(y_err) 

234 

235 ax.errorbar( 

236 amps_arr, 

237 dE_arr / e_arr, 

238 yerr=y_err, 

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

240 markersize=4, 

241 capsize=2, 

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

243 ) 

244 

245 try: 

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

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

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

249 

250 ax.plot( 

251 exp_x, 

252 exp_fit, 

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

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

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

256 ) 

257 except ValueError: 

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

259 

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

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

262 

263 if plot_log_scale: 

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

265 ax.set_yscale("log") 

266 ax.set_xscale("log") 

267 ax.set_xlim(power_of_10_range(amps_arr)) 

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

269 else: 

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

271 

272 ax.set_title( 

273 helper.plot_summary_string( 

274 name="Energy Resolution", 

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

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

277 channels=channel, 

278 attenuation=atten_val, 

279 pas_mode=pas_mode, 

280 gain=gain, 

281 ) 

282 ) 

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

284 

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

286 

287 plt.close() 

288 plt.cla() 

289 plt.clf() 

290 return 

291 

292 

293def plot_sigma_e( 

294 df: pl.DataFrame, 

295 channel: int, 

296 plot_dir: pathlib.Path, 

297) -> None: 

298 """ 

299 Args: 

300 df: Dataframe. Needs columns 'energy_std, 

301 'amp', 'gain', 'board_id', 

302 'run_number', and 'att_val'. 

303 plot_dir: Where to save plots to 

304 

305 Returns: 

306 None 

307 """ 

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

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

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

311 pas_mode = df[0]["pas_mode"][0] 

312 

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

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

315 energies_lf = energies_lf.explode("energies") 

316 energies_df = ( 

317 energies_lf.group_by("amp") 

318 .agg( 

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

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

321 ) 

322 .sort(by="amp") 

323 .collect() 

324 ) 

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

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

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

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

329 

330 _, ax = plt.subplots() 

331 ax.grid(True) 

332 ax.errorbar( 

333 amps_arr, 

334 dE_arr, 

335 yerr=d_dE_arr, 

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

337 markersize=4, 

338 capsize=2, 

339 ) 

340 ax.set_title( 

341 helper.plot_summary_string( 

342 name="Energy Resolution", 

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

344 run_numbers=run_num, 

345 channels=channel, 

346 attenuation=atten_val, 

347 pas_mode=pas_mode, 

348 gain=gain, 

349 ) 

350 ) 

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

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

353 

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

355 plt.close() 

356 plt.cla() 

357 plt.clf() 

358 return 

359 

360 

361def plot_sigma_T( 

362 df: pl.DataFrame, 

363 channel: int, 

364 plot_dir: pathlib.Path, 

365 plot_log_scale: bool = False, 

366) -> None: 

367 """ 

368 Args: 

369 df: Dataframe. Needs columns 'time_std, 

370 'amp', 'gain', 'board_id', 

371 'run_number', 'att_val', and 'channel'. 

372 plot_dir: Where to save plots to 

373 plot_log_scale: Plot using log scale. Default is False 

374 

375 Returns: 

376 None 

377 """ 

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

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

380 pas_mode = df[0]["pas_mode"][0] 

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

382 

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

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

385 times_lf = times_lf.explode("times") 

386 times_df = ( 

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

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

389 times_lf.explode("times") 

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

391 .agg( 

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

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

394 ) 

395 .sort(by="amp") 

396 .collect() 

397 ) 

398 

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

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

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

402 

403 _, ax = plt.subplots() 

404 ax.grid(True) 

405 

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

407 

408 ax.errorbar( 

409 amps_arr, 

410 dT_arr, 

411 yerr=d_dT_arr, 

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

413 markersize=4, 

414 capsize=2, 

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

416 ) 

417 

418 try: 

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

420 

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

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

423 

424 ax.plot( 

425 exp_x, 

426 exp_fit, 

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

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

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

430 ) 

431 except ValueError: 

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

433 

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

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

436 

437 if plot_log_scale: 

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

439 ax.set_yscale("log") 

440 ax.set_xscale("log") 

441 ax.set_xlim(power_of_10_range(amps_arr)) 

442 ax.set_ylim(power_of_10_range(dT_arr)) 

443 else: 

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

445 

446 ax.set_title( 

447 helper.plot_summary_string( 

448 name="Timing Resolution", 

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

450 run_numbers=run_num, 

451 channels=channel, 

452 attenuation=atten_val, 

453 pas_mode=pas_mode, 

454 gain=gain, 

455 ) 

456 ) 

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

458 

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

460 

461 plt.close() 

462 plt.cla() 

463 plt.clf() 

464 return 

465 

466 

467def plot_timing_mean( 

468 df: pl.DataFrame, 

469 channel: int, 

470 plot_dir: pathlib.Path, 

471) -> None: 

472 """ 

473 Args: 

474 df: Dataframe. Needs columns 'time_mean', 

475 'time_std', 'amp', 'gain', 'board_id', 

476 'run_number', 'att_val', and 'channel'. 

477 plot_dir: Where to save plots to 

478 

479 Returns: 

480 None 

481 """ 

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

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

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

485 pas_mode = df[0]["pas_mode"][0] 

486 

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

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

489 times_lf = times_lf.explode("times") 

490 times_df = ( 

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

492 .agg( 

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

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

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

496 ) 

497 .sort(by="amp") 

498 .collect() 

499 ) 

500 

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

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

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

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

505 

506 _, ax = plt.subplots() 

507 ax.grid(True) 

508 

509 ax.errorbar( 

510 amps_arr, 

511 t_arr, 

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

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

514 markersize=4, 

515 capsize=2, 

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

517 ) 

518 

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

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

521 

522 ax.set_title( 

523 helper.plot_summary_string( 

524 name="Timing Mean", 

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

526 run_numbers=run_num, 

527 channels=channel, 

528 attenuation=atten_val, 

529 pas_mode=pas_mode, 

530 gain=gain, 

531 ) 

532 ) 

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

534 

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

536 plt.savefig(plot_dir / filename) 

537 

538 plt.close() 

539 plt.cla() 

540 plt.clf() 

541 

542 return 

543 

544 

545def plot_risetime( 

546 df: pl.DataFrame, 

547 channel: int, 

548 plot_dir: pathlib.Path, 

549) -> None: 

550 """ 

551 Args: 

552 df: Dataframe. Needs columns 'rise_time', 

553 'amp', 'gain', 'board_id', 

554 'run_number', 'att_val', and 'channel'. 

555 plot_dir: Where to save plots to 

556 

557 Returns: 

558 None 

559 """ 

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

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

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

563 pas_mode = df[0]["pas_mode"][0] 

564 

565 risetimes_df = ( 

566 df.lazy() 

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

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

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

570 .sort(by="amp") 

571 .collect() 

572 ) 

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

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

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

576 

577 _, ax = plt.subplots() 

578 

579 ax.grid(zorder=0) 

580 ax.set_title( 

581 helper.plot_summary_string( 

582 name="Rise times", 

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

584 run_numbers=run_num, 

585 channels=channel, 

586 attenuation=atten_val, 

587 pas_mode=pas_mode, 

588 gain=gain, 

589 ) 

590 ) 

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

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

593 # ax.set_ylim(0, 100) 

594 

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

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

597 plt.cla() 

598 plt.clf() 

599 return 

600 

601 

602def plot_zero_crossing( 

603 df: pl.DataFrame, 

604 channel: int, 

605 plot_dir: pathlib.Path, 

606) -> None: 

607 """ 

608 Args: 

609 df: Dataframe. Needs columns 'zero_crossing_time', 

610 'amp', 'gain', 'board_id', 

611 'run_number', 'att_val', and 'channel'. 

612 plot_dir: Where to save plots to 

613 

614 Returns: 

615 None 

616 """ 

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

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

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

620 pas_mode = df[0]["pas_mode"][0] 

621 zero_crossing_df = ( 

622 df.lazy() 

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

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

625 .agg( 

626 [ 

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

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

629 ] 

630 ) 

631 .sort(by="amp") 

632 .collect() 

633 ) 

634 

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

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

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

638 

639 _, ax = plt.subplots() 

640 

641 ax.grid(zorder=0) 

642 ax.set_title( 

643 helper.plot_summary_string( 

644 name="Zero Crossing Time from Peak", 

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

646 run_numbers=run_num, 

647 channels=channel, 

648 attenuation=atten_val, 

649 pas_mode=pas_mode, 

650 gain=gain, 

651 ) 

652 ) 

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

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

655 # ax.set_ylim(80, 120) 

656 

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

658 

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

660 plt.cla() 

661 plt.clf() 

662 return 

663 

664 

665def plot_gain_ratios( 

666 df: pl.DataFrame, 

667 channel: int, 

668 plot_dir: pathlib.Path, 

669) -> None: 

670 """ 

671 Args: 

672 df: Dataframe. Needs columns 'energy_mean', 

673 'amp', 'gain', 'board_id', 

674 'run_number', 'att_val', and 'channel'. 

675 plot_dir: Where to save plots to 

676 

677 Returns: 

678 None 

679 """ 

680 gain_ratios = ( 

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

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

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

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

685 ) 

686 

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

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

689 pas_mode = df[0]["pas_mode"][0] 

690 

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

692 plt.title( 

693 helper.plot_summary_string( 

694 name="Gain Ratio Summary", 

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

696 run_numbers=run_num, 

697 channels=channel, 

698 attenuation=atten_val, 

699 pas_mode=pas_mode, 

700 ) 

701 ) 

702 plt.grid() 

703 plt.ylabel("Gain Energy Ratio") 

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

705 

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

707 plt.cla() 

708 plt.clf() 

709 return 

710 

711 

712def plot_INL( 

713 df: pl.DataFrame, 

714 channel: int, 

715 plot_dir: pathlib.Path, 

716 skip_last_n: Optional[int] = None, 

717) -> None: 

718 """ 

719 Args: 

720 df: Dataframe. Needs columns 'energy_mean', 

721 'energy_std', 'amp', 'gain', 'board_id', 

722 'run_number', 'att_val', and 'channel'. 

723 plot_dir: Where to save plots to 

724 

725 skip_last_n: skip last n points 

726 

727 Returns: 

728 None 

729 """ 

730 

731 if skip_last_n is not None: 

732 skip_last_n = -skip_last_n 

733 

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

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

736 energies_lf = energies_lf.explode("energies") 

737 energies_df = ( 

738 energies_lf.group_by("amp") 

739 .agg( 

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

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

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

743 ) 

744 .sort(by="amp") 

745 .collect() 

746 ) 

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

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

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

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

751 

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

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

754 return 

755 

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

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

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

759 pas_mode = df[0]["pas_mode"][0] 

760 

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

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

763 

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

765 fig.suptitle( 

766 helper.plot_summary_string( 

767 name="Linearity", 

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

769 run_numbers=run_num, 

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

771 attenuation=atten_val, 

772 gain=gain, 

773 pas_mode=pas_mode, 

774 ) 

775 ) 

776 

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

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

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

780 

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

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

783 ax2.set_xscale("log") 

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

785 

786 ax1.grid() 

787 ax2.grid() 

788 

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

790 

791 popt, pcov = curve_fit( 

792 helper.lin, 

793 amps_arr[:skip_last_n], 

794 e_arr[:skip_last_n], 

795 # amps_arr[amps_arr <= plateau_amp], 

796 # e_arr[amps_arr <= plateau_amp], 

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

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

799 sigma=dE_arr[:skip_last_n], 

800 absolute_sigma=True, 

801 ) 

802 

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

804 

805 popt, pcov = curve_fit( 

806 helper.lin, 

807 amps_arr[:skip_last_n], 

808 e_arr[:skip_last_n], 

809 # amps_arr[amps_arr <= plateau_amp], 

810 # e_arr[amps_arr <= plateau_amp], 

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

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

813 sigma=dE_arr[:skip_last_n], 

814 absolute_sigma=True, 

815 ) 

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

817 

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

819 

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

821 ax1.text( 

822 0.1, 

823 0.9, 

824 text, 

825 horizontalalignment="left", 

826 verticalalignment="top", 

827 transform=ax1.transAxes, 

828 ) 

829 

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

831 

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

833 

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

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

836 

837 if title == "LG": 

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

839 ax2.set_ylim(-3, 3) 

840 elif title == "HG": 

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

842 ax2.set_ylim(-3, 3) 

843 

844 plt.tight_layout() 

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

846 plt.close() 

847 plt.cla() 

848 plt.clf() 

849 

850 return 

851 

852 

853def plot_autocorrelation( 

854 df: pl.DataFrame, 

855 channel: int, 

856 plot_dir: pathlib.Path, 

857) -> None: 

858 """ 

859 Args: 

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

861 plot_dir: Where to save plots to 

862 

863 Returns: 

864 None 

865 """ 

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

867 

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

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

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

871 pas_mode = df[0]["pas_mode"][0] 

872 

873 _, ax = plt.subplots() 

874 ax.grid(True) 

875 norm_ac = autocorr / autocorr[0] 

876 ax.set_title( 

877 helper.plot_summary_string( 

878 name="Autocorrelcation", 

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

880 run_numbers=run_num, 

881 channels=channel, 

882 attenuation=atten_val, 

883 pas_mode=pas_mode, 

884 gain=gain, 

885 ) 

886 ) 

887 ax.set_xlabel("Lag") 

888 ax.set_ylabel("ACF") 

889 

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

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

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

893 

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

895 plt.cla() 

896 plt.clf() 

897 plt.close() 

898 return 

899 

900 

901""" 

902========================================================= 

903 Plots for the low and hi gain webpages 

904========================================================= 

905""" 

906 

907 

908def plot_energy_hist( 

909 df: pl.DataFrame, 

910 channel: int, 

911 plot_dir: pathlib.Path, 

912) -> None: 

913 """ 

914 Args: 

915 df: Dataframe 

916 plot_dir: Where to save plots to 

917 

918 Returns: 

919 None 

920 """ 

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

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

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

924 pas_mode = df[0]["pas_mode"][0] 

925 

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

927 

928 if gain == "hi": 

929 hist_bins = np.linspace( 

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

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

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

933 ) 

934 elif gain == "lo": 

935 hist_bins = np.linspace( 

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

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

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

939 ) 

940 

941 y, bins, h = plt.hist( 

942 energies, 

943 bins=hist_bins.tolist(), 

944 ) 

945 skew = scipy.stats.skew(energies) 

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

947 

948 if TYPE_CHECKING: 

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

950 h[0].set_label( 

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

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

953 ) 

954 

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

956 fit_pars = helper.calc_gaussian(energies, bins) 

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

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

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

960 plt.plot( 

961 xaxis, 

962 gauss_fit, 

963 label=rf"""Gaussian Fit 

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

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

966 ) 

967 

968 plt.title( 

969 helper.plot_summary_string( 

970 name="Energy", 

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

972 run_numbers=run_num, 

973 channels=channel, 

974 attenuation=atten_val, 

975 pas_mode=pas_mode, 

976 gain=gain, 

977 ) 

978 ) 

979 plt.ylabel("Entries") 

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

981 

982 plt.grid() 

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

984 

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

986 plt.cla() 

987 plt.clf() 

988 plt.close() 

989 return 

990 

991 

992def plot_timing_hist( 

993 df: pl.DataFrame, 

994 channel: int, 

995 plot_dir: pathlib.Path, 

996) -> None: 

997 """ 

998 Args: 

999 df: Dataframe 

1000 plot_dir: Where to save plots to 

1001 

1002 Returns: 

1003 None 

1004 """ 

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

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

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

1008 pas_mode = df[0]["pas_mode"][0] 

1009 

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

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

1012 

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

1014 

1015 y, bins, h = plt.hist( 

1016 times, 

1017 bins=hist_bins.tolist(), 

1018 ) 

1019 skew = scipy.stats.skew(times) 

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

1021 

1022 if TYPE_CHECKING: 

1023 h = cast(BarContainer, h) 

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

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

1026γ={skew:.02f} 

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

1028 

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

1030 fit_pars = helper.calc_gaussian(times, bins) 

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

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

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

1034 plt.plot( 

1035 xaxis, 

1036 gauss_fit, 

1037 label=rf"""Gaussian Fit 

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

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

1040 ) 

1041 

1042 plt.title( 

1043 helper.plot_summary_string( 

1044 name="Timing", 

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

1046 run_numbers=run_num, 

1047 channels=channel, 

1048 attenuation=atten_val, 

1049 pas_mode=pas_mode, 

1050 gain=gain, 

1051 ) 

1052 ) 

1053 plt.ylabel("Entries") 

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

1055 

1056 plt.grid() 

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

1058 

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

1060 plt.cla() 

1061 plt.clf() 

1062 plt.close() 

1063 return 

1064 

1065 

1066def plot_ofc_samples( 

1067 df: pl.DataFrame, 

1068 channel: int, 

1069 plot_dir: pathlib.Path, 

1070) -> None: 

1071 """ 

1072 Args: 

1073 df: Dataframe 

1074 plot_dir: Where to save plots to 

1075 

1076 Returns: 

1077 None 

1078 """ 

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

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

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

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

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

1084 

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

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

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

1088 pas_mode = df[0]["pas_mode"][0] 

1089 

1090 for i in range(10): 

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

1092 

1093 plt.plot( 

1094 constants.INTERLEAVED_TIMES - helper.t_align, 

1095 mean_interleaved_pulse, 

1096 "k-", 

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

1098 ) 

1099 

1100 plt.plot( 

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

1102 mean_interleaved_pulse[max_phase_indices], 

1103 "ro", 

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

1105 ) 

1106 

1107 plt.grid() 

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

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

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

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

1112 plt.title( 

1113 helper.plot_summary_string( 

1114 name="OFC samples", 

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

1116 run_numbers=run_num, 

1117 channels=channel, 

1118 attenuation=atten_val, 

1119 pas_mode=pas_mode, 

1120 gain=gain, 

1121 ) 

1122 ) 

1123 

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

1125 plt.cla() 

1126 plt.clf() 

1127 plt.close() 

1128 return 

1129 

1130 

1131def plot_pulse_gain_overlay( 

1132 df: pl.DataFrame, 

1133 channel: int, 

1134 plot_dir: pathlib.Path, 

1135) -> None: 

1136 """ 

1137 Args: 

1138 df: Dataframe 

1139 plot_dir: Where to save plots to 

1140 

1141 Returns: 

1142 None 

1143 """ 

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

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

1146 return 

1147 

1148 mean_interleaved_pulse_hi: np.ndarray = ( 

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

1150 ) 

1151 mean_interleaved_pulse_lo: np.ndarray = ( 

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

1153 ) 

1154 

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

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

1157 pas_mode = df[0]["pas_mode"][0] 

1158 

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

1160 

1161 mean_interleaved_pulse_hi = mean_interleaved_pulse_hi / max(mean_interleaved_pulse_hi) 

1162 mean_interleaved_pulse_lo = mean_interleaved_pulse_lo / max(mean_interleaved_pulse_lo) 

1163 

1164 # Make sure plotting window falls within bounds of array 

1165 argmax_minus200 = np.argmax(mean_interleaved_pulse_hi) - 200 

1166 if argmax_minus200 < 0: 

1167 mean_interleaved_pulse_lo = np.concatenate( 

1168 [mean_interleaved_pulse_lo[argmax_minus200:], mean_interleaved_pulse_lo[:argmax_minus200]] 

1169 ) 

1170 mean_interleaved_pulse_hi = np.concatenate( 

1171 [mean_interleaved_pulse_hi[argmax_minus200:], mean_interleaved_pulse_hi[:argmax_minus200]] 

1172 ) 

1173 argmax_plus1000 = np.argmax(mean_interleaved_pulse_hi) + 1000 

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

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

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

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

1178 

1179 plt.plot( 

1180 mean_interleaved_pulse_lo[ 

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

1182 ], 

1183 label="lo", 

1184 ) 

1185 plt.plot( 

1186 mean_interleaved_pulse_hi[ 

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

1188 ], 

1189 label="hi", 

1190 ) 

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

1192 

1193 plt.title( 

1194 helper.plot_summary_string( 

1195 name="Gain Ratio", 

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

1197 run_numbers=run_num, 

1198 channels=channel, 

1199 attenuation=atten_val, 

1200 pas_mode=pas_mode, 

1201 ) 

1202 ) 

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

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

1205 plt.grid() 

1206 

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

1208 plt.cla() 

1209 plt.clf() 

1210 plt.close() 

1211 return 

1212 

1213 

1214def plot_pulse_means_rms( 

1215 df: pl.DataFrame, 

1216 plot_dir: pathlib.Path, 

1217 channels: List[int], 

1218 pas_mode=None, 

1219 board_id=None, 

1220 atten_val=None, 

1221) -> None: 

1222 """ 

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

1224 

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

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

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

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

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

1230 :type plot_dir: pathlib.Path 

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

1232 :type channels: list[int] 

1233 """ 

1234 

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

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

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

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

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

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

1241 

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

1243 fig, ax = plt.subplots() 

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

1245 ax.xaxis.set_tick_params(pad=0.1) 

1246 fig2, ax2 = plt.subplots(1) 

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

1248 ax2.xaxis.set_tick_params(pad=0.1) 

1249 max_means = 0 

1250 max_rms = 0 

1251 min_means = 0 

1252 min_rms = 0 

1253 

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

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

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

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

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

1259 color = color_dict[gain] 

1260 title = title_dict[gain] 

1261 

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

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

1264 max_means = max(max_means, max(means)) 

1265 min_means = min(min_means, min(means)) 

1266 

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

1268 if len(means) > 1: 

1269 mean = np.mean(stds) 

1270 std = np.std(stds) 

1271 else: 

1272 mean = stds[0] 

1273 std = 0 

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

1275 max_rms = max(max_rms, max(stds)) 

1276 min_rms = min(min_rms, min(stds)) 

1277 

1278 ax.set_title( 

1279 helper.plot_summary_string( 

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

1281 board_id=board_id, 

1282 run_numbers=run_num, 

1283 channels=helper.list_to_text_ranges(channels), 

1284 attenuation=atten_val, 

1285 pas_mode=pas_mode, 

1286 ) 

1287 ) 

1288 ax.set_ylabel(ylabel_dict[varn]) 

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

1290 ax.legend() 

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

1292 fig.clf() 

1293 

1294 ax2.set_title( 

1295 helper.plot_summary_string( 

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

1297 board_id=board_id, 

1298 run_numbers=run_num, 

1299 channels=helper.list_to_text_ranges(channels), 

1300 attenuation=atten_val, 

1301 pas_mode=pas_mode, 

1302 ) 

1303 ) 

1304 ax2.set_ylabel(ylabel_dict[varn]) 

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

1306 ax2.legend() 

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

1308 fig2.clf() 

1309 

1310 plt.cla() 

1311 plt.clf() 

1312 plt.close() 

1313 

1314 

1315def plot_all_phases_energy( 

1316 df: pl.DataFrame, 

1317 channel: int, 

1318 plot_dir: pathlib.Path, 

1319) -> None: 

1320 """ 

1321 Args: 

1322 df: Dataframe 

1323 plot_dir: Where to save plots to 

1324 

1325 Returns: 

1326 None 

1327 """ 

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

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

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

1331 pas_mode = df[0]["pas_mode"][0] 

1332 

1333 energies: np.ndarray = df.select(pl.col("energies").list.eval(pl.element().list.to_array(constants.N_PHASES)))[ 

1334 "energies" 

1335 ][0].to_numpy() 

1336 

1337 plt.boxplot( 

1338 energies, 

1339 positions=range(constants.N_PHASES), 

1340 tick_labels=[f"{i}" if i % 5 == 0 else "" for i in range(constants.N_PHASES)], 

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

1342 ) 

1343 

1344 plt.title( 

1345 helper.plot_summary_string( 

1346 name="Energy vs Phase", 

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

1348 run_numbers=run_num, 

1349 channels=channel, 

1350 attenuation=atten_val, 

1351 pas_mode=pas_mode, 

1352 gain=gain, 

1353 ) 

1354 ) 

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

1356 plt.xlabel("Phase") 

1357 

1358 plt.grid(linewidth=0.5) 

1359 

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

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

1362 plt.cla() 

1363 plt.clf() 

1364 plt.close() 

1365 return 

1366 

1367 

1368def plot_all_phases_timing( 

1369 df: pl.DataFrame, 

1370 channel: int, 

1371 plot_dir: pathlib.Path, 

1372) -> None: 

1373 """ 

1374 Args: 

1375 df: Dataframe 

1376 plot_dir: Where to save plots to 

1377 

1378 Returns: 

1379 None 

1380 """ 

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

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

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

1384 pas_mode = df[0]["pas_mode"][0] 

1385 

1386 times: np.ndarray = df.select(pl.col("times").list.eval(pl.element().list.to_array(constants.N_PHASES)))["times"][ 

1387 0 

1388 ].to_numpy() 

1389 

1390 plt.boxplot( 

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

1392 positions=range(constants.N_PHASES), 

1393 tick_labels=[f"{i}" if i % 5 == 0 else "" for i in range(constants.N_PHASES)], 

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

1395 ) 

1396 

1397 plt.title( 

1398 helper.plot_summary_string( 

1399 name="Timing vs Phase", 

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

1401 run_numbers=run_num, 

1402 channels=channel, 

1403 attenuation=atten_val, 

1404 pas_mode=pas_mode, 

1405 gain=gain, 

1406 ) 

1407 ) 

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

1409 plt.xlabel("Phase") 

1410 

1411 plt.grid(linewidth=0.5) 

1412 

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

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

1415 plt.cla() 

1416 plt.clf() 

1417 plt.close() 

1418 return