Coverage for polars_analysis / cli_frame.py: 50%

1import logging 

2from pathlib import Path 

3from typing import Annotated, List, Optional 

4 

5import typer 

6 

7from polars_analysis import frame, utils 

8from polars_analysis.data_sources import DeltaSource 

9 

10# Instantiate logger 

11log = logging.getLogger(__name__) 

12 

13app = typer.Typer( 

14 no_args_is_help=True, 

15 help="Frame analysis and plotting", 

16) 

17 

18 

19@app.command("misalignment-scan", no_args_is_help=True) 

20def misalignment_scan( 

21 min_run_number: Annotated[int, typer.Option(help="Min run number to include")], 

22 max_run_number: Annotated[int, typer.Option(help="Max run number to include")], 

23 skip_single_adc: Annotated[bool, typer.Option(help="Skip single ADC and only scan trigger mode runs")] = True, 

24 test_channel: Annotated[int, typer.Option(help="Channel for more detailed printout")] = 7, 

25 data_dir: Annotated[Path, typer.Option(envvar="DATA_DIR", help="path to delta table")] = Path("data/samples/"), 

26 frame_dir: Annotated[Path, typer.Option(envvar="FRAME_DIR", help="path to directory with frame data")] = Path( 

27 "data/frame" 

28 ), 

29 plot_dir: Annotated[Path, typer.Option(envvar="BOARDS_PLOT_DIR", help="path to directory to save plots")] = Path( 

30 "plots" 

31 ), 

32 derived_dir: Annotated[ 

33 Path, typer.Option(envvar="DERIVED_DIR", help="path to directory to save derived values") 

34 ] = Path("derived"), 

35): 

36 loader = DeltaSource(data_dir, derived_dir, frame_dir) 

37 

38 frame.scan_for_misalignment( 

39 loader, 

40 min_run_number, 

41 max_run_number, 

42 plot_dir, 

43 skip_single_adc, 

44 test_channel, 

45 ) 

46 

47 

48@app.command("plot-extended-readout", no_args_is_help=True) 

49def plot_extended_readout( 

50 run_number: Annotated[int, typer.Argument(help="Run number to plot")], 

51 trigger_window: Annotated[int, typer.Option(help="Number of samples in trigger window")] = 128, 

52 trigger_rate: Annotated[float, typer.Option(help="Felix trigger rate in Hz")] = 152.588, 

53 skip_channels: Annotated[ 

54 Optional[List[str]], 

55 typer.Option(help="list of channels to skip in coherent noise calculation and baseline plots"), 

56 ] = None, 

57 data_dir: Annotated[Path, typer.Option(envvar="DATA_DIR", help="path to delta table")] = Path("data/samples/"), 

58 frame_dir: Annotated[Path, typer.Option(envvar="FRAME_DIR", help="path to directory with frame data")] = Path( 

59 "data/frame" 

60 ), 

61 plot_dir: Annotated[Path, typer.Option(envvar="RUNS_PLOT_DIR", help="path to directory to save plots")] = Path( 

62 "plots" 

63 ), 

64 derived_dir: Annotated[ 

65 Path, typer.Option(envvar="DERIVED_DIR", help="path to directory to save derived values") 

66 ] = Path("derived"), 

67 swap_frame18: Annotated[bool, typer.Option(help="Swap which ADCs are associated to Frame1 and Frame8")] = False, 

68 base_corr_int_period: Annotated[ 

69 Optional[float], 

70 typer.Option(help="Target time period in seconds to integrate over when correcting for baseline wandering."), 

71 ] = None, 

72): 

73 loader = DeltaSource(data_dir, derived_dir, frame_dir) 

74 

75 if skip_channels: 

76 skip_channels_result = utils.parse_skip_channels(skip_channels) 

77 if not skip_channels_result: 

78 log.error(f"Failed to find a number in one of the requested skip channels: {skip_channels}") 

79 log.error("Exiting") 

80 exit(1) 

81 skip_channels_lo, skip_channels_hi = skip_channels_result 

82 else: 

83 skip_channels_lo, skip_channels_hi = None, None 

84 

85 frame.plot_extended_readout_from_loader( 

86 loader, 

87 run_number, 

88 plot_dir, 

89 trigger_window, 

90 trigger_rate, 

91 skip_channels_lo, 

92 skip_channels_hi, 

93 swap_frame18, 

94 base_corr_int_period, 

95 )