Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/llm_datadist_v1/utils/log.py: 93%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ------------------------------------------------------------------- 

4# ----------------------------------------------------------------------------------------------------------- 

5# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

6# This program is free software, you can redistribute it and/or modify it under the terms and conditions of 

7# CANN Open Software License Agreement Version 2.0 (the "License"). 

8# Please refer to the License for details. You may not use this file except in compliance with the License. 

9# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, 

10# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. 

11# See LICENSE in the root of the software repository for the full text of the License. 

12# ----------------------------------------------------------------------------------------------------------- 

13 

14import logging 

15import os 

16import sys 

17import threading 

18import traceback 

19 

20_global_logger = None 

21_global_logger_lock = threading.Lock() 

22_log_level_map = { 

23 "0": logging.DEBUG, 

24 "1": logging.INFO, 

25 "2": logging.WARNING, 

26 "3": logging.ERROR, 

27} 

28LOG_FORMAT = ( 

29 "[%(levelname)s] %(name)s(%(process)d,%(processName)s):%(asctime)s.%(msecs)03d " 

30 "[%(filename)s:%(lineno)d]%(thread)d %(funcName)s: %(message)s" 

31) 

32 

33 

34def get_logger(): 

35 global _global_logger 

36 if _global_logger: 

37 return _global_logger 

38 with _global_logger_lock: 

39 if _global_logger: 

40 return _global_logger 

41 logger = logging.getLogger("LLM_ENGINE") 

42 logger.findCaller = _find_caller 

43 log_level = os.getenv("ASCEND_GLOBAL_LOG_LEVEL") 

44 logger.setLevel(_log_level_map.get(log_level, logging.ERROR)) 

45 handler = logging.StreamHandler(sys.stdout) 

46 handler.setFormatter(logging.Formatter(LOG_FORMAT, datefmt="%Y-%m-%d %H:%M:%S")) 

47 logger.addHandler(handler) 

48 _global_logger = logger 

49 return _global_logger 

50 

51 

52def info(msg, *args, **kwargs): 

53 get_logger().info(msg, *args, **kwargs) 

54 

55 

56def warn(msg, *args, **kwargs): 

57 get_logger().warn(msg, *args, **kwargs) 

58 

59 

60def error(msg, *args, **kwargs): 

61 get_logger().error(msg, *args, **kwargs) 

62 

63 

64def _get_stack_info(frame): 

65 return "Stack (most recent call last):\n" + "".join(traceback.format_stack(frame)) 

66 

67 

68def _find_caller(stack_info=False, stack_level=1): 

69 f = sys._getframe(3) 

70 log_file = f.f_code.co_filename 

71 rv = "(unknown file)", 0, "(unknown function)", None 

72 while f: 

73 f = f.f_back 

74 code = f.f_code 

75 if code.co_filename != log_file: 

76 sinfo = _get_stack_info(f) if stack_info else None 

77 rv = (code.co_filename, f.f_lineno, code.co_name, sinfo) 

78 break 

79 return rv