constants.py

Go to the documentation of this file.

""" 
# +==== BEGIN rotary_logger =================+
# LOGO: 
# ..........####...####..........
# ......###.....#.#########......
# ....##........#.###########....
# ...#..........#.############...
# ...#..........#.#####.######...
# ..#.....##....#.###..#...####..
# .#.....#.##...#.##..##########.
# #.....##########....##...######
# #.....#...##..#.##..####.######
# .#...##....##.#.##..###..#####.
# ..#.##......#.#.####...######..
# ..#...........#.#############..
# ..#...........#.#############..
# ...##.........#.############...
# ......#.......#.#########......
# .......#......#.########.......
# .........#####...#####.........
# /STOP
# PROJECT: rotary_logger
# FILE: constants.py
# CREATION DATE: 29-10-2025
# LAST Modified: 12:24:31 27-03-2026
# DESCRIPTION: 
# A module that provides a universal python light on iops way of logging to files your program execution.
# /STOP
# COPYRIGHT: (c) Asperguide
# PURPOSE: This is the file that contains the values that are not meant to change, this is regardless of the downstream code
# // AR
# +==== END rotary_logger =================+
"""
 
import os
import sys
from enum import Enum
from pathlib import Path
from dataclasses import dataclass, field
from typing import IO, Optional, Dict, Any, Final, TYPE_CHECKING
from io import TextIOWrapper
 
if TYPE_CHECKING:
    from .rotary_logger_cls import FileInstance
 
# Stream redirection
MIM_STDOUT = sys.stdout
MIM_STDERR = sys.stderr
MIM_STDIN = sys.stdin
sys.stdout = MIM_STDOUT
sys.stdin = MIM_STDIN
sys.stderr = MIM_STDERR
 
# Enable unbuffered output to ensure logs are captured immediately when redirected to files
# This is critical when stdout/stderr are redirected (e.g., to a file)
if not sys.flags.optimize or 'PYTHONUNBUFFERED' not in os.environ:
    os.environ['PYTHONUNBUFFERED'] = '1'
    sys.stdout.reconfigure(line_buffering=True)  # type: ignore
    sys.stderr.reconfigure(line_buffering=True)  # type: ignore
 
IS_A_TTY: bool = sys.stdout.isatty()
IS_PIPE: bool = not IS_A_TTY
 
MODULE_NAME: str = "[Rotary Logger]"
 
ERROR: int = 1
SUCCESS: int = 0
 
SPACE: str = " "
 
B1: int = 1  # bytes
KB1: int = B1 * 1024  # bytes
MB1: int = KB1 * KB1  # bytes
GB1: int = MB1 * KB1  # bytes
TB1: int = GB1 * KB1  # bytes
 
STDIN: str = "stdin"
STDOUT: str = "stdout"
STDERR: str = "stderr"
STDUNKNOWN: str = "stdunknown"
 
 
class StdMode(Enum):
    """Enumeration of the standard stream identifiers.
 
    Used throughout the package to distinguish which standard stream a
    TeeStream instance is wrapping, and to look up the correct folder
    name and log prefix for that stream.
    """
    STDIN = STDIN
    STDOUT = STDOUT
    STDERR = STDERR
    STDUNKNOWN = STDUNKNOWN
 
 
LOG_FOLDER_BASE_NAME: str = "logs"
 
FOLDER_STDOUT: str = "stdout"
FOLDER_STDERR: str = "stderr"
FOLDER_STDIN: str = "stdin"
FOLDER_STDUNKNOWN: str = "std_unknown"
 
CORRECT_FOLDER: Dict[StdMode, str] = {
    StdMode.STDIN: FOLDER_STDIN,
    StdMode.STDOUT: FOLDER_STDOUT,
    StdMode.STDERR: FOLDER_STDERR,
    StdMode.STDUNKNOWN: FOLDER_STDUNKNOWN,
}
 
BUFFER_FLUSH_SIZE: int = 8 * KB1  # flush every 8 KB
 
DEFAULT_ENCODING: str = "utf-8"
DEFAULT_LOG_MAX_FILE_SIZE: int = 2 * GB1  # MB ~ 2 GB
DEFAULT_LOG_BUFFER_FLUSH_SIZE: int = BUFFER_FLUSH_SIZE
DEFAULT_LOG_FOLDER: Path = Path(__file__).parent / LOG_FOLDER_BASE_NAME
 
 
ERROR_MODE_WARN: str = "Warn"
ERROR_MODE_WARN_NO_PIPE: str = "Warn No pipe"
ERROR_MODE_EXIT: str = "Exit"
ERROR_MODE_EXIT_NO_PIPE: str = "Exit No pipe"
 
 
class ErrorMode(Enum):
    """Tee error handling policy enumeration.
 
    Members represent different policies for handling broken-pipe
    situations (e.g. warn, exit, or variants for when stdout/stderr
    is a pipe).
    """
    WARN = ERROR_MODE_WARN
    WARN_NO_PIPE = ERROR_MODE_WARN_NO_PIPE
    EXIT = ERROR_MODE_EXIT
    EXIT_NO_PIPE = ERROR_MODE_EXIT_NO_PIPE
 
 
FILE_LOG_DATE_FORMAT: str = "%Y_%m_%dT%Hh%Mm%Ss"
 
 
@dataclass
class FileInfo:
    """Container for an open log file.
 
    Attributes:
        path: the Path to the file on disk.
        descriptor: the open file object (text mode) or None.
        written_bytes: number of bytes already written to the file.
    """
    path: Optional[Path] = None
    descriptor: Optional[IO[Any]] = None
    written_bytes: int = 0
 
 
@dataclass
class Prefix:
    """Flags describing which streams should be prefixed when mirrored.
 
    Each flag is a boolean indicating whether the corresponding
    standard stream (stdin/stdout/stderr) should receive a textual
    prefix when written to disk.
    """
    std_in: bool = False
    std_out: bool = False
    std_err: bool = False
 
 
BROKEN_PIPE_ERROR: str = f"{MODULE_NAME} Broken pipe on stdout"
 
PREFIX_STDOUT: str = "[STDOUT]"
PREFIX_STDERR: str = "[STDERR]"
PREFIX_STDIN: str = "[STDIN]"
PREFIX_STDUNKNOWN: str = "[STDUNKNOWN]"
CORRECT_PREFIX: Dict[StdMode, str] = {
    StdMode.STDIN: PREFIX_STDIN,
    StdMode.STDOUT: PREFIX_STDOUT,
    StdMode.STDERR: PREFIX_STDERR,
    StdMode.STDUNKNOWN: PREFIX_STDUNKNOWN
}
 
PREFIX_FUNCTION_CALL_EMPTY: str = ""
PREFIX_FUNCTION_CALL_WRITE: str = "[WRITE]"
PREFIX_FUNCTION_CALL_WRITELINES: str = "[WRITELINES]"
PREFIX_FUNCTION_CALL_FLUSH: str = "[FLUSH]"
PREFIX_FUNCTION_CALL_READ: str = "[READ]"
PREFIX_FUNCTION_CALL_READLINE: str = "[READLINE]"
PREFIX_FUNCTION_CALL_READLINES: str = "[READLINES]"
 
 
class PrefixFunctionCall(Enum):
    """Optional per-call prefix appended to log entries for precise tracing.
 
    When enabled, each log entry is tagged with the name of the stream
    method that produced it (e.g. [WRITE], [READLINE]), making it easy
    to distinguish writes from reads in a shared log file.
    """
    EMPTY = PREFIX_FUNCTION_CALL_EMPTY
    WRITE = PREFIX_FUNCTION_CALL_WRITE
    WRITELINES = PREFIX_FUNCTION_CALL_WRITELINES
    FLUSH = PREFIX_FUNCTION_CALL_FLUSH
    READ = PREFIX_FUNCTION_CALL_READ
    READLINE = PREFIX_FUNCTION_CALL_READLINE
    READLINES = PREFIX_FUNCTION_CALL_READLINES
 
 
LOG_TO_FILE_ENV: bool = os.environ.get(
    "LOG_TO_FILE",
    "true"
).lower() in ("1", "true", "yes")
 
RAW_LOG_FOLDER_ENV: str = os.environ.get(
    "LOG_FOLDER_NAME",
    str(DEFAULT_LOG_FOLDER)
)
 
 
@dataclass
class FileStreamInstances:
    """Container for the FileInstance objects corresponding to each stream.
 
    Attributes:
        stdout: the FileInstance for stdout, or None if not capturing.
        stderr: the FileInstance for stderr, or None if not capturing.
        stdin: the FileInstance for stdin, or None if not capturing.
    """
    stdout: Optional["FileInstance"] = None
    stderr: Optional["FileInstance"] = None
    stdin: Optional["FileInstance"] = None
    stdunknown: Optional["FileInstance"] = None
    merged_streams: Dict[StdMode, bool] = field(default_factory=lambda: {
        StdMode.STDIN: False,
        StdMode.STDOUT: False,
        StdMode.STDERR: False,
        StdMode.STDUNKNOWN: False
    })
 
 
@dataclass(frozen=True)
class LogToggle:
    """
    The settings for the logger to know if it can output the line for the given level.
    """
    program_log: bool = True
    success: bool = True
    info: bool = True
    warning: bool = True
    error: bool = True
    critical: bool = True
    debug: bool = True
 
 
RAW_STDIN: Final[Optional[TextIOWrapper]] = sys.__stdin__
RAW_STDOUT: Final[Optional[TextIOWrapper]] = sys.__stdout__
RAW_STDERR: Final[Optional[TextIOWrapper]] = sys.__stderr__