xml parsing working

Author: lukaspie

src/pynxtools_xps/specs/sle/flatten_xml.py

@@ -103,8 +103,6 @@ def extract_device_info(elem: ET.Element) -> Dict[str, Any]:
 
     device_settings = extract_devices(elem)
 
-    print(device_settings)
-
     return {unique_name: device_settings}
 
 
@@ -270,7 +268,7 @@ def flatten_schedule(xml: ET.Element) -> List[Dict[str, Any]]:
     return collect
 
 
-def flatten_context(xml: ET.Element) -> List[Dict[str, Any]]:
+def flatten_context(xml: ET.Element) -> Dict[str, Any]:
     """
     Flatten the nested XML context, keeping only the needed metadata.
 
@@ -281,10 +279,13 @@ def flatten_context(xml: ET.Element) -> List[Dict[str, Any]]:
 
     Returns
     -------
-    collect : list
+    Dict[str, Any]:
         Dictionary with device metadata.
 
     """
+    if xml is not None:
+        return {}
+
     device_metadata: Dict[str, Any] = {}
 
     for elem in xml.iter("DeviceContext"):
@@ -304,13 +305,16 @@ def flatten_metainfo(xml: ET.Element) -> List[Dict[str, Any]]:
 
     Returns
     -------
-    collect : list
+    Dict[str, Any]:
         Dictionary with metainfo.
 
     """
+    if xml is not None:
+        return {}
+
     metainfo: Dict[str, Any] = {}
 
-    for elem in xml.iter("DeviceContext"):
+    for elem in xml.iter("Parameter"):
         process_xml_element(elem, metainfo)
 
     return metainfo

src/pynxtools_xps/specs/sle/sle_specs.py

@@ -31,7 +31,8 @@
 import struct
 import copy
 import logging
-from typing import Dict, Any, List
+from pathlib import Path
+from typing import Dict, Any, List, Union
 import warnings
 from packaging.version import Version, InvalidVersion
 import sqlite3
@@ -46,6 +47,7 @@
     re_map_keys,
     re_map_values,
     drop_unused_keys,
+    update_dict_without_overwrite,
 )
 from pynxtools_xps.value_mappers import (
     convert_energy_type,
@@ -101,12 +103,12 @@ def __init__(self):
             SleProdigyParser,
         ]
 
-        self.sql_connection: str
+        self.file: Union[str, Path] = ""
 
         super().__init__()
 
     def _get_sle_version(self):
-        con = sqlite3.connect(self.sql_connection)
+        con = sqlite3.connect(self.file)
         query = 'SELECT Value FROM Configuration WHERE Key="Version"'
         return execute_sql_query_on_con(con, query)[0][0]
 
@@ -191,7 +193,7 @@ def is_version_supported(version, supported_version_ranges):
                 f"Version {version} of SPECS Prodigy is currently not supported."
             )
 
-        return supporting_parsers[-1]  # always use newest parser
+        return supporting_parsers[-1]()  # always use newest parser
 
     def parse_file(self, file: str, **kwargs: Dict[str, Any]) -> Dict[str, Any]:
         """
@@ -212,7 +214,7 @@ def parse_file(self, file: str, **kwargs: Dict[str, Any]) -> Dict[str, Any]:
             Dict with parsed data.
 
         """
-        self.sql_connection = file
+        self.file = file
         return super().parse_file(file, **kwargs)
 
     def construct_data(self):
@@ -448,7 +450,7 @@ def __init__(self):
         self.spectra: List[Dict[str, Any]] = []
         self.xml_schedule: ET.Element = None
         self.xml_context: ET.Element = None
-        self.xml_metainfo: ET.Element = None  ######## TODO
+        self.xml_metainfo: ET.Element = None
 
         self.sum_channels: bool = False
         self.remove_align: bool = True
@@ -498,7 +500,12 @@ def parse_file(self, file: str, **kwargs: Dict[str, Any]) -> List[Dict[str, Any]
         self._get_xml_context()
         self._get_xml_metainfo()
 
-        self.spectra = flatten_schedule(self.xml)
+        self.spectra = flatten_schedule(self.xml_schedule)
+
+        for spectrum in self.spectra:
+            update_dict_without_overwrite(spectrum, flatten_context(self.xml_context))
+            update_dict_without_overwrite(spectrum, flatten_metainfo(self.xml_metainfo))
+
         self._attach_node_ids()
         self._remove_empty_nodes()
         self._attach_device_protocols()
@@ -525,7 +532,7 @@ def initiate_file_connection(self, file: str):
         sql_connection = file
         self.con = sqlite3.connect(sql_connection)
 
-    def _execute_sql_query(self, con: sqlite3.Connection, query: str):
+    def _execute_sql_query(self, query: str):
         return execute_sql_query_on_con(self.con, query)
 
     def _get_version(self):
@@ -537,8 +544,11 @@ def _get_app_version(self):
         self.app_version = self._execute_sql_query(query)[0][0]
 
     def _get_xml_from_key(self, key: str):
-        query = f"SELECT Value FROM Configuration WHERE Key={key}"
-        return ET.fromstring(self._execute_sql_query(query)[0][0])
+        query = f"SELECT Value FROM Configuration WHERE Key='{key}'"
+        try:
+            return ET.fromstring(self._execute_sql_query(query)[0][0])
+        except IndexError:
+            return None
 
     def _get_xml_schedule(self):
         """Parse the schedule into an XML object."""
@@ -549,9 +559,8 @@ def _get_xml_context(self):
         self.xml_context = self._get_xml_from_key("Context")
 
     def _get_xml_metainfo(self):
-        XML = self._get_xml_from_key("MetaInfo")
-        for i in XML.iter("Parameter"):
-            self.metainfo[i.attrib["name"].replace(" ", "_")] = i.text
+        """Parse the metainfo into an XML object."""
+        self.xml_metainfo = self._get_xml_from_key("MetaInfo")
 
     def _append_scan_data(self):
         """