meerschaum.utils.debug

  1#! /usr/bin/env python
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Functions to handle debug statements
  7"""
  8
  9from __future__ import annotations
 10
 11from datetime import datetime, timezone
 12import meerschaum as mrsm
 13from meerschaum.utils.typing import Union, Optional, List
 14
 15
 16_rich_text = None
 17def _import_rich_text_for_dprint():
 18    """
 19    Avoid calling `attempt_import()` on every dprint.
 20    """
 21    global _rich_text
 22    if _rich_text is not None:
 23        return _rich_text
 24
 25    from meerschaum.utils.packages import import_rich, attempt_import
 26    _ = import_rich()
 27    _rich_text = attempt_import('rich.text', lazy=False)
 28    return _rich_text
 29
 30
 31def dprint(
 32    msg: str,
 33    leader: bool = True,
 34    timestamp: bool = True,
 35    package: bool = True,
 36    color: Optional[Union[str, List[str]]] = None,
 37    attrs: Optional[List[str]] = None,
 38    nopretty: bool = False,
 39    _progress: Optional['rich.progress.Progress'] = None,
 40    _task: Optional[int] = None,
 41    **kw
 42) -> None:
 43    """Print a debug message."""
 44    import meerschaum.config.paths as paths
 45    if attrs is None:
 46        attrs = []
 47
 48    if not isinstance(color, bool) and not nopretty:
 49        try:
 50            from meerschaum.utils.formatting import CHARSET, ANSI, colored
 51        except Exception as e:
 52            CHARSET, ANSI, colored = 'ascii', False, None
 53        from meerschaum.config import _config
 54        cf = _config('formatting')
 55        _color = color
 56    else:
 57        CHARSET, ANSI, colored, _color, cf = 'ascii', False, None, None, None
 58
 59    if timestamp:
 60        from meerschaum.utils.dtypes import get_current_timestamp
 61        now = get_current_timestamp('ms').replace(tzinfo=None)
 62    else:
 63        now = None
 64
 65    import logging, sys, inspect
 66    logging.basicConfig(format='%(message)s')
 67    log = logging.getLogger(__name__)
 68
 69    parent_frame = inspect.stack()[1][0]
 70    parent_info = inspect.getframeinfo(parent_frame)
 71    parent_lineno = parent_info.lineno
 72    parent_globals = parent_frame.f_globals
 73    parent_package = parent_globals['__name__']
 74    msg = str(msg)
 75    premsg = ""
 76
 77    if now:
 78        premsg = now.isoformat().split('T', maxsplit=1)[-1][:-3] + (' | ' if package else ':')
 79    if package:
 80        premsg = premsg + parent_package + ':' + str(parent_lineno)
 81    if premsg:
 82        premsg += "\n"
 83    if leader and cf is not None:
 84        try:
 85            debug_leader = cf['formatting']['debug'][CHARSET]['icon'] if cf is not None else ''
 86        except KeyError:
 87            print(
 88                "Failed to load config. " +
 89                "Please delete the following directories and restart Meerschaum:"
 90            )
 91            for p in [paths.CONFIG_DIR_PATH, paths.PERMANENT_PATCH_DIR_PATH]:
 92                print('  - ' + str(p))
 93            debug_leader = ''
 94        premsg = ' ' + debug_leader + ' ' + premsg
 95
 96    if ANSI:
 97        if _color is not None:
 98            if isinstance(_color, str):
 99                _color = [_color]
100        else:
101            if cf is not None and not nopretty:
102                try:
103                    _color = cf['formatting']['debug']['ansi']['rich'] if cf is not None else {}
104                except KeyError:
105                    _color = {}
106            else:
107                _color = {}
108        if colored is not None:
109            premsg = colored(premsg, **_color)
110
111    if _progress is not None:
112        rich_text = _import_rich_text_for_dprint()
113        text = rich_text.Text.from_ansi(premsg + msg)
114        _progress.console.log(text)
115    else:
116        print(premsg + msg)
117
118
119def _checkpoint(
120    _progress: Optional['rich.progress.Progress'] = None,
121    _task: Optional[int] = None,
122    _total: Optional[int] = None,
123    **kw
124) -> None:
125    """If the `_progress` and `_task` objects are provided, increment the task by one step.
126    If `_total` is provided, update the total instead.
127    """
128    if _progress is not None and _task is not None:
129        _kw = {'total': _total} if _total is not None else {'advance': 1}
130        _progress.update(_task, **_kw)
131
132
133def trace(browser: bool = True):
134    """
135    Open a web-based debugger to trace the execution of the program.
136    """
137    from meerschaum.utils.packages import attempt_import
138    heartrate = attempt_import('heartrate')
139    heartrate.trace(files=heartrate.files.all, browser=browser)