Classes | |
| class | LoggingFormatter |
| class | LogLevelError |
Functions | |
| RichHandler | setup_rich_handler () |
| str | logging_log_level_from_int (int level) |
| int | logging_log_level_from_str (str level) |
| dict[str:str] | parse_opmon_conf (logging.Logger log, dict[str:str]|"conffwk.dal.OpMonConf" conf, dict[str:str]|conffwk.dal.OpMonURI uri, str session, str application) |
| str | to_string (OpMonId opmon_id) |
| str | extract_opmon_file_path (str file_path, OpMonId|None origin=None) |
Variables | |
| dict | logging_log_levels |
| logging_log_level_keys = list(logging_log_levels.keys()) | |
| logging_log_level_values = list(logging_log_levels.values()) | |
| dict | oks_log_levels |
| oks_log_level_keys = list(oks_log_levels.keys()) | |
| oks_log_level_values = list(oks_log_levels.values()) | |
| dict | oks_to_logging_map |
| log_level_keys = logging_log_level_keys + oks_log_level_keys | |
| log_level_values = logging_log_level_values + logging_log_level_values | |
| tuple | e_log_levels |
| dict | CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]} |
| CONSOLE_THEMES = Theme({"info": "dim cyan", "warning": "magenta", "danger": "bold red"}) | |
| str | full_log_format = "%(asctime)s %(levelname)s %(filename)s %(name)s %(message)s" |
| str | rich_log_format = "%(filename)s %(name)s %(message)s" |
| str | date_time_format = "[%Y/%m/%d %H:%M:%S]" |
| time_zone = pytz.utc | |
Function Documentation
◆ extract_opmon_file_path()
| str opmonlib.utils.extract_opmon_file_path | ( | str | file_path, |
| OpMonId | None | origin = None ) |
Verify the file path can be opened.
Definition at line 277 of file utils.py.
277def extract_opmon_file_path(file_path: str, origin: OpMonId | None = None) -> str:
278 """Verify the file path can be opened."""
279 hook = "://"
280 hook_position = file_path.find(hook)
281 fname = None
282 if hook_position == -1:
283 fname = file_path
284 else:
285 fname = file_path[hook_position + len(hook) :]
286
287 if origin:
288 slash_pos = fname.rfind("/")
289 if slash_pos == -1:
290 dot_pos = fname.find(".")
291 else:
292 dot_pos = fname.find(".", slash_pos)
293 origin = to_string(origin)
294 if dot_pos == -1:
295 fname += "." + origin + ".json"
296 else:
297 fname = fname[:dot_pos] + "." + origin + fname[dot_pos:]
298
299 try:
300 with open(fname, "a"):
301 pass
302 except OSError:
303 err_str = f"Can not open file {fname}"
304 raise OSError(err_str) from None
305
306 return fname
◆ logging_log_level_from_int()
| str opmonlib.utils.logging_log_level_from_int | ( | int | level | ) |
Get the level name from its int value.
Definition at line 143 of file utils.py.
143def logging_log_level_from_int(level: int) -> str:
144 """Get the level name from its int value."""
145 if not isinstance(level, int):
146 return level
147 for k, v in logging_log_levels.items():
148 if v == level:
149 return k
150 for k, v in oks_log_levels.items():
151 if v == level:
152 return oks_to_logging_map[k]
153 err_str = f"Requested log level with value {level} is not standard ({e_log_levels})"
154 raise ValueError(err_str) from None
155
156
◆ logging_log_level_from_str()
| int opmonlib.utils.logging_log_level_from_str | ( | str | level | ) |
Get the level int from its str value.
Definition at line 157 of file utils.py.
157def logging_log_level_from_str(level: str) -> int:
158 """Get the level int from its str value."""
159 if not isinstance(level, str):
160 return level
161 for k, v in logging_log_levels.items():
162 if k == level.upper():
163 return v
164 for k in oks_log_levels.keys():
165 if k == level:
166 return logging_log_levels[oks_to_logging_map[k]]
167 err_str = f"Requested log level with value {level} is not standard ({e_log_levels})"
168 raise ValueError(err_str) from None
169
170
◆ parse_opmon_conf()
| dict[str:str] opmonlib.utils.parse_opmon_conf | ( | logging.Logger | log, |
| dict[str:str] | "conffwk.dal.OpMonConf" | conf, | ||
| dict[str:str] | conffwk.dal.OpMonURI | uri, | ||
| str | session, | ||
| str | application ) |
Parse the OpMonConf and OpMonURI.
Definition at line 171 of file utils.py.
177) -> dict[str:str]:
178 """Parse the OpMonConf and OpMonURI."""
179 if not conf:
180 log.error("Missing opmon configuration, exiting.")
181 sys.exit(1)
182 if not uri:
183 log.error("Missing opmon URI, exiting.")
184 sys.exit(1)
185
186 opmon_type = (
187 uri.get("type") if isinstance(uri, dict) else getattr(uri, "type", None)
188 )
189 if opmon_type:
190 log.debug("Found OpMon type: %s", opmon_type)
191 else:
192 log.debug(
193 "Missing 'type' in the opmon configuration, [yellow]using default value "
194 "'stdout'[/yellow]."
195 )
196 opmon_type = "stdout"
197
198 path = uri.get("path") if isinstance(uri, dict) else getattr(uri, "path", None)
199 if path:
200 log.debug("Found OpMon path: %s", path)
201 elif opmon_type != "stdout":
202 log.error("Missing 'path' in the opmon configuration, exiting.")
203 sys.exit(1)
204 else:
205 if path == []:
206 path = ""
207 log.debug("No OpMon path required for type 'stdout'.")
208
209 if opmon_type == "stream" and "monkafka" not in path:
210 msg = "OpMon 'stream' configuration must publish to kafka, exiting."
211 raise ValueError(msg) from None
212 if opmon_type != "stream" and "monkafka" in path:
213 msg = "To use kafka, the type must be set to stream."
214 raise ValueError(msg) from None
215
216 bootstrap = None
217 topic = None
218 if opmon_type == "file" and not Path(path).parent.is_dir():
219 err_str = "Requested directory to put file in does not exist."
220 raise ValueError(err_str) from None
221 if "monkafka" in path:
222 bootstrap, topic = path.split("/", 1)
223 if not topic:
224 topic = "opmon_stream"
225 log.debug("Using OpMon topic: [green]'%s'[/green]", topic)
226 log.debug("Using OpMon bootstrap: [green]'%s'[/green]", bootstrap)
227
228 level = (
229 conf.get("level") if isinstance(conf, dict) else getattr(conf, "level", None)
230 )
231 if level:
232 log.debug("Found OpMon level: [green]%s[/green]", level)
233 else:
234 log.debug(
235 "Missing 'level' in the OpMon configuration, [yellow]using default "
236 "'DEBUG'[/yellow]."
237 )
238 level = logging.DEBUG
239
240 interval_s = (
241 conf.get("interval_s")
242 if isinstance(conf, dict)
243 else getattr(conf, "interval_s", None)
244 )
245 if interval_s:
246 log.debug("Found OpMon interval_s: %s", interval_s)
247 else:
248 log.debug(
249 "Missing 'interval_s' in the opmon configuration, [yellow]using default "
250 "10s[/yellow]."
251 )
252 interval_s = 10.0
253
254 return OpMonConf(
255 opmon_type, bootstrap, topic, level, interval_s, path, session, application
256 )
257
258
◆ setup_rich_handler()
| RichHandler opmonlib.utils.setup_rich_handler | ( | ) |
Initialize a Rich handler for terminal logging.
Definition at line 125 of file utils.py.
125def setup_rich_handler() -> RichHandler:
126 """Initialize a Rich handler for terminal logging."""
127 try:
128 width = os.get_terminal_size()[0]
129 except OSError:
130 width = 150
131 handler = RichHandler(
132 console=Console(width=width),
133 omit_repeated_times=False,
134 markup=True,
135 rich_tracebacks=True,
136 show_path=False,
137 tracebacks_width=width,
138 )
139 handler.setFormatter(LoggingFormatter(fmt=rich_log_format))
140 return handler
141
142
◆ to_string()
| str opmonlib.utils.to_string | ( | OpMonId | opmon_id | ) |
Map the OpMonId to a string.
Definition at line 259 of file utils.py.
259def to_string(opmon_id: OpMonId) -> str:
260 """Map the OpMonId to a string."""
261 ret = opmon_id.get("session")
262 if not ret:
263 err_msg = "Missing session in OpMonId."
264 raise ValueError(err_msg) from None
265
266 application = opmon_id.get("application")
267 if application:
268 ret += "." + application
269
270 substructures = opmon_id.get("substructure")
271 for substructure in substructures:
272 ret += "." + substructure
273
274 return ret
275
276
Variable Documentation
◆ CONSOLE_THEMES
| opmonlib.utils.CONSOLE_THEMES = Theme({"info": "dim cyan", "warning": "magenta", "danger": "bold red"}) |
◆ CONTEXT_SETTINGS
| dict opmonlib.utils.CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]} |
◆ date_time_format
◆ e_log_levels
| tuple opmonlib.utils.e_log_levels |
◆ full_log_format
| str opmonlib.utils.full_log_format = "%(asctime)s %(levelname)s %(filename)s %(name)s %(message)s" |
◆ log_level_keys
| opmonlib.utils.log_level_keys = logging_log_level_keys + oks_log_level_keys |
◆ log_level_values
| opmonlib.utils.log_level_values = logging_log_level_values + logging_log_level_values |
◆ logging_log_level_keys
| opmonlib.utils.logging_log_level_keys = list(logging_log_levels.keys()) |
◆ logging_log_level_values
| opmonlib.utils.logging_log_level_values = list(logging_log_levels.values()) |
◆ logging_log_levels
| dict opmonlib.utils.logging_log_levels |
◆ oks_log_level_keys
| opmonlib.utils.oks_log_level_keys = list(oks_log_levels.keys()) |
◆ oks_log_level_values
| opmonlib.utils.oks_log_level_values = list(oks_log_levels.values()) |
◆ oks_log_levels
| dict opmonlib.utils.oks_log_levels |
◆ oks_to_logging_map
| dict opmonlib.utils.oks_to_logging_map |
◆ rich_log_format
| str opmonlib.utils.rich_log_format = "%(filename)s %(name)s %(message)s" |
◆ time_zone
Generated on Sat Jun 28 2025 for DUNE-DAQ by
1.12.0