reorg parsers

ftp_csv_receiver.py

@@ -6,7 +6,7 @@ import logging
 from hashlib import sha256
 from pathlib import Path
 
-from utils.config import loader_ftp_csv
+from utils.config import loader_ftp_csv as setting
 from utils.database.connection import connetti_db
 from utils.ftp import user_admin, file_management
 
@@ -114,7 +114,7 @@ class ASEHandler(FTPHandler):
 def main():
     """Main function to start the FTP server."""
     # Load the configuration settings
-    cfg = loader_ftp_csv.Config()
+    cfg = setting.Config()
 
     try:
         # Initialize the authorizer and handler

load_orchestrator.py

@@ -78,7 +78,7 @@ async def load_csv(cfg: object) -> tuple:
             logger.debug("Connessione al database stabilita")
 
             query = f"""
-                SELECT id, unit_type, tool_type
+                SELECT id, unit_type, tool_type, unit_name, tool_name
                 FROM {cfg.dbname}.{cfg.dbrectable}
                 WHERE locked = 0 AND status = {CSV_RECEIVED}
                 LIMIT 1
@@ -88,8 +88,8 @@ async def load_csv(cfg: object) -> tuple:
             result = cur.fetchone()
 
             if result:
-                id, unit_type, tool_type = result
-                logger.info(f"Trovato CSV da elaborare: ID={id}, Tipo={unit_type}_{tool_type}")
+                id, unit_type, tool_type, unit_name, tool_name = result
+                logger.info(f"Trovato CSV da elaborare: ID={id}, Tipo={unit_type}_{tool_type}, Nome={unit_name}_{tool_name}")
 
                 lock_query = f"UPDATE {cfg.dbname}.{cfg.dbrectable} SET locked = 1 WHERE id = {id}"
                 logger.debug(f"Lock del record: {lock_query}")
@@ -97,21 +97,32 @@ async def load_csv(cfg: object) -> tuple:
                 conn.commit()
 
                 # Costruisce il nome del modulo da caricare dinamicamente
-                module_name = f'utils.parsers.{unit_type.lower()}_{tool_type.lower()}'
-                logger.debug(f"Caricamento modulo dinamico: {module_name}")
+                module_names = [f'utils.parsers.by_name.{unit_name.lower()}_{tool_name.lower()}',
+                                f'utils.parsers.by_name.{unit_name.lower()}_{tool_type.lower()}',
+                                f'utils.parsers.by_name.{unit_name.lower()}_all',
+                                f'utils.parsers.by_type.{unit_type.lower()}_{tool_type.lower()}']
+                modulo = None
+                for module_name in module_names:
+                    try:
+                        logger.debug(f"Caricamento dinamico del modulo: {module_name}")
+                        modulo = importlib.import_module(module_name)
+                        logger.debug(f"Funzione 'main_loader' caricata dal modulo {module_name}")
+                        return True, True
+                    except (ImportError, AttributeError) as e:
+                        logger.error(f"Errore nel caricamento del modulo {module_name}: {e}", exc_info=debug_mode)
 
-                try:
-                    modulo = importlib.import_module(module_name)
-                    funzione = getattr(modulo, "main_loader")
-                    logger.debug(f"Funzione 'main_loader' trovata nel modulo {module_name}")
-
-                    # Esegui la funzione async
-                    await funzione(cfg, id)
-                    logger.info(f"Elaborazione completata per ID={id}")
-                    return True, True
-                except (ImportError, AttributeError) as e:
-                    logger.error(f"Errore nel caricamento del modulo o della funzione: {e}", exc_info=debug_mode)
+                if not modulo:
+                    logger.error(f"Nessun modulo trovato {module_names}")
                     return True, False
+
+                # Ottiene la funzione 'main_loader' dal modulo
+
+                funzione = getattr(modulo, "main_loader")
+
+                # Esegui la funzione
+                await funzione(cfg, id)
+                logger.info(f"Elaborazione completata per ID={id}")
+                return True, True
             else:
                 logger.debug("Nessun record disponibile per l'elaborazione")
                 return False, False

utils/csv/data_preparation.py

@@ -1,5 +1,6 @@
 #!.venv/bin/python
 from utils.database.connection import connetti_db
+from utils.database.nodes_query import get_nodes_type
 import utils.timestamp.date_check as date_check
 import logging
 import re
@@ -32,7 +33,7 @@ def get_data(cfg: object, id: int) -> tuple:
         conn.close()
         return unit_name, tool_name, tool_data
 
-def make_matrix(cfg: object, id: int) -> list:
+def make_pipe_sep_matrix(cfg: object, id: int) -> list:
     """
     Processes raw tool data and transforms it into a matrix format for database insertion.
 
@@ -67,7 +68,7 @@ def make_matrix(cfg: object, id: int) -> list:
 
     return matrice_valori
 
-def make_loc_matrix(cfg: object, id: int) -> list:
+def make_ain_din_matrix(cfg: object, id: int) -> list:
     """
     Processes raw location (LOC) tool data and transforms it into a matrix format for database insertion.
 
@@ -101,7 +102,7 @@ def make_loc_matrix(cfg: object, id: int) -> list:
 
     return matrice_valori
 
-def make_matrix_with_channels(cfg: object, id: int, node_channels: list) -> list:
+def make_channels_matrix(cfg: object, id: int) -> list:
     UnitName, ToolNameID, ToolData = get_data(cfg, id)
     righe = ToolData.splitlines()
     matrice_valori = []
@@ -110,11 +111,11 @@ def make_matrix_with_channels(cfg: object, id: int, node_channels: list) -> list
         EventDate, EventTime = timestamp.split(' ')
         valori_splitted = [valore for valore in rilevazioni.split(';') if valore != '|']
         valori_iter = iter(valori_splitted)
+        node_channels, node_types, node_ains, node_dins = get_nodes_type(cfg, ToolNameID, UnitName)
+
         valori_nodi = [list(islice(valori_iter, channels)) for channels in node_channels]
 
         for num_nodo, valori in enumerate(valori_nodi, start=1):
             matrice_valori.append([UnitName, ToolNameID, num_nodo, date_check.conforma_data(EventDate), EventTime, batlevel, temperature] + valori + ([None] * (19 - len(valori))))
 
     return matrice_valori
-
-

utils/csv/loaders.py

@@ -0,0 +1,22 @@
+from utils.database.loader_action import load_data, update_status, DATA_LOADED
+from utils.csv.data_preparation import make_pipe_sep_matrix, make_ain_din_matrix, make_channels_matrix
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+async def main_loader(cfg: object, id: int, action: str) -> None:
+    type_matrix_mapping = {
+        "pipe_separator": make_pipe_sep_matrix,
+        "analogic_digital": make_ain_din_matrix,
+        "channels": make_channels_matrix
+    }
+    if action in type_matrix_mapping:
+        function_to_call = type_matrix_mapping[action]
+        # Create a matrix of values from the data
+        matrice_valori = function_to_call(cfg, id)
+        # Load the data into the database
+        if load_data(cfg, matrice_valori):
+            update_status(cfg, id, DATA_LOADED)
+    else:
+        logger.warning(f"Action '{action}' non riconosciuta.")

utils/parsers/by_name/__init__.py

@@ -0,0 +1 @@
+"""Parser delle centraline"""

utils/parsers/by_type/__init__.py

@@ -0,0 +1 @@
+"""Parser delle centraline"""

utils/parsers/by_type/cr1000x_cr1000x.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/by_type/d2w_d2w.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/by_type/g301_g301.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/by_type/g801_iptm.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/by_type/g801_loc.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as analog_dig_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return analog_dig_main_loader(cfg, id, "analogic_digital")

utils/parsers/by_type/g801_mums.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/by_type/g802_dsas.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/by_type/g802_loc.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as analog_dig_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return analog_dig_main_loader(cfg, id, "analogic_digital")

utils/parsers/by_type/g802_modb.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/by_type/g802_mums.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/by_type/gs1_gs1.py

@@ -1,4 +1,4 @@
-from .tlp_tlp import main_loader as tlp_tlp_main_loader
+from ..tlp_tlp import main_loader as tlp_tlp_main_loader
 
 async def main_loader(cfg: object, id: int) -> None:
     return tlp_tlp_main_loader(cfg, id)

utils/parsers/by_type/hortus_hortus.py

@@ -0,0 +1,4 @@
+from utils.csv.loaders import main_loader as pipe_sep_main_loader
+
+async def main_loader(cfg: object, id: int) -> None:
+    return pipe_sep_main_loader(cfg, id, "pipe_separator")

utils/parsers/cr1000x_cr1000x.py

@@ -1,15 +0,0 @@
-#!.venv/bin/python
-# Import necessary modules
-from utils.database.loader_action import load_data, update_status, DATA_LOADED
-from utils.csv.data_preparation import make_matrix
-import logging
-
-logger = logging.getLogger(__name__)
-
-# Define the main function for loading data
-async def main_loader(cfg: object, id: int) -> None:
-    # Create a matrix of values from the data
-    matrice_valori = make_matrix(cfg, id)
-    # Load the data into the database
-    if load_data(cfg, matrice_valori):
-        update_status(cfg, id, DATA_LOADED)

utils/parsers/d2w_d2w.py

@@ -1,15 +0,0 @@
-#!.venv/bin/python
-# Import necessary modules
-from utils.database.loader_action import load_data, update_status, DATA_LOADED
-from utils.csv.data_preparation import make_matrix
-import logging
-
-logger = logging.getLogger(__name__)
-
-# Define the main function for loading data
-async def main_loader(cfg: object, id: int) -> None:
-    # Create a matrix of values from the data
-    matrice_valori = make_matrix(cfg, id)
-    # Load the data into the database
-    if load_data(cfg, matrice_valori):
-        update_status(cfg, id, DATA_LOADED)

utils/parsers/g301_g301.py

@@ -1,15 +0,0 @@
-#!.venv/bin/python
-# Import necessary modules
-from utils.database.loader_action import load_data, update_status, DATA_LOADED
-from utils.csv.data_preparation import make_matrix
-import logging
-
-logger = logging.getLogger(__name__)
-
-# Define the main function for loading data
-async def main_loader(cfg: object, id: int) -> None:
-    # Create a matrix of values from the data
-    matrice_valori = make_matrix(cfg, id)
-    # Load the data into the database
-    if load_data(cfg, matrice_valori):
-        update_status(cfg, id, DATA_LOADED)

utils/parsers/g801_iptm.py

@@ -1,4 +0,0 @@
-from .g801_mums import main_loader as g801_mums_main_loader
-
-async def main_loader(cfg: object, id: int) -> None:
-    return g801_mums_main_loader(cfg, id)

utils/parsers/g801_loc.py

@@ -1,12 +0,0 @@
-#!.venv/bin/python
-from utils.database.loader_action import load_data
-from utils.csv.data_preparation import make_loc_matrix
-import logging
-
-logger = logging.getLogger(__name__)
-
-async def main_loader(cfg, id):
-    matrice_valori = make_loc_matrix(cfg, id)
-    load_data(cfg, matrice_valori)
-
-

utils/parsers/g801_mums.py

@@ -1,15 +0,0 @@
-#!.venv/bin/python
-# Import necessary modules
-from utils.database.loader_action import load_data, update_status, DATA_LOADED
-from utils.csv.data_preparation import make_matrix
-import logging
-
-logger = logging.getLogger(__name__)
-
-# Define the main function for loading data
-async def main_loader(cfg: object, id: int) -> None:
-    # Create a matrix of values from the data
-    matrice_valori = make_matrix(cfg, id)
-    # Load the data into the database
-    if load_data(cfg, matrice_valori):
-        update_status(cfg, id, DATA_LOADED)

utils/parsers/g802_dsas.py

@@ -1,4 +0,0 @@
-from .g801_mums import main_loader as g801_mums_main_loader
-
-async def main_loader(cfg: object, id: int) -> None:
-    return g801_mums_main_loader(cfg, id)

utils/parsers/g802_loc.py

@@ -1,4 +0,0 @@
-from .g801_loc import main_loader as g801_loc_main_loader
-
-async def main_loader(cfg: object, id: int) -> None:
-    return g801_loc_main_loader(cfg, id)

utils/parsers/g802_modb.py

@@ -1,4 +0,0 @@
-from .g801_mums import main_loader as g801_mums_main_loader
-
-async def main_loader(cfg: object, id: int) -> None:
-    return g801_mums_main_loader(cfg, id)

utils/parsers/g802_mums.py

@@ -1,4 +0,0 @@
-from .g801_mums import main_loader as g801_mums_main_loader
-
-async def main_loader(cfg: object, id: int) -> None:
-    return g801_mums_main_loader(cfg, id)

utils/parsers/hortus_hortus.py

@@ -1,4 +0,0 @@
-from .cr1000x_cr1000x import main_loader as cr1000x_cr1000x_main_loader
-
-async def main_loader(cfg: object, id: int) -> None:
-    return cr1000x_cr1000x_main_loader(cfg, id)