save current version

Author: lukaspie

src/pynxtools_xps/specs/sle/flatten_xml.py

@@ -7,7 +7,7 @@
 
 import re
 import copy
-from typing import Tuple, Dict, Any
+from typing import Tuple, Dict, Any, List
 
 from lxml import etree as ET
 
@@ -40,7 +40,7 @@ def extract_devices(elem: ET.Element) -> Dict[str, Any]:
 
 
 def step_profiling(elem: ET.Element) -> Dict[str, Any]:
-    settings = {}
+    settings: Dict[str, Any] = {}
 
     for setting in elem.iter():
         print(setting.tag, setting.attrib)
@@ -174,7 +174,7 @@ def _get_spectrum_metadata(spectrum: ET.Element) -> Dict[str, Any]:
 }
 
 
-def flatten_xml(xml: ET.Element) -> Dict[str, Any]:
+def flatten_xml(xml: ET.Element) -> List[Dict[str, Any]]:
     """
     Flatten the nested XML structure, keeping only the needed metadata.
 
@@ -208,7 +208,7 @@ def process_element(elem: ET.Element, settings: Dict[str, Any]):
 
     for measurement_type in MEASUREMENT_METHOD_MAP:
         for group in xml.iter(measurement_type):
-            data = {}
+            data: Dict[str, Any] = {}
             data["devices"] = []
             data["analysis_method"] = convert_measurement_method(measurement_type)
             process_element(group, data)

src/pynxtools_xps/specs/sle/sle_specs.py

@@ -83,8 +83,7 @@ class SleMapperSpecs(XPSMapper):
 
     def __init__(self):
         self.parsers = [
-            SleProdigyParserV1,
-            SleProdigyParserV4,
+            SleProdigyParser,
         ]
 
         self.versions_map = {}
@@ -425,20 +424,52 @@ def _update_xps_dict_with_spectrum(
 POSSIBLE_DATE_FORMATS: List[str] = ["%Y-%b-%d %H:%M:%S.%f"]
 
 
-class SleProdigyParser(ABC):
+class SleProdigyParser:
     """
     Generic parser without reading capabilities,
     to be used as template for implementing parsers for different versions.
     """
 
+    supported_versions = [
+        "1.2",
+        "1.8",
+        "1.9",
+        "1.10",
+        "1.11",
+        "1.12",
+        "1.13",
+        "4.63",
+        "4.64",
+        "4.65",
+        "4.66",
+        "4.67",
+        "4.68",
+        "4.69",
+        "4.70",
+        "4.71",
+        "4.72",
+        "4.73",
+        "4.100",
+    ]
+
     def __init__(self):
         self.con = ""
         self.spectra: List[Dict[str, Any]] = []
         self.xml: ET.Element = None
         self.sum_channels: bool = False
         self.remove_align: bool = True
 
-        self.encoding: List[str, float] = ["f", 4]
+        self.encodings_dtype = {
+            "short": np.int16,
+            "double": np.float64,
+            "float": np.float32,
+        }
+        self.encoding = np.float32
+
+        encodings_map: Dict[str, List[str, float]] = {
+            "double": ["d", 8],
+            "float": ["f", 4],
+        }
 
     def initiate_file_connection(self, filepath: str):
         """Set the sqllite connection of the file to be opened."""
@@ -468,17 +499,17 @@ def parse_file(
             Flat list of dictionaries containing one spectrum each.
 
         """
-        if "remove_align" in kwargs:
-            self.remove_align = kwargs["remove_align"]
-
-        if "sum_channels" in kwargs:
-            self.sum_channels = kwargs["sum_channels"]
+        self.remove_align = kwargs.get("remove_align", True)
+        self.sum_channels = kwargs.get("sum_channels", False)
 
         # initiate connection to sql file
         self.initiate_file_connection(filepath)
 
         # read and parse sle file
         self._get_xml_schedule()
+        self._get_xml_schedule()
+        self._get_xml_schedule()
+
         self.spectra = flatten_xml(self.xml)
         self._attach_node_ids()
         self._remove_empty_nodes()
@@ -501,12 +532,32 @@ def parse_file(
 
         return self.spectra
 
-    def _get_xml_schedule(self):
-        """Parse the schedule into an XML object."""
+    def _addVersion(self):
+        self.cursor.execute('SELECT Value FROM Configuration WHERE Key="Version"')
+        self.version = self.cursor.fetchone()[0]
+
+    def _addAppVersion(self):
+        self.cursor.execute('SELECT Value FROM Configuration WHERE Key="AppVersion"')
+        self.app_version = self.cursor.fetchone()[0]
+
+    def _get_xml_from_key(key: str):
         cur = self.con.cursor()
-        query = 'SELECT Value FROM Configuration WHERE Key="Schedule"'
+        query = f"SELECT Value FROM Configuration WHERE Key={key}"
         cur.execute(query)
-        self.xml = ET.fromstring(cur.fetchall()[0][0])
+        return ET.fromstring(self.cursor.fetchone()[0])
+
+    def _get_xml_schedule(self):
+        """Parse the schedule into an XML object."""
+        self.xml_schedule = _get_xml_from_key("Schedule")
+
+    def _get_xml_context(self):
+        """Parse the context into an XML object."""
+        self.xml_context = _get_xml_from_key("Context")
+
+    def _get_xml_metainfo(self):
+        XML = _get_xml_from_key("MetaInfo")
+        for i in XML.iter("Parameter"):
+            self.metainfo[i.attrib["name"].replace(" ", "_")] = i.text
 
     def _append_scan_data(self):
         """
@@ -1342,11 +1393,6 @@ def _check_encoding(self):
         cur.execute(query)
         data, chunksize = cur.fetchall()[0]
 
-        encodings_map: Dict[str, List[str, float]] = {
-            "double": ["d", 8],
-            "float": ["f", 4],
-        }
-
         if data / chunksize == 4:
             self.encoding = encodings_map["float"]
         elif data / chunksize == 8:
@@ -1445,32 +1491,3 @@ def get_sle_version(self) -> str:
         cur.execute(query)
         version = cur.fetchall()[0][0]
         return version
-
-
-class SleProdigyParserV1(SleProdigyParser):
-    """
-    Parser for SLE version 1.
-    """
-
-    supported_versions = ["1.2", "1.8", "1.9", "1.10", "1.11", "1.12", "1.13"]
-
-
-class SleProdigyParserV4(SleProdigyParser):
-    """
-    Parser for SLE version 4.
-    """
-
-    supported_versions = [
-        "4.63",
-        "4.64",
-        "4.65",
-        "4.66",
-        "4.67",
-        "4.68",
-        "4.69",
-        "4.70",
-        "4.71",
-        "4.72",
-        "4.73",
-        "4.100",
-    ]

src/pynxtools_xps/specs/xml/xml_specs.py

@@ -22,7 +22,7 @@
 #
 import re
 import xml.etree.ElementTree as ET
-from typing import Tuple, List, Any
+from typing import Tuple, List, Any, Dict
 import copy
 import xarray as xr
 import numpy as np
@@ -265,7 +265,7 @@ def parse_file(self, file: str, **kwargs):
         ----------
         """
         root_element = ET.parse(file).getroot()
-        root_element.attrib[self.child_nm_reslvers] = []
+        root_element.attrib[self.child_nm_reslvers] = []  # type: ignore[assignment]
         child_num = len(root_element)
         parent_path = self._root_path
         skip_child = -1

src/pynxtools_xps/specs/xy/xy_specs.py

@@ -851,10 +851,4 @@ def _parse_datetime(self, date: str) -> str:
 
         possible_time_formats = ["%m/%d/%y %H:%M:%S"]
 
-        date_object = datetime.datetime.strptime(date, "%m/%d/%y %H:%M:%S").replace(
-            tzinfo=tz
-        )
-
-        return date_object.isoformat()
-
-        return parse_datetime(date, possible_time_formats, tzinfo)
+        return parse_datetime(date, possible_time_formats, tz)