Module stdstream_cloner
Standard Stream Cloner into a Log File.
The role of an instance of stdstream_cloner_t is to automatically clone a standard stream (among
stdout and stderr) into a log file whenever the standard function print is called. The print function turns its
input into a character string, which is then sent to a standard stream for output. If an instance of
stdstream_cloner_t has been created for the corresponding stream, it will intercept the string to
send it to the stream, as expected, and to additionally write it to the log file.
Cloning is enabled at instantiation time. Afterwards, it can be disabled and re-enabled at will by calling the
methods Disable and Enable, respectively. The methods PrintToStreamOnly and PrintToLogOnly allow to bypass the
cloning process, outputting to the standard stream alone or to the log file alone, respectively.
Character strings to be output are passed as is to the standard stream, but are processed as follows before being
written to the log file:
- Bell characters "\a" are removed;
- Formfeeds "\f" and vertical tabs "\v" are replaced with newlines "\n";
- A carriage return "\r" at the beginning or at the end of the string rolls back in the log file right after the latest newline;
- A sequence of backspaces "\b" at the beginning or at the end of the string rolls back accordingly in the log file. This should not roll back passed the latest newline character, or the log file contents might be 'messed up' until a new newline is written. However, no such check is made.
- Carriage returns and backspaces in the middle of the string are replaced with "⇦" and "←", respectively.
Minimal Example
>>> from stdstream_cloner import stdstream_cloner_t
>>> stdout_cloner = stdstream_cloner_t("stdout.log", stdstream_cloner_t.STREAM_NAME_OUT)
>>> stdout_cloner.PrintToStreamOnly("Only in Console\n") # Note the appended newline character. See class documentation.
>>> stdout_cloner.PrintToLogOnly("Only in Log File\n") # Note the appended newline character. See class documentation.
>>> print("In Both the Console and the Log File")
>>> stdout_cloner.Disable()
Expand source code
# Copyright CNRS/Inria/UCA
# Contributor(s): Eric Debreuve
#
# eric.debreuve@cnrs.fr
#
# This software is governed by the CeCILL license under French law and
# abiding by the rules of distribution of free software. You can use,
# modify and/ or redistribute the software under the terms of the CeCILL
# license as circulated by CEA, CNRS and INRIA at the following URL
# "http://www.cecill.info".
#
# As a counterpart to the access to the source code and rights to copy,
# modify and redistribute granted by the license, users are provided only
# with a limited warranty and the software's author, the holder of the
# economic rights, and the successive licensors have only limited
# liability.
#
# In this respect, the user's attention is drawn to the risks associated
# with loading, using, modifying and/or developing or reproducing the
# software by the user in light of its specific status of free software,
# that may mean that it is complicated to manipulate, and that also
# therefore means that it is reserved for developers and experienced
# professionals having in-depth computer knowledge. Users are therefore
# encouraged to load and test the software's suitability as regards their
# requirements in conditions enabling the security of their systems and/or
# data to be ensured and, more generally, to use and operate it in the
# same conditions as regards security.
#
# The fact that you are presently reading this means that you have had
# knowledge of the CeCILL license and that you accept its terms.
"""
Standard Stream Cloner into a Log File.
The role of an instance of `stdstream_cloner.stdstream_cloner_t` is to automatically clone a standard stream (among
stdout and stderr) into a log file whenever the standard function `print` is called. The `print` function turns its
input into a character string, which is then sent to a standard stream for output. If an instance of
`stdstream_cloner.stdstream_cloner_t` has been created for the corresponding stream, it will intercept the string to
send it to the stream, as expected, and to additionally write it to the log file.
Cloning is enabled at instantiation time. Afterwards, it can be disabled and re-enabled at will by calling the
methods `Disable` and `Enable`, respectively. The methods `PrintToStreamOnly` and `PrintToLogOnly` allow to bypass the
cloning process, outputting to the standard stream alone or to the log file alone, respectively.
Character strings to be output are passed as is to the standard stream, but are processed as follows before being
written to the log file:
- Bell characters "\a" are removed;
- Formfeeds "\f" and vertical tabs "\v" are replaced with newlines "\n";
- A carriage return "\r" at the beginning or at the end of the string rolls back in the log file right after the latest
newline;
- A sequence of backspaces "\b" at the beginning or at the end of the string rolls back accordingly in the log file.
This should not roll back passed the latest newline character, or the log file contents might be 'messed up' until a new
newline is written. However, no such check is made.
- Carriage returns and backspaces in the middle of the string are replaced with "⇦" and "←", respectively.
Minimal Example
---------------
>>> from stdstream_cloner import stdstream_cloner_t
>>> stdout_cloner = stdstream_cloner_t("stdout.log", stdstream_cloner_t.STREAM_NAME_OUT)
>>> stdout_cloner.PrintToStreamOnly("Only in Console\n") # Note the appended newline character. See class documentation.
>>> stdout_cloner.PrintToLogOnly("Only in Log File\n") # Note the appended newline character. See class documentation.
>>> print("In Both the Console and the Log File")
>>> stdout_cloner.Disable()
"""
import os as opsy
import pathlib as phlb
import sys as syst
from typing import Callable, Final, Optional, TextIO, Union
class stdstream_cloner_t:
"""
Standard Stream Cloner into a Log File. See the module documentation for a general description of the role of the
class.
The class implements cloning by replacing the requested standard stream in the `sys` module of the standard library
with an instance of itself, while implementing a stream-cloning method `write` and 'acquiring' the method
`sys.REQUESTED_STANDARD_STREAM.flush` where REQUESTED_STANDARD_STREAM is either `stdout` or `stderr`. It also
'acquires' the method `sys.REQUESTED_STANDARD_STREAM.write` under the name `PrintToStreamOnly` to allow outputting
to the standard stream alone, and provides a method `PrintToLogOnly` to allow outputting to the log file alone.
Attributes
----------
stream_name : str
The cloned standard stream name among `stdstream_cloner_t.STREAM_NAME_OUT` for stdout and
`stdstream_cloner_t.STREAM_NAME_ERR` for stderr.
std_stream : TextIO
The cloned standard stream among `sys.stdout` and `sys.stderr`
log_file_path : pathlib.Path
Path to the log file.
log_file_accessor : int or None
File descriptor as returned by `os.open`, or None when the log file is closed.
flush : Callable[[], None]
The cloned standard stream `flush` method.
PrintToStreamOnly : Callable[[str], int]
The cloned standard stream `write` method. To be used to send output to the standard stream only. Note that
since it is simply a pointer to the low-level method `write`, it can only print a single character string at
once, and it does so without any formatting. In particular, no newline is appended, as done by default by the
standard `print` function.
See the `PrintToLogOnly` method.
Methods
-------
write(message)
Equivalent of the cloned standard stream `write` method, with output cloning to the log file. This method is not
meant to be called directly; It is called through the `print` function.
PrintToLogOnly(message)
To be used to send output to the log file only. See the note on being a low-level method in the description of
the `PrintToStreamOnly` attribute.
Enable()
Re-enable standard stream cloning after having disabled it with the `Disable` method.
Disable()
Disable standard stream cloning. Cloning can be re-enabled later with the `Enable` method.
"""
__slots__ = (
"stream_name",
"std_stream",
"log_file_path",
"log_file_accessor",
"flush",
"PrintToStreamOnly",
"_latest_newline_position", # Position of the latest newline character "\n" in the log file
)
STREAM_NAME_ERR: Final[str] = "err"
STREAM_NAME_OUT: Final[str] = "out"
stream_name: str
std_stream: TextIO
log_file_path: phlb.Path
log_file_accessor: Optional[int]
flush: Callable[[], None]
PrintToStreamOnly: Callable[[str], int]
_latest_newline_position: Optional[int]
def __init__(self, path: Union[str, phlb.Path], stream_name: str, /):
"""
Parameters
----------
path : Union[str, pathlib.Path]
Requested log file path. Valid paths are: path to an un-existing file, in which case the log file will be
created, or path to an existing file, in which case the file will be appended to. In any other case (for
example, path to an existing folder), a runtime exception is raised.
stream_name : str
Should be `stdstream_cloner_t.STREAM_NAME_OUT` for stdout or `stdstream_cloner_t.STREAM_NAME_ERR` for
stderr.
"""
# As a precaution, every slot is first initialized to None
for slot in self.__class__.__slots__:
setattr(self, slot, None)
self.stream_name = stream_name
if stream_name == self.__class__.STREAM_NAME_OUT:
self.std_stream = syst.stdout
elif stream_name == self.__class__.STREAM_NAME_ERR:
self.std_stream = syst.stderr
else:
raise ValueError(
f"{stream_name}: Invalid stream type; "
f'Expected: "{self.__class__.STREAM_NAME_OUT}" or "{self.__class__.STREAM_NAME_ERR}"'
)
self.log_file_path = phlb.Path(path) # Ensures a copy is made
self.flush = self.std_stream.flush
self.PrintToStreamOnly = self.std_stream.write
self._OpenLog()
self.Enable()
def write(self, message: str, /) -> int:
""""""
n_characters = self.std_stream.write(message)
_ = self.PrintToLogOnly(message)
return n_characters
def PrintToLogOnly(self, message: str) -> int:
""""""
message = message.replace("\a", "")
for move in ("\f", "\v"):
if move in message:
message = message.replace(move, "\n")
if message.startswith("\r"):
message = message[1:]
if self._latest_newline_position is not None:
opsy.lseek(
self.log_file_accessor,
self._latest_newline_position + 1,
opsy.SEEK_SET,
)
elif message.startswith("\b"):
length_before = message.__len__()
message = message.lstrip("\b")
length_after = message.__len__()
opsy.lseek(
self.log_file_accessor, length_after - length_before, opsy.SEEK_CUR
)
set_position = False
rewind_length = 0
if message.endswith("\b"):
length_before = message.__len__()
message = message.rstrip("\b")
length_after = message.__len__()
rewind_length = length_after - length_before
elif message.endswith("\r"):
message = message[:-1]
set_position = True
n_characters = message.__len__()
if n_characters > 0:
for unwanted, replacement in zip(("\r", "\b"), ("⇦", "←")):
if unwanted in message: # Should not happen
message = message.replace(unwanted, replacement)
self._UpdateLatestNewlinePosition(message)
opsy.write(self.log_file_accessor, message.encode())
if rewind_length < 0:
opsy.lseek(self.log_file_accessor, rewind_length, opsy.SEEK_CUR)
elif set_position and (self._latest_newline_position is not None):
opsy.lseek(
self.log_file_accessor, self._latest_newline_position + 1, opsy.SEEK_SET
)
return n_characters
def Enable(self) -> None:
""""""
# To start cloning with an empty stream
self.flush()
if self.stream_name == self.__class__.STREAM_NAME_OUT:
syst.stdout = self
else:
syst.stderr = self
def Disable(self) -> None:
""""""
self.flush()
self._CloseLog()
if self.stream_name == self.__class__.STREAM_NAME_OUT:
syst.stdout = self.std_stream
else:
syst.stderr = self.std_stream
def _UpdateLatestNewlinePosition(self, message: str) -> None:
"""
Must be called before writing to log file so that file descriptor has not moved yet.
"""
newline_position = message.rfind("\n")
if newline_position != -1:
self._latest_newline_position = (
opsy.lseek(self.log_file_accessor, 0, opsy.SEEK_CUR) + newline_position
)
def _OpenLog(self) -> None:
""""""
if self.log_file_accessor is None:
already_exists = self.log_file_path.exists()
if (not already_exists) or self.log_file_path.is_file():
if already_exists:
opening_mode = opsy.O_WRONLY
else:
opening_mode = opsy.O_CREAT | opsy.O_WRONLY
self.log_file_accessor = opsy.open(self.log_file_path, opening_mode)
self._latest_newline_position = None
else:
raise RuntimeError(
f"{self.log_file_path}: File exists and is not a regular file"
)
else:
raise RuntimeError("Trying to open an already-opened log file")
def _CloseLog(self) -> None:
""""""
if self.log_file_accessor is None:
raise RuntimeError("Trying to close an unopened log file")
else:
opsy.close(self.log_file_accessor)
self.log_file_accessor = NoneClasses
class stdstream_cloner_t (path: Union[str, pathlib.Path], stream_name: str, /)-
Standard Stream Cloner into a Log File. See the module documentation for a general description of the role of the class.
The class implements cloning by replacing the requested standard stream in the
sysmodule of the standard library with an instance of itself, while implementing a stream-cloning methodwriteand 'acquiring' the methodsys.REQUESTED_STANDARD_STREAM.flushwhere REQUESTED_STANDARD_STREAM is eitherstdoutorstderr. It also 'acquires' the methodsys.REQUESTED_STANDARD_STREAM.writeunder the namePrintToStreamOnlyto allow outputting to the standard stream alone, and provides a methodPrintToLogOnlyto allow outputting to the log file alone.Attributes
stream_name:str- The cloned standard stream name among
stdstream_cloner_t.STREAM_NAME_OUTfor stdout andstdstream_cloner_t.STREAM_NAME_ERRfor stderr. std_stream:TextIO- The cloned standard stream among
sys.stdoutandsys.stderr log_file_path:pathlib.Path- Path to the log file.
log_file_accessor:intorNone- File descriptor as returned by
os.open, or None when the log file is closed. flush:Callable[[], None]- The cloned standard stream
flushmethod. PrintToStreamOnly:Callable[[str], int]- The cloned standard stream
writemethod. To be used to send output to the standard stream only. Note that since it is simply a pointer to the low-level methodwrite, it can only print a single character string at once, and it does so without any formatting. In particular, no newline is appended, as done by default by the standardprintfunction. See thePrintToLogOnlymethod.
Methods
write(message) Equivalent of the cloned standard stream
writemethod, with output cloning to the log file. This method is not meant to be called directly; It is called through theprintfunction. PrintToLogOnly(message) To be used to send output to the log file only. See the note on being a low-level method in the description of thePrintToStreamOnlyattribute. Enable() Re-enable standard stream cloning after having disabled it with theDisablemethod. Disable() Disable standard stream cloning. Cloning can be re-enabled later with theEnablemethod.Parameters
path:Union[str, pathlib.Path]- Requested log file path. Valid paths are: path to an un-existing file, in which case the log file will be created, or path to an existing file, in which case the file will be appended to. In any other case (for example, path to an existing folder), a runtime exception is raised.
stream_name:str- Should be
stdstream_cloner_t.STREAM_NAME_OUTfor stdout orstdstream_cloner_t.STREAM_NAME_ERRfor stderr.
Expand source code
class stdstream_cloner_t: """ Standard Stream Cloner into a Log File. See the module documentation for a general description of the role of the class. The class implements cloning by replacing the requested standard stream in the `sys` module of the standard library with an instance of itself, while implementing a stream-cloning method `write` and 'acquiring' the method `sys.REQUESTED_STANDARD_STREAM.flush` where REQUESTED_STANDARD_STREAM is either `stdout` or `stderr`. It also 'acquires' the method `sys.REQUESTED_STANDARD_STREAM.write` under the name `PrintToStreamOnly` to allow outputting to the standard stream alone, and provides a method `PrintToLogOnly` to allow outputting to the log file alone. Attributes ---------- stream_name : str The cloned standard stream name among `stdstream_cloner_t.STREAM_NAME_OUT` for stdout and `stdstream_cloner_t.STREAM_NAME_ERR` for stderr. std_stream : TextIO The cloned standard stream among `sys.stdout` and `sys.stderr` log_file_path : pathlib.Path Path to the log file. log_file_accessor : int or None File descriptor as returned by `os.open`, or None when the log file is closed. flush : Callable[[], None] The cloned standard stream `flush` method. PrintToStreamOnly : Callable[[str], int] The cloned standard stream `write` method. To be used to send output to the standard stream only. Note that since it is simply a pointer to the low-level method `write`, it can only print a single character string at once, and it does so without any formatting. In particular, no newline is appended, as done by default by the standard `print` function. See the `PrintToLogOnly` method. Methods ------- write(message) Equivalent of the cloned standard stream `write` method, with output cloning to the log file. This method is not meant to be called directly; It is called through the `print` function. PrintToLogOnly(message) To be used to send output to the log file only. See the note on being a low-level method in the description of the `PrintToStreamOnly` attribute. Enable() Re-enable standard stream cloning after having disabled it with the `Disable` method. Disable() Disable standard stream cloning. Cloning can be re-enabled later with the `Enable` method. """ __slots__ = ( "stream_name", "std_stream", "log_file_path", "log_file_accessor", "flush", "PrintToStreamOnly", "_latest_newline_position", # Position of the latest newline character "\n" in the log file ) STREAM_NAME_ERR: Final[str] = "err" STREAM_NAME_OUT: Final[str] = "out" stream_name: str std_stream: TextIO log_file_path: phlb.Path log_file_accessor: Optional[int] flush: Callable[[], None] PrintToStreamOnly: Callable[[str], int] _latest_newline_position: Optional[int] def __init__(self, path: Union[str, phlb.Path], stream_name: str, /): """ Parameters ---------- path : Union[str, pathlib.Path] Requested log file path. Valid paths are: path to an un-existing file, in which case the log file will be created, or path to an existing file, in which case the file will be appended to. In any other case (for example, path to an existing folder), a runtime exception is raised. stream_name : str Should be `stdstream_cloner_t.STREAM_NAME_OUT` for stdout or `stdstream_cloner_t.STREAM_NAME_ERR` for stderr. """ # As a precaution, every slot is first initialized to None for slot in self.__class__.__slots__: setattr(self, slot, None) self.stream_name = stream_name if stream_name == self.__class__.STREAM_NAME_OUT: self.std_stream = syst.stdout elif stream_name == self.__class__.STREAM_NAME_ERR: self.std_stream = syst.stderr else: raise ValueError( f"{stream_name}: Invalid stream type; " f'Expected: "{self.__class__.STREAM_NAME_OUT}" or "{self.__class__.STREAM_NAME_ERR}"' ) self.log_file_path = phlb.Path(path) # Ensures a copy is made self.flush = self.std_stream.flush self.PrintToStreamOnly = self.std_stream.write self._OpenLog() self.Enable() def write(self, message: str, /) -> int: """""" n_characters = self.std_stream.write(message) _ = self.PrintToLogOnly(message) return n_characters def PrintToLogOnly(self, message: str) -> int: """""" message = message.replace("\a", "") for move in ("\f", "\v"): if move in message: message = message.replace(move, "\n") if message.startswith("\r"): message = message[1:] if self._latest_newline_position is not None: opsy.lseek( self.log_file_accessor, self._latest_newline_position + 1, opsy.SEEK_SET, ) elif message.startswith("\b"): length_before = message.__len__() message = message.lstrip("\b") length_after = message.__len__() opsy.lseek( self.log_file_accessor, length_after - length_before, opsy.SEEK_CUR ) set_position = False rewind_length = 0 if message.endswith("\b"): length_before = message.__len__() message = message.rstrip("\b") length_after = message.__len__() rewind_length = length_after - length_before elif message.endswith("\r"): message = message[:-1] set_position = True n_characters = message.__len__() if n_characters > 0: for unwanted, replacement in zip(("\r", "\b"), ("⇦", "←")): if unwanted in message: # Should not happen message = message.replace(unwanted, replacement) self._UpdateLatestNewlinePosition(message) opsy.write(self.log_file_accessor, message.encode()) if rewind_length < 0: opsy.lseek(self.log_file_accessor, rewind_length, opsy.SEEK_CUR) elif set_position and (self._latest_newline_position is not None): opsy.lseek( self.log_file_accessor, self._latest_newline_position + 1, opsy.SEEK_SET ) return n_characters def Enable(self) -> None: """""" # To start cloning with an empty stream self.flush() if self.stream_name == self.__class__.STREAM_NAME_OUT: syst.stdout = self else: syst.stderr = self def Disable(self) -> None: """""" self.flush() self._CloseLog() if self.stream_name == self.__class__.STREAM_NAME_OUT: syst.stdout = self.std_stream else: syst.stderr = self.std_stream def _UpdateLatestNewlinePosition(self, message: str) -> None: """ Must be called before writing to log file so that file descriptor has not moved yet. """ newline_position = message.rfind("\n") if newline_position != -1: self._latest_newline_position = ( opsy.lseek(self.log_file_accessor, 0, opsy.SEEK_CUR) + newline_position ) def _OpenLog(self) -> None: """""" if self.log_file_accessor is None: already_exists = self.log_file_path.exists() if (not already_exists) or self.log_file_path.is_file(): if already_exists: opening_mode = opsy.O_WRONLY else: opening_mode = opsy.O_CREAT | opsy.O_WRONLY self.log_file_accessor = opsy.open(self.log_file_path, opening_mode) self._latest_newline_position = None else: raise RuntimeError( f"{self.log_file_path}: File exists and is not a regular file" ) else: raise RuntimeError("Trying to open an already-opened log file") def _CloseLog(self) -> None: """""" if self.log_file_accessor is None: raise RuntimeError("Trying to close an unopened log file") else: opsy.close(self.log_file_accessor) self.log_file_accessor = NoneClass variables
var STREAM_NAME_ERR : Final[str]var STREAM_NAME_OUT : Final[str]
Instance variables
var PrintToStreamOnly : Callable[[str], int]-
Return an attribute of instance, which is of type owner.
var flush : Callable[[], NoneType]-
Return an attribute of instance, which is of type owner.
var log_file_accessor : Union[int, NoneType]-
Return an attribute of instance, which is of type owner.
var log_file_path : pathlib.Path-
Return an attribute of instance, which is of type owner.
var std_stream :-
Return an attribute of instance, which is of type owner.
var stream_name : str-
Return an attribute of instance, which is of type owner.
Methods
def Disable(self) ‑> NoneType-
Expand source code
def Disable(self) -> None: """""" self.flush() self._CloseLog() if self.stream_name == self.__class__.STREAM_NAME_OUT: syst.stdout = self.std_stream else: syst.stderr = self.std_stream def Enable(self) ‑> NoneType-
Expand source code
def Enable(self) -> None: """""" # To start cloning with an empty stream self.flush() if self.stream_name == self.__class__.STREAM_NAME_OUT: syst.stdout = self else: syst.stderr = self def PrintToLogOnly(self, message: str) ‑> int-
Expand source code
def PrintToLogOnly(self, message: str) -> int: """""" message = message.replace("\a", "") for move in ("\f", "\v"): if move in message: message = message.replace(move, "\n") if message.startswith("\r"): message = message[1:] if self._latest_newline_position is not None: opsy.lseek( self.log_file_accessor, self._latest_newline_position + 1, opsy.SEEK_SET, ) elif message.startswith("\b"): length_before = message.__len__() message = message.lstrip("\b") length_after = message.__len__() opsy.lseek( self.log_file_accessor, length_after - length_before, opsy.SEEK_CUR ) set_position = False rewind_length = 0 if message.endswith("\b"): length_before = message.__len__() message = message.rstrip("\b") length_after = message.__len__() rewind_length = length_after - length_before elif message.endswith("\r"): message = message[:-1] set_position = True n_characters = message.__len__() if n_characters > 0: for unwanted, replacement in zip(("\r", "\b"), ("⇦", "←")): if unwanted in message: # Should not happen message = message.replace(unwanted, replacement) self._UpdateLatestNewlinePosition(message) opsy.write(self.log_file_accessor, message.encode()) if rewind_length < 0: opsy.lseek(self.log_file_accessor, rewind_length, opsy.SEEK_CUR) elif set_position and (self._latest_newline_position is not None): opsy.lseek( self.log_file_accessor, self._latest_newline_position + 1, opsy.SEEK_SET ) return n_characters def write(self, message: str, /) ‑> int-
Expand source code
def write(self, message: str, /) -> int: """""" n_characters = self.std_stream.write(message) _ = self.PrintToLogOnly(message) return n_characters