Coverage for src / molecular_simulations / logging_config.py: 96%

1from __future__ import annotations 

2import logging 

3import logging.config 

4import os 

5import socket 

6from datetime import datetime 

7 

8def configure_logging( 

9 level: str | int | None = None, 

10 to_file: str | None = None, 

11 fmt: str | None = None, 

12) -> None: 

13 """ 

14 Opt-in configuration for apps/CLIs/examples. 

15 Library code should *not* call this implicitly. 

16 """ 

17 level = (level or os.getenv("MS_LOG_LEVEL") or "INFO") 

18 fmt = fmt or os.getenv("MS_LOG_FMT") or ( 

19 "%(asctime)s | %(levelname)s | %(name)s | %(message)s " 

20 "[host=%(hostname)s pid=%(process)d rank=%(mpirank)s]" 

21 ) 

22 

23 class _ContextFilter(logging.Filter): 

24 def filter(self, record: logging.LogRecord) -> bool: 

25 record.hostname = socket.gethostname() 

26 # Fill MPI rank if available; else 0 

27 try: 

28 from mpi4py import MPI # noqa: WPS433 

29 record.mpirank = MPI.COMM_WORLD.Get_rank() 

30 except Exception: 

31 record.mpirank = 0 

32 return True 

33 

34 handlers = { 

35 "console": { 

36 "class": "logging.StreamHandler", 

37 "level": level, 

38 "filters": ["ctx"], 

39 "formatter": "standard", 

40 } 

41 } 

42 

43 if to_file or os.getenv("MS_LOG_FILE"): 

44 handlers["file"] = { 

45 "class": "logging.handlers.RotatingFileHandler", 

46 "level": level, 

47 "filters": ["ctx"], 

48 "formatter": "standard", 

49 "filename": to_file or os.getenv("MS_LOG_FILE"), 

50 "maxBytes": 10 * 1024 * 1024, 

51 "backupCount": 3, 

52 "encoding": "utf-8", 

53 } 

54 

55 config = { 

56 "version": 1, 

57 "disable_existing_loggers": False, 

58 "filters": {"ctx": {"()": _ContextFilter}}, 

59 "formatters": {"standard": {"format": fmt}}, 

60 "handlers": handlers, 

61 "root": {"level": level, "handlers": list(handlers)}, 

62 } 

63 

64 logging.config.dictConfig(config) 

65