Permalink
Cannot retrieve contributors at this time
Name already in use
A tag already exists with the provided branch name. Many Git commands accept both tag and branch names, so creating this branch may cause unexpected behavior. Are you sure you want to create this branch?
EOASkripts/src/utils/load_config.py
Go to fileThis commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
171 lines (144 sloc)
4.75 KB
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
# -*- coding: utf-8; mode: python -*- | |
import configparser | |
import logging | |
import subprocess | |
from subprocess import PIPE, STDOUT | |
from pathlib import Path | |
import os | |
import shlex | |
import time | |
import shutil | |
import tempfile | |
def copy_file_overwrite( src, dst ): | |
if os.path.exists( dst ): | |
shutil.rmtree( dst ) | |
shutil.copy( src, dst) | |
def copy_dir_overwrite( src, dst ): | |
if os.path.exists( dst ): | |
shutil.rmtree( dst ) | |
shutil.copytree( src, dst) | |
class ToFile: | |
def __init__( self, filename ): | |
self.filename = filename | |
class ToLog: | |
pass | |
def exec_command( | |
command, | |
error_msg = "ERROR while running {command}", | |
wd = None, | |
output_to = ToLog(), | |
log_level = "INFO", | |
# ignore_fail = False | |
exit_code_ok = lambda x: x == 0, | |
): | |
logging.log( | |
getattr(logging,log_level), | |
f"executing '{command}'", | |
) | |
arguments = shlex.split(command) | |
if isinstance( output_to, ToFile ): | |
log_file = Path( output_to.filename ) | |
log_dir = log_file.resolve().parent | |
logging.info( | |
f"logfile '{log_file}'", | |
) | |
if not (log_dir.exists() and log_dir.is_dir()): | |
os.makedirs( log_dir ) | |
stdout_file = open( output_to.filename, "w") | |
elif isinstance( output_to, ToLog ): | |
logging.info( | |
f"output to log file", | |
) | |
stdout_file = PIPE | |
else: | |
logging.info( | |
f"output dumped", | |
) | |
stdout_file = None | |
process = subprocess.Popen( | |
arguments, | |
cwd = wd, | |
stdout= stdout_file, | |
stderr= STDOUT, | |
) | |
def check_io(stream): | |
while True: | |
output = stream.readline() | |
if output: | |
logging.debug( "> " + output.decode().strip() ) | |
else: | |
break | |
# keep checking stdout/stderr until the child exits | |
logging.debug( f"{command} output:" ) | |
if isinstance( output_to, ToFile ): | |
stdout_file.close() | |
elif isinstance( output_to, ToLog ): | |
while process.poll() is None: | |
check_io(process.stdout) | |
ret = process.wait() # 0 means success | |
if (not exit_code_ok(ret)) and ret != 0: | |
logging.error( error_msg.format( command = command) ) | |
raise( Exception( error_msg.format( command=command ) ) ) | |
def check_executable( executable_name ): | |
logging.info(f"checking executable {executable_name}...") | |
exec_command( | |
f"which {executable_name}", | |
f"PROGRAM not found: {executable_name}", | |
log_level = "NOTSET" | |
) | |
logging.info("ok :-)") | |
def check_file( filename ): | |
logging.info(f"checking '{filename}'...") | |
if not os.path.exists( filename ): | |
raise( Exception( f"'{filename}' is missing!" ) ) | |
logging.info("ok :-)") | |
################################## | |
# Reading the configuration file # | |
################################## | |
def load_config( | |
cfg_file, | |
log_level, # log level in the terminal | |
log_file, | |
log_level_file = logging.DEBUG | |
): | |
CONFIG = configparser.ConfigParser() | |
CONFIG.read( cfg_file ) | |
###################### | |
# Setting up logging # | |
###################### | |
log_dir = Path(log_file).absolute().parent | |
log_dir.mkdir( | |
parents = True, | |
exist_ok = True | |
) | |
time.sleep( 1 ) | |
# always log to file: | |
logging.basicConfig( | |
level=logging.DEBUG, | |
format='%(asctime)s - %(levelname)s - %(message)s', | |
filename = log_file, | |
filemode = "w" | |
) | |
rootLogger = logging.getLogger() | |
# set up logging to terminal: | |
terminal_formatter = \ | |
logging.Formatter( | |
"%(levelname)s - %(message)s" | |
) | |
consoleHandler = logging.StreamHandler() | |
consoleHandler.setFormatter(terminal_formatter) | |
consoleHandler.setLevel( log_level ) | |
rootLogger.addHandler(consoleHandler) | |
# Setup of various dictionaries for localization of various elements | |
# dictLangFootnotes = {"it" : "Note a piè pagina", "fr" : "notes en bas de page", "de" : "Fußnoten", "en" : "Footnotes"} | |
# dict_and = {"en" : "and", "de" : "und", "fr" : "et", "it" : "e"} | |
# dict_ed = {"en" : "ed.", "de" : "Hrsg."} | |
# dict_eds = {"en" : "eds.", "de" : "Hrsg."} | |
# use the translation file that is used also for XSL | |
# translation_xml = etree.parse( str( TRANSLATION_FILE ) ) | |
# dictLangFootnotes = translation_xml.find("//entry[@name='footnotes']").attrib | |
# dict_and = translation_xml.find("//entry[@name='and']").attrib | |
# dict_ed = translation_xml.find("//entry[@name='editor-abbr']").attrib | |
# dict_eds = translation_xml.find("//entry[@name='editors-abbr']").attrib | |
return CONFIG |