src/core/tools.py

import functools
import hashlib
import logging

import numpy as np
import pandas as pd
from flask import session

from src.core.config import get_settings

settings = get_settings()


logger = logging.getLogger(__name__)


def formatter(df, col_format):
    transform = False
    if isinstance(df, (float, int)):
        df = pd.Series([df])
        transform = True

    if col_format == "CURRENCY":
        output = df.apply(lambda x: f"{x:,.0f} €")
    elif col_format == "MINS":
        output = df.apply(lambda x: f"{int(x):02}min {(x-int(x)) * 60:02,.0f}s")
    elif col_format == "PERCENT":
        output = df.apply(lambda x: f"{x:.0%}")
    elif col_format == "PERCENT+":
        output = df.apply(lambda x: f"{x:+,.0%}")
    elif col_format == "PT+":
        output = df.apply(lambda x: f"{x*100:+.0f} pt")
    elif col_format == "INT":
        output = df.apply(lambda x: f"{x:,.0f}")
    elif col_format == "NUMBER":
        output = df.apply(lambda x: f"{x:,.0f}".replace(",", " "))
    elif col_format == "FLOAT":
        output = df.apply(lambda x: f"{x:,.1f}")
    elif col_format == "DATE":
        output = df.apply(lambda x: f"{x:%d %b %Y}")
    elif col_format == "DATE_TIME":
        output = df.apply(lambda x: f"{x:%d %b %Y %H:%M}")
    elif col_format == "TIME":
        output = df.apply(
            lambda x: "{:02.0f}h {:02.0f}min".format(*divmod(float(x) * 60, 60))
        )
    elif col_format == "HOURS":
        output = df.apply(lambda x: f"{x:,.0f} h")
    elif col_format == "STRING":
        output = df.apply(lambda x: f"{x}")
    elif col_format == "YEARS":
        output = (df / 365).apply(lambda x: f"{x:.1f} years")
    elif col_format == "MONTHS":
        output = (df / (365 / 12)).apply(lambda x: f"{x:.0f} months")
    elif col_format == "KM":
        output = df.apply(lambda x: f"{x:.0f} km")
    elif df.dtype == "float64":
        output = df.apply(lambda x: f"{x:.1f}")
    else:
        output = df.apply(lambda x: f"{x}")
    if transform:
        output = output.item()
    return output


def compound_exp(r):
    """
    returns the result of compounding the set of returns in r
    """
    return np.expm1(np.log1p(r).sum())


def hash_single(arg):
    if isinstance(arg, pd.Series) or isinstance(arg, pd.DataFrame):
        return hashlib.sha256(
            pd.util.hash_pandas_object(arg, index=True).values
        ).hexdigest()
    elif isinstance(arg, tuple) or isinstance(arg, list):
        m = hashlib.md5()
        for s in arg:
            m.update(str(s).encode())
        return m.hexdigest()
    elif isinstance(arg, str):
        m = hashlib.md5()
        m.update(arg.encode())
        return m.hexdigest()
    elif arg is None:
        return "None"
    else:
        return hash(arg)


def hash_multiple(args, kwargs):
    hashed_args = tuple(hash_single(arg) for arg in args)
    # (0, 'bb7831021d8a3e98102cca4d329b1201a5d9dff5538a8ebb4229994ac60f6fb1')
    hashed_kwargs = tuple(hash_single(kwarg) for kwarg in kwargs.values())
    return hash_single(hashed_args + hashed_kwargs)


def get_group_user() -> str:
    if "GROUP" in session:
        return session["GROUP"]
    return None