21944637261

Committed 12 Feb 2026 11:24AM UTC coverage: 86.453% (+0.1%) from 86.341%

Build # 21944637261

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #77 from kit-data-manager/improv_code_cleanup

Various code fixes and cleanups
Cleaner type hints
Cleaner inheritance
More None checks

parse functions (parse_setup, parse_run, parse) do not return raw parse result anymore (was never used anywhere anyway)

Code now passes a static code check such as pyright.

Run Details

159 of 198 new or added lines in 30 files covered. (80.3%)

3 existing lines in 3 files now uncovered.

2425 of 2805 relevant lines covered (86.45%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

39.73

/src/parser/impl/Atlas3dParser.py

from typing import List

from src.model.RunMD import RunMD
from src.model.SchemaConcepts.Acquisition_simplified import Acquisition
from src.parser.RunMD_Parser import RunMD_Parser
from src.parser.SetupMD_Parser import SetupMD_Parser
from src.model.SetupMD import SetupMD
from src.parser.mapping_util import map_a_dict
from src.Preprocessor import Preprocessor
from src.model.SchemaConcepts.Dataset_simplified import Dataset
from src.resources.maps.mapping import setup_zeiss
from src.util import normalize_path, input_to_dict
from src.model.SchemaConcepts.codegen.SchemaClasses_TOMO import DatasetType
from src.model.SchemaConcepts.TOMO_Image import TOMO_Image
import re


class Atlas3dParser(SetupMD_Parser, RunMD_Parser):

    @staticmethod
    def supported_input_sources() -> List[str]:
        return ["Zeiss Auriga"]

    def __init__(self):
        self.internal_mapping = input_to_dict(setup_zeiss.read_text())

    def parse_run(self, payload):
        parsed = self._read_input(payload)

        runMD = RunMD()
        
        if parsed and parsed.get("ATLAS3D-Job"):
            resultMD = parsed["ATLAS3D-Job"].get("ATLAS3D-Run")

            pattern_to_DatasetType = r"^Filename(?:[A-Z])?$"
            
            images = resultMD.get("Image")
            for imgmd in images:
                image_fields = list(imgmd.keys())
                #print("===image_fields===>",image_fields)
                matchingFilenames = [elem for elem in image_fields if re.match(pattern_to_DatasetType, elem)]
                #print("==matchingFilenames==>",matchingFilenames)
                if len(matchingFilenames) != 0:
                    for field in matchingFilenames:
    
                        if imgmd.get(field) and imgmd.get(field).split("\\")[0] in DatasetType:
                            fp = normalize_path(imgmd.get(field))
                            img = TOMO_Image(localPath=fp)
                            runMD.add_image(img, DatasetType(imgmd.get(field).split("\\")[0]))

        return runMD


    def parse_setup(self, payload):
        parsed = self._read_input(payload)

        mapping_dict = self.internal_mapping
        ac_md = map_a_dict(parsed, mapping_dict)

        Preprocessor.normalize_all_units(ac_md)

        acquisition = self._create_acquisition(ac_md)

        # Ensure datasets is always a list
        datasets = self._create_datasets(ac_md)
        #print(datasets)

        if len(datasets) == 1:
            acquisition.dataset_template = datasets[0]
        if len(datasets) > 1:
            acquisition.datasets = datasets
        return SetupMD(acquisition_metadata=acquisition)

    def _create_acquisition(self, ac_md) -> Acquisition:

        ac_md_format = {
            "genericMetadata": ac_md["genericMetadata"]
        }

        acquisition = Acquisition(**ac_md_format)
        #datasets = self._create_datasets(metadata_dict)
        #acquisition.datasets = datasets
        return acquisition

    def _create_datasets(self, ac_md) -> list:
        datasets = []
        for ds in ac_md["dataset"]:
            datasets.append(self._create_dataset(ds))
        return datasets

    def _create_dataset(self, ds_dict) -> Dataset:
        # Ensure instrument.eBeam.apertureSetting.size is properly formatted
        #if "instrument" in ds_dict and "eBeam" in ds_dict["instrument"]:
            #eBeam = ds_dict["instrument"]["eBeam"]
            #if "apertureSetting" in eBeam and "size" in eBeam["apertureSetting"]:
                #size_value = eBeam["apertureSetting"]["size"]
                #if isinstance(size_value, str):  # Convert only if it's a string
                    #eBeam["apertureSetting"]["size"] = self._parse_aperture_size(size_value)

        # Create the Dataset object with validated data
        ds = Dataset(**ds_dict)
        return ds

    def _parse_aperture_size(self, size_str):
        """
        Converts apertureSetting mapped value '[1] 30 µm (5.0 kV)' into a dictionary with value and unit.
        """
        apertureSetting_size_pattern = r"\[\d+\]\s*(\d+)\s*(µm|mm|nm)\s*(?:\[.*?\])?\s*\(([\d\.]+)\s*kV\)"
        match = re.match(apertureSetting_size_pattern, size_str)
        if match:
            return {
                "value": int(match.group(1)),
                "unit": "um"
            }
        return size_str

    @staticmethod
    def retrievable_datasets():
        return False

    @staticmethod
    def expected_input_format():
        return "text/xml"

1	from typing import List	1✔
2
3	from src.model.RunMD import RunMD	1✔
4	from src.model.SchemaConcepts.Acquisition_simplified import Acquisition	1✔
5	from src.parser.RunMD_Parser import RunMD_Parser	1✔
6	from src.parser.SetupMD_Parser import SetupMD_Parser	1✔
7	from src.model.SetupMD import SetupMD	1✔
8	from src.parser.mapping_util import map_a_dict	1✔
9	from src.Preprocessor import Preprocessor	1✔
10	from src.model.SchemaConcepts.Dataset_simplified import Dataset	1✔
11	from src.resources.maps.mapping import setup_zeiss	1✔
12	from src.util import normalize_path, input_to_dict	1✔
13	from src.model.SchemaConcepts.codegen.SchemaClasses_TOMO import DatasetType	1✔
14	from src.model.SchemaConcepts.TOMO_Image import TOMO_Image	1✔
15	import re	1✔
16
17
18	class Atlas3dParser(SetupMD_Parser, RunMD_Parser):	1✔
19
20	@staticmethod	1✔
21	def supported_input_sources() -> List[str]:	1✔
22	return ["Zeiss Auriga"]	×
23
24	def __init__(self):	1✔
25	self.internal_mapping = input_to_dict(setup_zeiss.read_text())	1✔
26
27	def parse_run(self, payload):	1✔
28	parsed = self._read_input(payload)	×
29
30	runMD = RunMD()	×
31
NEW 32	if parsed and parsed.get("ATLAS3D-Job"):	×
NEW 33	resultMD = parsed["ATLAS3D-Job"].get("ATLAS3D-Run")	×
34
NEW 35	pattern_to_DatasetType = r"^Filename(?:[A-Z])?$"	×
36
NEW 37	images = resultMD.get("Image")	×
NEW 38	for imgmd in images:	×
NEW 39	image_fields = list(imgmd.keys())	×
40	#print("===image_fields===>",image_fields)
NEW 41	matchingFilenames = [elem for elem in image_fields if re.match(pattern_to_DatasetType, elem)]	×
42	#print("==matchingFilenames==>",matchingFilenames)
NEW 43	if len(matchingFilenames) != 0:	×
NEW 44	for field in matchingFilenames:	×
45
NEW 46	if imgmd.get(field) and imgmd.get(field).split("\\")[0] in DatasetType:	×
NEW 47	fp = normalize_path(imgmd.get(field))	×
NEW 48	img = TOMO_Image(localPath=fp)	×
NEW 49	runMD.add_image(img, DatasetType(imgmd.get(field).split("\\")[0]))	×
50
NEW 51	return runMD	×
52
53
54	def parse_setup(self, payload):	1✔
UNCOV 55	parsed = self._read_input(payload)	×
56
57	mapping_dict = self.internal_mapping	×
58	ac_md = map_a_dict(parsed, mapping_dict)	×
59
60	Preprocessor.normalize_all_units(ac_md)	×
61
62	acquisition = self._create_acquisition(ac_md)	×
63
64	# Ensure datasets is always a list
65	datasets = self._create_datasets(ac_md)	×
66	#print(datasets)
67
68	if len(datasets) == 1:	×
69	acquisition.dataset_template = datasets[0]	×
NEW 70	if len(datasets) > 1:	×
71	acquisition.datasets = datasets	×
NEW 72	return SetupMD(acquisition_metadata=acquisition)	×
73
74	def _create_acquisition(self, ac_md) -> Acquisition:	1✔
75
76	ac_md_format = {	×
77	"genericMetadata": ac_md["genericMetadata"]
78	}
79
80	acquisition = Acquisition(**ac_md_format)	×
81	#datasets = self._create_datasets(metadata_dict)
82	#acquisition.datasets = datasets
83	return acquisition	×
84
85	def _create_datasets(self, ac_md) -> list:	1✔
86	datasets = []	×
87	for ds in ac_md["dataset"]:	×
88	datasets.append(self._create_dataset(ds))	×
89	return datasets	×
90
91	def _create_dataset(self, ds_dict) -> Dataset:	1✔
92	# Ensure instrument.eBeam.apertureSetting.size is properly formatted
93	#if "instrument" in ds_dict and "eBeam" in ds_dict["instrument"]:
94	#eBeam = ds_dict["instrument"]["eBeam"]
95	#if "apertureSetting" in eBeam and "size" in eBeam["apertureSetting"]:
96	#size_value = eBeam["apertureSetting"]["size"]
97	#if isinstance(size_value, str): # Convert only if it's a string
98	#eBeam["apertureSetting"]["size"] = self._parse_aperture_size(size_value)
99
100	# Create the Dataset object with validated data
101	ds = Dataset(**ds_dict)	×
102	return ds	×
103
104	def _parse_aperture_size(self, size_str):	1✔
105	"""
106	Converts apertureSetting mapped value '[1] 30 µm (5.0 kV)' into a dictionary with value and unit.
107	"""
108	apertureSetting_size_pattern = r"\[\d+\]\s(\d+)\s(µm\|mm\|nm)\s(?:\[.?\])?\s\(([\d\.]+)\skV\)"	×
109	match = re.match(apertureSetting_size_pattern, size_str)	×
110	if match:	×
111	return {	×
112	"value": int(match.group(1)),
113	"unit": "um"
114	}
115	return size_str	×
116
117	@staticmethod	1✔
118	def retrievable_datasets():	1✔
119	return False	×
120
121	@staticmethod	1✔
122	def expected_input_format():	1✔
123	return "text/xml"	×

kit-data-manager / tomo_mapper / 21944637261

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous