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    if attrs is None:
 45        attrs = []
 46    if not isinstance(color, bool) and not nopretty:
 47        try:
 48            from meerschaum.utils.formatting import CHARSET, ANSI, colored
 49        except Exception as e:
 50            CHARSET, ANSI, colored = 'ascii', False, None
 51        from meerschaum.config._paths import CONFIG_DIR_PATH, PERMANENT_PATCH_DIR_PATH
 52        from meerschaum.config import _config
 53        cf = _config('formatting')
 54        _color = color
 55    else:
 56        CHARSET, ANSI, colored, _color, cf = 'ascii', False, None, None, None
 57
 58    if timestamp:
 59        from meerschaum.utils.dtypes import get_current_timestamp
 60        now = get_current_timestamp('ms').replace(tzinfo=None)
 61    else:
 62        now = None
 63
 64    import logging, sys, inspect
 65    logging.basicConfig(format='%(message)s')
 66    log = logging.getLogger(__name__)
 67
 68    parent_frame = inspect.stack()[1][0]
 69    parent_info = inspect.getframeinfo(parent_frame)
 70    parent_lineno = parent_info.lineno
 71    parent_globals = parent_frame.f_globals
 72    parent_package = parent_globals['__name__']
 73    msg = str(msg)
 74    premsg = ""
 75
 76    if now:
 77        premsg = now.isoformat().split('T', maxsplit=1)[-1][:-3] + (' | ' if package else ':')
 78    if package:
 79        premsg = premsg + parent_package + ':' + str(parent_lineno)
 80    if premsg:
 81        premsg += "\n"
 82    if leader and cf is not None:
 83        try:
 84            debug_leader = cf['formatting']['debug'][CHARSET]['icon'] if cf is not None else ''
 85        except KeyError:
 86            print(
 87                "Failed to load config. " +
 88                "Please delete the following directories and restart Meerschaum:"
 89            )
 90            for p in [CONFIG_DIR_PATH, PERMANENT_PATCH_DIR_PATH]:
 91                print('  - ' + str(p))
 92            debug_leader = ''
 93        premsg = ' ' + debug_leader + ' ' + premsg
 94
 95    if ANSI:
 96        if _color is not None:
 97            if isinstance(_color, str):
 98                _color = [_color]
 99        else:
100            if cf is not None and not nopretty:
101                try:
102                    _color = cf['formatting']['debug']['ansi']['rich'] if cf is not None else {}
103                except KeyError:
104                    _color = {}
105            else:
106                _color = {}
107        if colored is not None:
108            premsg = colored(premsg, **_color)
109
110    if _progress is not None:
111        rich_text = _import_rich_text_for_dprint()
112        text = rich_text.Text.from_ansi(premsg + msg)
113        _progress.console.log(text)
114    else:
115        print(premsg + msg)
116
117
118def _checkpoint(
119    _progress: Optional['rich.progress.Progress'] = None,
120    _task: Optional[int] = None,
121    _total: Optional[int] = None,
122    **kw
123) -> None:
124    """If the `_progress` and `_task` objects are provided, increment the task by one step.
125    If `_total` is provided, update the total instead.
126    """
127    if _progress is not None and _task is not None:
128        _kw = {'total': _total} if _total is not None else {'advance': 1}
129        _progress.update(_task, **_kw)
130
131
132def trace(browser: bool = True):
133    """
134    Open a web-based debugger to trace the execution of the program.
135    """
136    from meerschaum.utils.packages import attempt_import
137    heartrate = attempt_import('heartrate')
138    heartrate.trace(files=heartrate.files.all, browser=browser)