mirror of
https://github.com/Alex38Lyon/Synthese-PSM_LARRA.git
synced 2026-06-01 22:00:53 +00:00
Script pyThtoDB
This commit is contained in:
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,68 @@
|
||||
import logging
|
||||
import sys
|
||||
import re
|
||||
|
||||
# Couleurs ANSI par niveau de log
|
||||
COLOR_CODES = {
|
||||
logging.DEBUG: "\033[94m", # Bleu
|
||||
logging.INFO: "\033[92m", # Vert
|
||||
logging.WARNING: "\033[93m", # Jaune
|
||||
logging.ERROR: "\033[91m", # Rouge
|
||||
logging.CRITICAL: "\033[1;91m", # Rouge vif
|
||||
}
|
||||
RESET = "\033[0m"
|
||||
|
||||
# Supprime les codes ANSI (pour l'écriture dans les fichiers)
|
||||
def strip_ansi_codes(text):
|
||||
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
|
||||
return ansi_escape.sub('', text)
|
||||
|
||||
# Formatter pour la console avec couleurs
|
||||
class ConsoleFormatter(logging.Formatter):
|
||||
def format(self, record):
|
||||
color = COLOR_CODES.get(record.levelno, "")
|
||||
message = super().format(record)
|
||||
return f"{color}{message}{RESET}"
|
||||
|
||||
# Formatter pour le fichier sans les couleurs
|
||||
class FileFormatter(logging.Formatter):
|
||||
def format(self, record):
|
||||
clean_msg = strip_ansi_codes(record.getMessage())
|
||||
record_copy = logging.LogRecord(
|
||||
name=record.name,
|
||||
level=record.levelno,
|
||||
pathname=record.pathname,
|
||||
lineno=record.lineno,
|
||||
msg=clean_msg,
|
||||
args=(),
|
||||
exc_info=record.exc_info,
|
||||
func=record.funcName,
|
||||
sinfo=record.stack_info
|
||||
)
|
||||
return super().format(record_copy)
|
||||
|
||||
# Fonction de configuration du logger
|
||||
def setup_logger(logfile="app.log", debug_log=False):
|
||||
logger = logging.getLogger("Logger")
|
||||
logger.setLevel(logging.DEBUG) # Toujours logger tout (filtrage par handler)
|
||||
logger.handlers.clear()
|
||||
|
||||
# Détermine le niveau de log selon la variable
|
||||
min_level = logging.DEBUG if debug_log else logging.INFO
|
||||
|
||||
# Console handler
|
||||
console_handler = logging.StreamHandler(sys.stdout)
|
||||
console_handler.setLevel(min_level)
|
||||
console_formatter = ConsoleFormatter("%(levelname)s: %(message)s")
|
||||
console_handler.setFormatter(console_formatter)
|
||||
logger.addHandler(console_handler)
|
||||
|
||||
# File handler
|
||||
file_handler = logging.FileHandler(logfile, encoding="utf-8")
|
||||
file_handler.setLevel(min_level)
|
||||
file_formatter = FileFormatter("%(asctime)s - %(levelname)s - %(message)s")
|
||||
file_handler.setFormatter(file_formatter)
|
||||
logger.addHandler(file_handler)
|
||||
|
||||
return logger
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
|
||||
matplotlib
|
||||
pandas
|
||||
Shapely
|
||||
Fiona
|
||||
pyproj
|
||||
scipy
|
||||
netCDF4
|
||||
xarray
|
||||
joblib
|
||||
geopandas
|
||||
motionless
|
||||
salem
|
||||
configparser
|
||||
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
import tempfile
|
||||
import shutil
|
||||
import os
|
||||
from os.path import join
|
||||
import subprocess
|
||||
import re
|
||||
import logging
|
||||
|
||||
log = logging.getLogger("Logger")
|
||||
|
||||
#################################################################################################
|
||||
# Codes de couleur ANSI
|
||||
class Colors:
|
||||
BLACK = '\033[90m'
|
||||
RED = '\033[91m'
|
||||
GREEN = '\033[92m'
|
||||
YELLOW = '\033[93m'
|
||||
BLUE = '\033[94m'
|
||||
MAGENTA = '\033[95m'
|
||||
CYAN = '\033[96m'
|
||||
WHITE = '\033[97m'
|
||||
|
||||
ERROR = '\033[91m'
|
||||
WARNING = '\033[95m'
|
||||
HEADER = '\033[96m'
|
||||
|
||||
ENDC = '\033[0m'
|
||||
BOLD = '\033[1m'
|
||||
UNDERLINE = '\033[4m'
|
||||
|
||||
#################################################################################################
|
||||
def compile_template(template, template_args, **kwargs):
|
||||
try:
|
||||
log = ""
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
config = template.format(**template_args, tmpdir=tmpdir.replace("\\", "/"))
|
||||
|
||||
print(f"{Colors.YELLOW}{config}{Colors.ENDC}\n")
|
||||
|
||||
config_file = join(tmpdir, "config.thconfig")
|
||||
log_file = join(tmpdir, "log.log")
|
||||
therion_path = kwargs["therion_path"] if "therion_path" in kwargs else "therion"
|
||||
with open(config_file, mode="w+", encoding="utf-8") as tmp:
|
||||
with open(log_file, mode="w+") as tmp2:
|
||||
tmp.write(config)
|
||||
tmp.flush()
|
||||
subprocess.check_output('''"{}" "{}" -l "{}"'''.format(therion_path, config_file, log_file), shell=True, )
|
||||
tmp2.flush()
|
||||
log = tmp2.read()
|
||||
if kwargs["cleanup"]:
|
||||
shutil.rmtree(tmpdir)
|
||||
print("\n" )
|
||||
return log, tmpdir
|
||||
|
||||
|
||||
except Exception as e:
|
||||
print(f"{Colors.ERROR}Error: Therion template compilation error: {Colors.ENDC}{e}")
|
||||
|
||||
#################################################################################################
|
||||
def compile_file(filename, **kwargs):
|
||||
try:
|
||||
tmpdir = os.path.dirname(filename)
|
||||
log_file = join(tmpdir, "therion.log").replace("\\", "/")
|
||||
therion_path = kwargs["therion_path"] if "therion_path" in kwargs else "therion"
|
||||
|
||||
process = subprocess.Popen(
|
||||
[therion_path, filename, "-l", log_file],
|
||||
cwd=tmpdir,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT, # fusion stdout + stderr
|
||||
universal_newlines=True, # décodage automatique en texte
|
||||
bufsize=1 # ligne par ligne
|
||||
)
|
||||
|
||||
log.info(f"Start therion compilation file : {Colors.MAGENTA}{filename}")
|
||||
# Lecture en temps réel
|
||||
for line in process.stdout:
|
||||
line = line.rstrip()
|
||||
lower_line = line.lower()
|
||||
if "error" in lower_line:
|
||||
log.error(f"\t\t{line}")
|
||||
elif "warning" in lower_line:
|
||||
log.warning(f"\t{line}")
|
||||
else:
|
||||
log.debug(f"\t\t{Colors.WHITE}{line}")
|
||||
|
||||
process.wait()
|
||||
|
||||
# Si la commande échoue, result.returncode sera non nul
|
||||
if process.returncode != 0:
|
||||
# Affichage des erreurs et de la sortie standard
|
||||
log.error(f"Error during Therion compilation")
|
||||
log.error(f"stderr: \n{Colors.MAGENTA}{process.stderr.decode()}")
|
||||
|
||||
log.info(f"Therion file : {Colors.MAGENTA}{filename} {Colors.GREEN}compilation succeeded")
|
||||
|
||||
except Exception as e:
|
||||
log.error(f"Error: Therion file {Colors.ENDC}{filename}{Colors.ERROR} compilation error: {Colors.ENDC}{e}")
|
||||
|
||||
|
||||
|
||||
|
||||
#################################################################################################
|
||||
def compile_file_th(filepath, **kwargs):
|
||||
template = """source {filepath}
|
||||
layout test
|
||||
scale 1 500
|
||||
endlayout
|
||||
"""
|
||||
template_args = {"filepath": filepath}
|
||||
logs, _ = compile_template(template, template_args, cleanup=True, **kwargs)
|
||||
return logs
|
||||
|
||||
# Attention, version avec therion en français ! à voir pour les autres langues
|
||||
lengthre = re.compile(r".*Longueur totale de la topographie = \s*(\S+)m")
|
||||
depthre = re.compile(r".*Longueur totale verticale =\s*(\S+)m")
|
||||
|
||||
#################################################################################################
|
||||
def get_stats_from_log(log):
|
||||
lenmatch = lengthre.findall(log)
|
||||
depmatch = depthre.findall(log)
|
||||
if len(lenmatch) == 1 and len(depmatch) == 1:
|
||||
return {"length": lenmatch[0], "depth": depmatch[0]}
|
||||
return {"length": 0, "depth": 0}
|
||||
|
||||
|
||||
syscoord = re.compile(r".*output coordinate system: \s*(\S+)")
|
||||
|
||||
#################################################################################################
|
||||
def get_syscoord_from_log(log):
|
||||
lenmatch = syscoord.findall(log)
|
||||
|
||||
if len(lenmatch) == 1:
|
||||
return {"syscoord": lenmatch[0]}
|
||||
return {"syscoord": 0}
|
||||
Reference in New Issue
Block a user