# -*- coding: utf-8 -*-
#
# This file is subject to the terms and conditions defined in
# file 'LICENSE.txt', which is part of this source code package.
#
""" Send all the print statements to a file or to the sink os ...
Also the output can be duplicated to a file
"""
import re
import sys
import os
from Muscat.Helpers.Logger import GetDiffTime
COLOR_REGEX_FILTER = re.compile(r'\x1b\[(?P<arg_1>\d+)(;(?P<arg_2>\d+)(;(?P<arg_3>\d+))?)?m')
[docs]def StdErrorPrint(*args, **kwargs) -> None:
"""Function to print to the standard error
"""
print(*args, file=sys.stderr, **kwargs)
[docs]class PrintBypass():
"""Class to be used with a "with" statement to locally redirect the
output (print statements) to other targets (files, sink...)
"""
def __init__(self):
self.stdin_ = sys.stdin
self.stdout_ = sys.stdout # Keep track of the previous value.
self.stderr_ = sys.stderr # Keep track of the previous value.
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.Restore()
[docs] def ToSink(self) -> None:
"""Redirect all the outputs to the sink os.devnull
"""
sys.stdout = open(os.devnull, "w")
sys.stderr = open(os.devnull, "w")
try:
from sympy import init_printing
init_printing(use_unicode=False)
except: # pragma no cover
pass
[docs] def ToDisk(self, filename: str, filenameCerr: str = None) -> None:
"""Redirect all the outputs to file
Parameters
----------
filename : str
File name for the standard output
filenameCerr : str, optional
File name fo the error output if non filename is used, by default None
"""
sys.stdout = open(filename, 'w') # Something here that provides a write method.
if filenameCerr is None:
sys.stderr = sys.stdout
else:
sys.stderr = open(filenameCerr, 'w')
[docs] def CopyOutputToDisk(self, filename: str, filenameCerr: str = None, filterString: str = " -> "):
"""Copy all the output to a file.
a file with the name filename+"2" is created with all the statement filtered by the string
in filterString
Parameters
----------
filename : str
File name for the standard output
filenameCerr : str, optional
File name fo the error output if non filename is used, by default None
filterString : str, optional
String to filter the standard output to the second file, if empty the second file is
not created
"""
class CopyOutputToDisk():
def __init__(self, bufferToCopy, filename):
self.FdOut = open(filename, 'w', encoding="utf-8")
self.sysOut = bufferToCopy
def close(self):
self.FdOut.close()
def flush(self):
self.FdOut.flush()
self.sysOut.flush()
def write(self, data):
# to clean the colors
filteredData = COLOR_REGEX_FILTER.sub('', data)
if len(filteredData):
if filteredData != "\n":
self.FdOut.write(("[%f] " % GetDiffTime()))
self.FdOut.write(filteredData)
self.sysOut.write(data)
class FilterToSecondFile():
def __init__(self, original, second, filterString):
self.filterString = filterString
self.original = original
self.second = second
self.cpt = 0
def close(self):
self.original.close()
self.second.close()
def flush(self):
self.original.flush()
self.second.flush()
def write(self, data):
if data.find(self.filterString) != -1:
if self.second is not None:
self.second.write(COLOR_REGEX_FILTER.sub('', data))
self.second.flush()
return
if self.original is not None:
self.original.write(data)
self.original.flush()
# to make a copy of the main output to a file
copyToDisk = CopyOutputToDisk(self.stdout_, filename)
# to split the output and send all the -> to a second file
if len(filterString) > 0:
sys.stdout = FilterToSecondFile(copyToDisk, open(filename+"2", 'w'), filterString)
else:
sys.stdout = copyToDisk
if filenameCerr is None:
sys.stderr = sys.stdout
else:
sys.stderr = CopyOutputToDisk(self.stderr_, filenameCerr)
[docs] def ToRedirect(self, coutObj, cerrObj=None) -> None:
"""Function to redirect all the outputs to a user definer function
if cerrObj is None coutObj will be used instead
Parameters
----------
coutObj :
User object implementing close(), flush(), write(data)
cerrObj : , optional
User object implementing close(), flush(), write(data)
"""
sys.stdout = coutObj
if cerrObj is not None:
sys.stderr = cerrObj
[docs] def Restore(self) -> None:
"""Restore the original output buffers.
Current buffers will be flushed and closed before restoring the original
"""
if sys.stderr is not self.stderr_ and sys.stderr is not sys.stdout:
sys.stderr.flush()
sys.stderr.close()
sys.stderr = self.stderr_ # restore the previous stderr.
if sys.stdout is not self.stdout_:
sys.stdout.flush()
sys.stdout.close()
sys.stdout = self.stdout_ # restore the previous stdout.
[docs] def Print(self, text: str) -> None:
"""To print to the original cout
Parameters
----------
text : str
message to be printed
"""
self.stdout_.write(text+"\n")
[docs] def PrintCerr(self, text) -> None:
"""To print to the original cerr
Parameters
----------
text : str
message to be printed
"""
self.stderr_.write(text+"\n")
[docs]def CheckIntegrity(GUI:bool=False):
# careful, this class is used during the test.
# do not use this class inside a CheckIntegrity
from Muscat.Helpers.IO.TemporaryDirectory import TemporaryDirectory
fname = TemporaryDirectory.GetTempPath() + "sink"
with PrintBypass() as f:
print("print Before ToSink")
f.ToSink()
f.Print("Print to the original cout")
f.PrintCerr("Print to the original cerr")
f.Restore()
print("print after ToSink and before ToFile")
print("")
f.ToDisk(filename=fname+"_cout.txt")
f.ToDisk(filename=fname+"_cout.txt", filenameCerr=fname+"_cerr.txt")
print("print inside to file cout")
StdErrorPrint("print inside to file cerr")
f.Restore()
print("print after ToFile and before custom class")
print("")
class mySink():
def flush(self): pass
def close(self): pass
def write(self, data): pass
f.ToRedirect(mySink(), mySink())
StdErrorPrint(' sent to cerr (mySink)')
print('sent to cout (mySink)')
f.Restore()
print("print after mySink and before CopyOutputToDisk")
print("")
f.CopyOutputToDisk(filename=fname+"_cout_copy.txt")
print("Hello World")
f.Restore()
f.CopyOutputToDisk(filename=fname+"_cout_copyI.txt", filenameCerr=fname+"_cerr_copyI.txt")
print("Hello World")
print("Debug Data to second standard output -> Hello World ")
f.Restore()
f.CopyOutputToDisk(filename=fname+"_cout_copyII.txt", filenameCerr=fname+"_cerr_copyII.txt",filterString="")
print("Hello World")
StdErrorPrint(' sent to cerr (copy to disk)')
print(' sent to cout (copy to disk) ')
StdErrorPrint(' sent to cerr (copy to disk) 2')
print(' sent to cout (copy to disk) 2')
print(' sent to cout (copy to disk) 2 -> Hello World')
f.Restore()
return "ok"
if __name__ == '__main__':
print(CheckIntegrity()) # pragma: no cover