easy-ml/init/__init__.py

# -*- coding: utf-8 -*-
"""
@author: yq
@time: 2024/10/31
@desc: 一些资源初始化
"""
import os
import sys
import threading

import matplotlib
from contextvars import ContextVar

from config import BaseConfig

if BaseConfig.java_home is not None:
    java_home = BaseConfig.java_home
    if os.path.basename(java_home) != "bin":
        java_home = os.path.join(java_home, 'bin')
    os.environ['PATH'] = f"{os.environ['PATH']}:{java_home}"

matplotlib.use('Agg')

import matplotlib.pyplot as plt

__all__ = ['init', 'warning_ignore', "context"]


class Context:
    def __init__(self):
        # 上下文，适合notebook下单个用户
        self._instance_lock = threading.Lock()
        self.context = {}

    def set(self, k: str, data: object):
        with self._instance_lock:
            self.context.update({k: data})

    def get(self, k: str):
        return self.context.get(k, None)

    def set_filter_info(self, key, overview, detail=None):
        data = {"overview": overview, "detail": detail}
        self.set(key, data)


class ContexThreading:
    def __init__(self):
        # 上下文，web下多用户需要线程隔离，notebook下会失效
        self.context = ContextVar('context')
        self.context.set({})

    def set(self, k: str, data: object):
        context_map: dict = self.context.get()
        context_map.update({k: data})
        self.context.set(context_map)

    def get(self, k: str):
        context_map: dict = self.context.get()
        return context_map.get(k, None)

    def set_filter_info(self, key, overview, detail=None):
        data = {"overview": overview, "detail": detail}
        self.set(key, data)


if BaseConfig.run_env == "jupyter":
    context = Context()
else:
    context = ContexThreading()


def init():
    plt.rcParams['font.sans-serif'] = ['SimHei']  # 设置支持中文的字体
    plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题
    plt.rcParams['figure.figsize'] = (8, 8)
    plt.rcParams['figure.max_open_warning'] = 1000
    # plt.ioff()


def warning_ignore():
    import warnings

    # warnings.simplefilter(action="ignore", category=RuntimeWarning)
    # warnings.simplefilter(action="ignore", category=UserWarning)
    warnings.simplefilter(action="ignore", category=FutureWarning)

    warnings.filterwarnings(action="ignore", module="matplotlib")
    warnings.filterwarnings(action="ignore", module="dataframe_image")
    warnings.filterwarnings(action="ignore", module="pandas")
    warnings.filterwarnings(action="ignore", module="scorecardpy")

    if "3.6" in sys.version:
        from pandas.core.common import SettingWithCopyWarning
        warnings.simplefilter(action="ignore", category=SettingWithCopyWarning)

    if "3.10" in sys.version:
        from pandas.errors import SettingWithCopyWarning
        warnings.simplefilter(action="ignore", category=SettingWithCopyWarning)


if __name__ == "__main__":
    pass