Misc Test

cancerlog.py

# =================================================================================
# Name        : cancerlog.py
# Author      : Ali Firooz
# Version     : 0.0.1
# Copyright   : Copyright © 2023 Ali Firooz. All Rights Reserved.
# Description : based on: https://ankitbko.github.io/blog/2021/04/logging-in-python/
# Usage       :
# =================================================================================
# TODO:
#   1. Add functionality to log other levels (info, warning, error, critical)
#   2. Figure out how to use MyLogger class. Or better yet,
#      how to tell each decorated function to use a specific logging level.
#   3. Add functionality to log to console.
#   4. finish the "Usage" section.


import logging
import varname as vn
import functools
import inspect
from os import path
from typing import Union


def dprint(*args, **kwargs):
    """
    Use this instead of print() to have variable print in a nice way.
    :param new_line: If you want an extra new line between each print.
    :param args: The variable(s) you want to debug.
    :param kwargs: If you have a function applied to a variable (like len()), use this.
    :return: None
    """
    if 'new_line' in kwargs.keys():
        new_line = kwargs.pop('new_line')
    else:
        new_line = False

    if 'end' in kwargs.keys():
        end = kwargs.pop('end')
    else:
        end = '\n'

    if 'debug' in kwargs.keys():
        debug = kwargs.pop('debug')
    else:
        debug = False

    # try:
    allvars = vn.argname('args')
    for t_key, t_val in zip(allvars, args):
        print(f"{t_key}: {t_val}")
        print() if new_line else None

    for t_key, t_val in kwargs.items():
        print(f"{t_key}: {t_val}")
        print() if new_line else None

    print(end=end)

    if debug:
        print('------------------')
        print(f"args: \n{args}\n\n"
              f"kwargs: \n{kwargs}")
        print('------------------')

    # except Exception as e:
    #     print('------------------')
    #     print(args)
    #     print(kwargs)
    #     print(e)
    #     print('------------------')


class MyLogger:
    # Constructor for initializing the logging system with a default logging level of DEBUG.
    def __init__(self, level=logging.DEBUG, filename="runtime.log"):
        self.debugf = '%(asctime)s - %(levelname)s - %(message)s'
        if level == logging.DEBUG:
            logging.basicConfig(level=level, format=self.debugf, filename=filename)

    # Method for getting a logger by name.
    def get_logger(self, name=None):
        return logging.getLogger(name)


# Function to get a default logger using the MyLogger class.
def get_default_logger():
    return MyLogger().get_logger()


def _getprefix(func=None) -> str:
    # Get the filename where the decorated function is defined.
    try:
        function_filename = path.split(inspect.getfile(func))[1]
        filename = f"{function_filename} -"
    except Exception:
        filename: str = "file:UNKNOWN -"

    # get class name
    # todo: need to check if this works
    try:
        clsname = func.__qualname__
        clsname = f"cls:{clsname} -"
    except Exception:
        clsname = "cls/func:UNKNOWN -"

    return f"cls/func:{clsname} {filename}"


# Decorator function logme that can be used to log function calls and exceptions.
def logme(_func=None, *, my_logger: Union[MyLogger, logging.Logger] = None):
    # Decorator function that wraps another function with logging functionality.
    def decorator_log(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Get the prefix (class_name.function_name - file_name) for the log message.
            prefix = _getprefix(func)

            # Get the default logger if my_logger is not provided.
            logger = get_default_logger()
            try:
                if my_logger is None:
                    # capture first arg to check for `self`
                    first_args = next(iter(args), None)
                    # does kwargs have any logger
                    logger_params = [
                                        x
                                        for x in kwargs.values()
                                        if isinstance(x, logging.Logger) or isinstance(x, MyLogger)
                                    ] + [
                                        # does args have any logger
                                        x
                                        for x in args
                                        if isinstance(x, logging.Logger) or isinstance(x, MyLogger)
                                    ]
                    # is first argument `self`
                    if hasattr(first_args, "__dict__"):
                        logger_params = logger_params + [
                            # does class (dict) members have any logger
                            x for x in first_args.__dict__.values()
                            if isinstance(x, logging.Logger) or isinstance(x, MyLogger)
                        ]
                    # get the next/first/default logger
                    h_logger = next(iter(logger_params), MyLogger())
                else:
                    # logger is passed explicitly to the decorator
                    h_logger = my_logger

                if isinstance(h_logger, MyLogger):
                    logger = h_logger.get_logger(func.__name__)
                else:
                    logger = h_logger

                # Prepare a signature of the function call with its arguments and keyword arguments.
                args_repr = [repr(a) for a in args]
                kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
                signature = ", ".join(args_repr + kwargs_repr)

                # Log the function call with its signature.
                logger.debug(f"{prefix} Called with args {signature}")
            except Exception:
                pass

            try:
                # Call the original function and capture its result.
                result = func(*args, **kwargs)
                return result
            except Exception as e:
                # Log an exception if one is raised during the function call and re-raise it.
                logger.exception(f"{prefix} Exception raised. exception: \n{str(e)}")
                raise e

        return wrapper

    # Check if this decorator is used with or without parentheses.
    if _func is None:
        # Return the decorator function when no arguments are provided.
        return decorator_log
    else:
        # Return the decorated function when arguments are provided.
        return decorator_log(_func)


if __name__ == '__main__':
    pass

orjson-msgspec-tobyte-speed.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "id": "initial_id",
   "metadata": {
    "collapsed": true,
    "ExecuteTime": {
     "end_time": "2024-04-25T16:27:22.686058Z",
     "start_time": "2024-04-25T16:26:53.226698Z"
    }
   },
   "source": [
    "# testing orjson vs msgspec vs numpy tobyte() speed\n",
    "\n",
    "import orjson as json\n",
    "import numpy as np\n",
    "import sys\n",
    "import msgspec\n",
    "\n",
    "spec_peaks = np.random.rand(1_000_000, 2)\n",
    "encoder = msgspec.json.Encoder()\n",
    "decoder = msgspec.json.Decoder()\n",
    "\n",
    "orjson_b = json.dumps(spec_peaks.tolist())\n",
    "msgspec_b = encoder.encode(spec_peaks.tolist())\n",
    "numpy_b = spec_peaks.tobytes()\n",
    "\n",
    "original_shape = spec_peaks.shape\n",
    "np_decoded = np.frombuffer(numpy_b, dtype=np.float64).reshape(original_shape)\n",
    "\n",
    "print('testing native numpy to bytes')\n",
    "print('Encoding:')\n",
    "%timeit spec_peaks.tobytes()\n",
    "print('Decoding:')\n",
    "%timeit np.frombuffer(numpy_b, dtype=np.float64).reshape(-1, 2)\n",
    "print('---------------------------------')\n",
    "\n",
    "print(\"testing orjson\")\n",
    "# b = json.dumps(spec_peaks.tolist(), option=json.OPT_INDENT_2).decode()\n",
    "print('Encoding:')\n",
    "%timeit json.dumps(spec_peaks.tolist())\n",
    "print('Decoding:')\n",
    "%timeit np.array(json.loads(orjson_b))\n",
    "print('---------------------------------')\n",
    "\n",
    "print(\"testing msgspec\")\n",
    "print('Encoding:')\n",
    "%timeit encoder.encode(spec_peaks.tolist())\n",
    "print('Decoding:')\n",
    "%timeit decoder.decode(msgspec_b)\n",
    "print('---------------------------------')\n",
    "\n",
    "print(f\"numpy b   = {sys.getsizeof(numpy_b) / (1024 * 1024):.2f} MB\")\n",
    "print(f\"orjson b  = {sys.getsizeof(orjson_b) / (1024 * 1024):.2f} MB\")\n",
    "print(f\"msgspec b = {sys.getsizeof(msgspec_b) / (1024 * 1024):.2f} MB\")\n",
    "\n",
    "print(f\"np_b == spec_peaks -> {(np_decoded == spec_peaks).sum() / 2 == len(spec_peaks)}\")\n"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "testing native numpy to bytes\n",
      "Encoding:\n",
      "1.44 ms ± 10.5 µs per loop (mean ± std. dev. of 7 runs, 1,000 loops each)\n",
      "Decoding:\n",
      "771 ns ± 38.4 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each)\n",
      "---------------------------------\n",
      "testing orjson\n",
      "Encoding:\n",
      "260 ms ± 3.92 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n",
      "Decoding:\n",
      "471 ms ± 5.33 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n",
      "---------------------------------\n",
      "testing msgspec\n",
      "Encoding:\n",
      "263 ms ± 4.08 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n",
      "Decoding:\n",
      "226 ms ± 2.02 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n",
      "---------------------------------\n",
      "numpy b   = 15.26 MB\n",
      "orjson b  = 38.66 MB\n",
      "msgspec b = 38.66 MB\n",
      "np_b == spec_peaks -> True\n"
     ]
    }
   ],
   "execution_count": 1
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}

py-logger.py

from cancerlog import logme


@logme
def factorial(n):
    if n == 0:
        return 1
    else:
        return n * factorial(n - 1)


class something:
    def __init__(self):
        pass

    @logme
    def clsfunc(self):
        print("clsfunc is running")


if __name__ == '__main__':
    # Input a number
    num = 5

    if num < 0:
        print("Factorial is not defined for negative numbers.")
    else:
        result = factorial(num)
        print(f"The factorial of {num} is {result}")

    s = something()
    s.clsfunc()