logpyros.py 9.66 KB
import os
import math
import tempfile
from io import StringIO
import sys

# --- update the path of Python in case to test this file alone
py_path = os.sys.path
py_pwd = os.path.normpath(os.getcwd() + "/../../../..")
if (py_pwd not in py_path):
    (os.sys.path).append(py_pwd)

# --- import celestial mechanics
import src.core.celme as celme

# --- DJANGO IMPORT ---
#from src.core.pyros_django.common.models import AgentLogs

class LogPyros:

    # === Level of log
    LOG_LEVEL_INFO = 0
    LOG_LEVEL_WARNING = -1
    LOG_LEVEL_ERROR = -2
    LOG_LEVEL_DEBUG = 1
    LOG_LEVEL_DEBUG2 = 2
    
    # === Constants
    NO_ERROR = 0

    ERR_ALIAS_NAME_NOT_DEFINED = 10
    
    ERR_FILE_NOT_EXISTS = 101
    ERR_PATH_NOT_EXISTS = 102
    ERR_EMPTY_FILENAME = 103
    ERR_EMPTY_PATHNAME = 104

    # === Private variables
    _last_errno = NO_ERROR
        
    _path_data = ''
    _agent_alias = ''
    _path_data_log  = ''
    _path_wwwpyros = ''
    
    _date = None
    _home = None
    _noon_hour = 12
    
# =====================================================================
# =====================================================================
# Private methods
# =====================================================================
# =====================================================================

    def _mkdir_path_log(self):
        # ---
        if self._agent_alias == "":
            self._last_errno = self.ERR_EMPTY_FILENAME
        else:
            self._path_www_log = self._path_wwwpyros + "/logs/" + self._agent_alias
            os.mkdir(self._path_www_log)
        # ---
        if self._path_data == "":
            self._last_errno = self.ERR_EMPTY_PATHNAME
        else:
            self._path_data_log = self._path_data + "/logs/" + self._agent_alias
            os.mkdir(self._path_data_log)
        
    def _jd2datedigit(self, jd):
        """ Convert a julian day into a string of YYYYMMDD
        """
        datetmp = self._date.date(jd);
        datedigit = (datetmp.digits(0))[0:8]
        return datedigit

    def _date2night(self, date):
        # night is the truncated part of the date of the previous noon 
        self._date.date(date)
        jd = self._date.jd()
        jd0 = math.floor(jd) - self._noon_hour/24
        d = celme.Date(jd0)
        djd = jd-jd0
        # print(f"jd0 = {d.iso()} djd={djd}")
        if djd>=1:
            jd0 += 1
        djd = jd-jd0
        if djd>=1:
            jd0 += 1
        d = celme.Date(jd0)
        night = d.digits(0)[0:8]
        return night

    def _night2dates(self, night):
        """
        jd_noon0 = previous noon
        jd_midnight = midnight instant
        jd_noon1 = next noon
        """
        d = celme.Date(night)
        d_noon0 = d + self._noon_hour/24
        jd_noon0 = d_noon0.jd()
        jd_midnight = jd_noon0 + 0.5
        jd_noon1 = jd_noon0 + 1.0
        return jd_noon0, jd_midnight, jd_noon1
    
    def _convert_args2str(self, *args, **kwargs):
        """
        return a string as the result of a print method
        """
        bak = sys.stdout # on sauvegarde l'ancien stdout
        result = StringIO()
        sys.stdout = result
        print (*args, **kwargs,end='')
        sys.stdout = bak # on restore stdout
        return result.getvalue()

# =====================================================================
# =====================================================================
# Private methods getter/setter
# =====================================================================
# =====================================================================

    def _set_debug_level(self, level:str):
        if type(level)=="bool":
            if level==True:
                level = 1
            else:
                level = 0
        self._debug_level = level
        
    def _get_debug_level(self):
        return self._debug_level
        
    def _set_agent_alias(self, agent_alias:str):
        if agent_alias=="":
            self._last_errno = self.ERR_ALIAS_NAME_NOT_DEFINED
            raise Exception(f"Agent alias must be not be empty string.")
        self._agent_alias = agent_alias
        return self.NO_ERROR

    def _get_agent_alias(self):
        return self._agent_alias

    def _set_path_data(self, path_data:str):
        if not os.path.exists(path_data):
            self._last_errno = self.ERR_PATH_NOT_EXISTS
            raise Exception(f"Path '{path_data}' for data does not exists. Create it first manually.")
        self._path_data = path_data
        return self.NO_ERROR

    def _get_path_data(self):
        return self._path_data

    def _set_path_wwwpyros(self, path_wwwpyros:str):
        if not os.path.exists(path_wwwpyros):
            self._path_wwwpyros = self.ERR_PATH_NOT_EXISTS
            raise Exception(f"Path '{path_wwwpyros}' for wwwpyros does not exists. Create it first manually.")
        self._path_wwwpyros = path_wwwpyros
        return self.NO_ERROR

    def _get_path_wwwpyros(self):
        return self._path_wwwpyros

    def _set_home(self, home:str):
        self._home.home(home)
        longitude = self._home.longitude
        self._noon_hour = (180 - longitude)/15.0
        # - place noon_hour in the range [0 24[
        if self._noon_hour < 0:
            self._noon_hour += 24
        return self.NO_ERROR

    def _get_home(self):
        return self._home.gps
    
    def _set_logtable(self, logtable):
        self._logtable = logtable
        return self.NO_ERROR

    def _get_logtable(self):
        return self._logtable        
    
# =====================================================================
# =====================================================================
# Property methods
# =====================================================================
# =====================================================================

    debug_level = property(_get_debug_level, _set_debug_level)
    
    agent_alias = property(_get_agent_alias, _set_agent_alias)
    
    path_data = property(_get_path_data, _set_path_data)
    
    path_wwwpyros = property(_get_path_wwwpyros, _set_path_wwwpyros)
    
    home = property(_get_home, _set_home)

    logtable = property(_get_logtable, _set_logtable)
        
# =====================================================================
# =====================================================================
# Methods for users
# =====================================================================
# =====================================================================
        
    def print(self, *args, **kwargs):
        msg = self._convert_args2str(*args, **kwargs)
        # --- prepare the super_msg
        if msg=="":
            super_msg = msg
        else:
            super_msg = f"({self._agent_alias}) {msg}"
        # --- classical print
        print(super_msg)
        # --- no more if the agent name is not defined
        if self._agent_alias=="":
            return
        # --- 
        self.file(msg)

    def printd(self, *args, **kwargs):
        if self.debug_level > 0:
            self.print(*args, **kwargs)
            
    def file(self, *args, **kwargs):
        msg = self._convert_args2str(*args, **kwargs)
        # --- no more if the agent name is not defined
        if self._agent_alias=="":
            return
        # --- check the path to write the log file
        path = os.path.normpath(self._path_data + os.sep + "logs" + os.sep + self._agent_alias)
        if not os.path.exists(path):
            try:
                os.makedirs(path)
            except:            
                p = f"Cannot create path {path}"
                raise Exception(p)
        # --- compute the current night digital date
        night = self._date2night("now")
        # --- build the log file name
        file = os.path.normpath(path + os.sep + self._agent_alias + '_' + night + '.log')
        # --- prepare the super_msg
        self._date.date("NOW")
        super_msg = f"{self._date.iso()} {msg}"
        # --- write super_msg in the log file
        with open(file,'a') as fic:
            fic.write(super_msg+"\n")

    def db(self, *args, **kwargs):
        msg = self._convert_args2str(*args, **kwargs)
        self._db = self.logtable.objects.create(name=self._agent_alias, message=msg)
    
# =====================================================================
# =====================================================================
# Special methods
# =====================================================================
# =====================================================================
        
    def __init__(self, agent_alias:str = "", path_data:str = "", path_wwwpyros:str = ""):
        self._last_errno = self.NO_ERROR
        # ---
        if agent_alias != "":
            self.agent_alias = agent_alias
        # ---
        if path_data != "":
            self.path_data = path_data
        else:
            self.path_data = tempfile.gettempdir()
        # ---
        if path_wwwpyros != "":
            self.path_wwwpyros = path_wwwpyros
        self._date = celme.Date()
        self._home = celme.Home()
        self._noon_hour = 12
        self.logtable = None

# =====================================================================
# =====================================================================
# Test if main
# =====================================================================
# =====================================================================
    
if __name__ == "__main__":
    agent_alias = "test"
    path_data = "c:/srv/work/pyros"
    log = LogPyros()
    log.agent_alias = agent_alias
    log.path_data = path_data
    log.home = "GPS 2 E 43 148"
    
    a = 2
    b = 'tutu'
    c = f"a={a} b={b}"
    log.print(c)
    
    log.print("toto","titi")
    
    #### Penser au mode debug