Coverage for src / lstautorta / config / hdf5_data_check.py: 0%

1from typing import Annotated, Any, Literal 

2 

3from annotated_types import Gt 

4from pydantic import BaseModel, Field 

5 

6 

7class GroupByConfiguration(BaseModel): 

8 groupby: list[str | dict[str, Any]] = Field( 

9 title="GroupBy Arguments", 

10 description='Argument passed to the "by" argument of a pandas DataFrame groupby function. ' 

11 'If a string, it is passed directly to groupby as "by" argument. If a Dict, the content is expanding to a pandas.Grouper kwargs.', 

12 examples=[["worker_node"], [{"key": "timestamp", "freq": "1S"}, "worker_node"]], 

13 ) 

14 kwargs: dict[str, Any] | None = Field( 

15 title="GroupBy kwargs", 

16 description="Content of the dict is passed to groupby as kwargs.", 

17 examples=[{"axis": 0}], 

18 default=None, 

19 ) 

20 computation: tuple[str, dict[str, Any] | None] = Field( 

21 title="Grouped by DataFrame Computation", 

22 description="Operation to apply to the grouped-by DataFrame, with optional kwargs", 

23 examples=[("mean", None), ("sum", {"numeric_only": False})], 

24 ) 

25 

26 

27class PlotConfiguration(BaseModel): 

28 groupby: GroupByConfiguration | None = Field( 

29 title="Optional groupby operation configuration", 

30 description="Optional configuration for a groupby operation to apply on the DataFrame before plotting", 

31 examples=[{"groupby": [{"key": "timestamp", "freq": "1S"}, "worker_node"], "computation": {"mean": None}}], 

32 default=None, 

33 ) 

34 kind: Literal["plot", "hist"] = Field( 

35 title="Plot Kind", description='Kind of plot to draw, eg "plot", "hist"', examples=["plot", "hist"] 

36 ) 

37 kwargs: dict[str, Any] | None = Field( 

38 title="Plot Function Keyword Arguments", 

39 description="Any item in this dictionnary will be passed to the plotting function as a keyword argument " 

40 "In particular, this should be used to pass the 'x' and 'y' arguments to seaborn plot, histplot, scatterplot, etc. functions.", 

41 examples=[{"x": "trigger_time", "y": "event_id_diffs", "markers": True}, {"bins": 100}], 

42 ) 

43 mask_event_quality: bool = Field( 

44 title="Mask Bad Events", 

45 description="If True, the events with event_quality != 0 will not be considered", 

46 examples=[True, False], 

47 default=False, 

48 ) 

49 mask_future_events: bool = Field( 

50 title="Mask Future Events", 

51 description='If True, the "future" events will not be plotted', 

52 examples=[True], 

53 default=True, 

54 ) 

55 mask_is_good_event: bool = Field( 

56 title="Mask Rejected Events", 

57 description='If True, the events with "is_good_event==False" will not be plotted"', 

58 examples=[True], 

59 default=False, 

60 ) 

61 np_function_x: tuple[str, dict[str, Any] | None] | None = Field( 

62 title="Numpy Function applied to x argument of plot function", 

63 description="numpy function applied to data plotted on x axis, such as log, log10, etc. " 

64 "and optionaly arguments as kwargs in a dictionary", 

65 examples=[("log10", None), ("log10", {"casting": "safe"})], 

66 default=None, 

67 ) 

68 np_function_y: tuple[str, dict[str, Any] | None] | None = Field( 

69 title="Numpy Function applied to y argument of plot function", 

70 description="numpy function applied to data plotted on y axis, such as log, log10, etc. " 

71 "and optionaly arguments as kwargs in a dictionary", 

72 examples=[("log10", None), ("log10", {"casting": "safe"})], 

73 default=None, 

74 ) 

75 title: str | None = Field( 

76 title="Title of the plot", 

77 description="Title to display in the plotted Figure", 

78 examples=["Delta event ID"], 

79 default=None, 

80 ) 

81 xlabel: str | None = Field( 

82 title="xlabel", description="Label for the x axis", examples=["Trigger Time (s)"], default=None 

83 ) 

84 ylabel: str | None = Field(title="ylabel", description="ylabel", examples=["Rate (Hz)"], default=None) 

85 

86 

87class HDFDataCheckConfiguration(BaseModel): 

88 """Configures what/how to plot in a data level auto_check routines""" 

89 

90 column_to_datetime: list[str] | None = Field( 

91 title="Timestamps columns", 

92 description="List of columns (before re-name!) to cast from timestamp to datetime (unit=s)", 

93 examples=["trigger_time"], 

94 ) 

95 data_path_in_hdf5: str = Field( 

96 title="Data Key in HDF5", 

97 description="Full path to the pytables Table containing the data to plot in the hdf5 files", 

98 examples=["/dl1/event/telescope/parameters/LST_LSTCam"], 

99 ) 

100 hdf5_open_nb_retries: Annotated[int, Gt(0)] = Field( 

101 title="Number of HDF5 reading tries", 

102 description="Number of times to try to read an HDF5 file that may be locked by another process", 

103 examples=[20], 

104 ) 

105 hdf5_open_wait_time_s: Annotated[float, Gt(0)] = Field( 

106 title="HDF5 reading re-try wait time", 

107 description="Amount of time in second to wait before re-trying to read an HDF5 file.", 

108 examples=[0.5], 

109 ) 

110 plots_config: dict[str, PlotConfiguration] = Field( 

111 title='Mapping of plots path ("name.png") to plot configuration', 

112 description="Mapping between a plot png path to a plot configuration", 

113 examples=[ 

114 { 

115 "Event ID": { 

116 "kind": "hist", 

117 "kwargs": {"x": "total_intensity", "alpha": 0.5, "bins": 100, "log_scale": (True, True)}, 

118 } 

119 } 

120 ], 

121 ) 

122 plot_refresh_interval_s: float = Field( 

123 title="Plot Refresh Interval", 

124 description="Amount of time in seconds between 2 plot of the data", 

125 examples=[10.0], 

126 ) 

127 size_storage_dataframe: Annotated[int, Gt(0)] = Field( 

128 title="Accumulation DataFrame Size", 

129 description="Number of rows to allocate to the DataFrame used to accumulate the files data. " 

130 "Should be large enough to easily be more than 1 run's number of events, eg 15 000 * 60*40 (15kHz for 40min)", 

131 examples=[36000000], 

132 ) 

133 write_df_store_key: str = Field( 

134 title="Written DF key in hdf5", 

135 description="Path in the written HDF5 files of the DataFrames data.", 

136 examples=["auto_check_df"], 

137 )