mountlog.py 11.7 KB
import os
import math
import tempfile
from io import StringIO
import sys

path = os.path.join(os.path.dirname(__file__), '..')
if path not in sys.path:
    sys.path.append(path)
# --- celme imports
modulename = 'celme'
if modulename in dir():
    del celme
if modulename not in dir():    
    import celme

class Mountlog:
    """
    Manage logs in display, file and database.

    First, create an instance:
    log = Mountlog("test",None)

    Second, use the print methods to log:
    log.print("Something to log")
    """

    # === 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_www = ''
    
    _date = None
    _home = None
    _noon_hour = 12
    _last_lines = list()
    _nbmax_last_lines = 30

# =====================================================================
# =====================================================================
# Private methods
# =====================================================================
# =====================================================================

    def _mkdir_path_log(self):
        # ---
        if self._agent_alias == "":
            self._last_errno = self.ERR_EMPTY_FILENAME
        else:
            self._path_www = self._path_www + "/logs/" + self._agent_alias
            os.mkdir(self._path_www)
        # ---
        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).__name__=="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_www(self, path_www:str):
        if not os.path.exists(path_www):
            self._path_www = self.ERR_PATH_NOT_EXISTS
            raise Exception(f"Path '{path_www}' for www does not exists. Create it first manually.")
        self._path_www = path_www
        return self.NO_ERROR

    def _get_path_www(self):
        return self._path_www

    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_nbmax_last_lines(self, nbmax_last_lines):
        self._nbmax_last_lines = nbmax_last_lines
        return self.NO_ERROR

    def _get_nbmax_last_lines(self):
        return self._nbmax_last_lines        
    
# =====================================================================
# =====================================================================
# Property methods
# =====================================================================
# =====================================================================

    agent_alias = property(_get_agent_alias, _set_agent_alias)
    
    path_data = property(_get_path_data, _set_path_data)
    
    path_www = property(_get_path_www, _set_path_www)
    
    home = property(_get_home, _set_home)
    
# =====================================================================
# =====================================================================
# Public Property methods
# =====================================================================
# =====================================================================

    debug_level = property(_get_debug_level, _set_debug_level)
    
    nbmax_last_lines = property(_get_nbmax_last_lines, _set_nbmax_last_lines)
        
# =====================================================================
# =====================================================================
# Methods for users
# =====================================================================
# =====================================================================

    def print(self, *args, **kwargs):
        """
        This is the method to print in the console display and in a log file.
        In the console display, the message is presented as:
        (Agent_name) message
        In the log file the message is presented as:
        Date-ISO message
        """
        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):
        """
        Same as print method but only if debug level is > threshold
        """
        if self._debug_level > 0:
            self.print(*args, **kwargs)

    def file(self, *args, **kwargs):
        """
        This is the method to print in a log file.
        In the log file the message is presented as:
        Date-ISO message
        The last file is also apended with the message
        """
        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")
        # --- last lines
        self._last_lines.append(super_msg)
        n = len(self._last_lines)
        if n>self.nbmax_last_lines:
            self._last_lines = self._last_lines[n-self.nbmax_last_lines:]
        # --- build the log file name
        file = os.path.normpath(path + os.sep + self._agent_alias + '_' + 'last' + '.log')
        # --- write super_msg in the log file
        with open(file,'w') as fic:
            for line in self._last_lines:
                fic.write(line+"\n")
    
# =====================================================================
# =====================================================================
# Special methods
# =====================================================================
# =====================================================================
        
    def __init__(self, agent_alias:str, home:str="GPS 0 E 43 150", path_data:str="", path_www:str=""):
        self._last_errno = self.NO_ERROR
        # ---
        if agent_alias != "":
            self.agent_alias = agent_alias
        else:
            self.agent_alias = "Unknown_agent"
        # ---
        if path_data != "":
            self.path_data = path_data
        else:
            self.path_data = tempfile.gettempdir()
        # ---
        if path_www != "":
            self.path_www = path_www
        else:
            self.path_www = self.path_data
        # ---
        self._date = celme.Date()
        self._home = celme.Home(home)
        self._noon_hour = 12 ; # local hour corresponding to the date change
        self.nbmax_last_lines = 30

# =====================================================================
# =====================================================================
# Test if main
# =====================================================================
# =====================================================================
    
if __name__ == "__main__":
    
    cwd = os.getcwd()

    example = 1
    print("Example       = {}".format(example))
            
    if example == 1:
        # === Instance parameters
        # --- agent_alias = string added at the begining of each line of display log 
        # --- agent_alias = is the name of log files
        # --- agent_alias = is the field name of database logs
        agent_alias = "test"
        # --- agent_alias = directory where are written log files 
        path_data = cwd
        # --- home = define the hour of noon when log files are archived 
        home = "GPS 2 E 43 148"
        
        # === Instanciation
        log = Mountlog(agent_alias,home,path_data)
        
        # === Use of logs. Parameters are the same as a Python print
    
        a = 2
        b = 'tutu'
        log.print(f"a={a} b={b}")
        
        log.print(a,b)