|
| 1 | +"""Data test.""" |
1 | 2 | import os |
2 | 3 | import glob |
3 | 4 | import pytest |
4 | | -import yaml |
5 | 5 | from pathlib import Path |
6 | | -from typing import Dict, Any, Union, List |
7 | 6 |
|
8 | | -from catcore.datamodel import catcore |
| 7 | +import src.catcore.datamodel.catcore |
9 | 8 | from linkml_runtime.loaders import yaml_loader |
10 | 9 |
|
11 | 10 | DATA_DIR_VALID = Path(__file__).parent / "data" / "valid" |
|
14 | 13 | VALID_EXAMPLE_FILES = glob.glob(os.path.join(DATA_DIR_VALID, '*.yaml')) |
15 | 14 | INVALID_EXAMPLE_FILES = glob.glob(os.path.join(DATA_DIR_INVALID, '*.yaml')) |
16 | 15 |
|
17 | | -# Mapping from identifier patterns or context to concrete class names |
18 | | -CHARACTERIZATION_TECHNIQUE_MAP = { |
19 | | - 'xray_source': { |
20 | | - 'Cu Kalpha': 'PowderXRD', |
21 | | - 'Al Kalpha': 'XPS', |
22 | | - }, |
23 | | - 'adsorbate_gas': 'BET', |
24 | | - 'reducing_gas_composition': 'TPR', |
25 | | - 'oxidizing_gas_composition': 'TPO', |
26 | | - 'excitation_laser_wavelength': 'RamanSpectroscopy', |
27 | | - 'minimum_wavenumber': 'InfraredSpectroscopy', |
28 | | - 'element_analyzed': 'XRayAbsorptionSpectroscopy', |
29 | | - 'nucleus': 'NMRSpectroscopy', |
30 | | - 'gun_type': 'TransmissionElectronMicroscopy', |
31 | | - 'image_resolution': 'ScanningElectronMicroscopy', |
32 | | - 'initial_temperature': 'Thermogravimetry', |
33 | | - 'combustion_temperature': 'ElementalAnalysis', |
34 | | - 'minimum_wavelength': 'UVVisSpectroscopy', |
35 | | - 'adsorption_gas': 'DRIFTS', |
36 | | - 'scan_rate': 'CyclicVoltammetry', |
37 | | - 'light_wavelength': 'DynamicLightScattering', |
38 | | - 'spray_voltage': 'ESI_MS', |
39 | | - 'excitation_wavelength': 'PhotoluminescenceSpectroscopy', |
40 | | - 'lifetime_fitting_model': 'PhotoluminescenceLifetime', |
41 | | - 'eluent': 'SizeExclusionChromatography', |
42 | | - 'gradient_program': 'HPLC_MS', |
43 | | - 'primary_energy': 'EDX', |
44 | | - 'electrode_configuration': 'ConductivityMeasurement', |
45 | | -} |
46 | | - |
47 | | -PREPARATION_METHOD_MAP = { |
48 | | - 'impregnation_type': 'Impregnation', |
49 | | - 'precipitating_agent': 'CoPrecipitation', |
50 | | - 'hydrolysis_ratio': 'SolGel', |
51 | | - 'filling_volume': 'Solvothermal', |
52 | | - 'plasma_type': 'PlasmaAssisted', |
53 | | - 'fuel': 'CombustionSynthesis', |
54 | | - 'substrate': 'AtomicLayerDeposition', |
55 | | - 'microwave_frequency': 'MicrowaveAssisted', |
56 | | - 'sonication_power': 'SonochemicalSynthesis', |
57 | | - 'flame_type': 'FlameSprayPyrolysis', |
58 | | - 'ball_material': 'MechanochemicalSynthesis', |
59 | | - 'reaction_vessel': 'MolecularSynthesis', |
60 | | -} |
61 | | - |
62 | | -SIMULATION_METHOD_MAP = { |
63 | | - 'exchange_correlation_functional': 'DFT', |
64 | | - 'force_field': 'MolecularDynamics', |
65 | | - 'rate_constants': 'Microkinetics', |
66 | | - 'interaction_potential': 'MonteCarlo', |
67 | | -} |
68 | | - |
69 | | -CALCULATED_PROPERTY_MAP = { |
70 | | - 'formation_energy': 'ThermodynamicStability', |
71 | | - 'piezoelectric_tensor': 'Piezoelectricity', |
72 | | - 'elastic_tensor': 'ElasticConstants', |
73 | | - 'surface_energy': 'Surfaces', |
74 | | - 'band_path': 'ElectronicStructure', |
75 | | - 'polarization_direction': 'Ferroelectrics', |
76 | | - 'direct_indirect': 'BandGap', |
77 | | - 'material_composition': 'DielectricTensors', |
78 | | - 'force_constant_method': 'PhononDispersion', |
79 | | - 'fit_method': 'EquationsOfState', |
80 | | - 'ph_range': 'AqueousStability', |
81 | | - 'grain_boundary_plane': 'GrainBoundaries', |
82 | | -} |
83 | | - |
84 | | -REACTOR_DESIGN_MAP = { |
85 | | - 'gas_distributor_type': 'FluidizedBedReactor', |
86 | | -} |
87 | | - |
88 | | -# Default classes for abstract types when no specific fields are found |
89 | | -DEFAULT_CLASSES = { |
90 | | - 'reactor_design_type': 'FixedBedReactor', |
91 | | - 'product_identification_method': 'GCMS', |
92 | | -} |
93 | | - |
94 | | - |
95 | | -def infer_class_type(data: Dict[str, Any], type_map: Dict[str, Any], default_class: str = None) -> str: |
96 | | - """Infer the concrete class type based on present fields.""" |
97 | | - for key, class_name in type_map.items(): |
98 | | - if key in data: |
99 | | - if isinstance(class_name, dict): |
100 | | - # Need to check value |
101 | | - for value_pattern, cn in class_name.items(): |
102 | | - if value_pattern in str(data[key]): |
103 | | - return cn |
104 | | - else: |
105 | | - return class_name |
106 | | - |
107 | | - # Return default class if no specific fields found |
108 | | - return default_class |
109 | | - |
110 | | - |
111 | | -def instantiate_polymorphic_objects(data: Union[Dict, List], parent_key: str = None) -> Union[Dict, List]: |
112 | | - """Recursively instantiate concrete classes for polymorphic fields.""" |
113 | | - if isinstance(data, list): |
114 | | - return [instantiate_polymorphic_objects(item, parent_key) for item in data] |
115 | | - |
116 | | - if not isinstance(data, dict): |
117 | | - return data |
118 | | - |
119 | | - # Check for explicit type hint (both 'type' and '@type' for LinkML compatibility) |
120 | | - explicit_type = data.get('type') or data.get('@type') |
121 | | - |
122 | | - # Recursively process nested structures first |
123 | | - result = {} |
124 | | - for key, value in data.items(): |
125 | | - if key in ('type', '@type'): # Skip the type hint fields |
126 | | - continue |
127 | | - result[key] = instantiate_polymorphic_objects(value, key) |
128 | | - |
129 | | - # Now handle polymorphic instantiation for specific keys |
130 | | - if parent_key == 'characterization_technique': |
131 | | - class_name = explicit_type or infer_class_type(result, CHARACTERIZATION_TECHNIQUE_MAP) |
132 | | - if class_name: |
133 | | - cls = getattr(catcore, class_name) |
134 | | - return cls(**result) |
135 | | - |
136 | | - elif parent_key == 'preparation_method': |
137 | | - class_name = explicit_type or infer_class_type(result, PREPARATION_METHOD_MAP) |
138 | | - if class_name: |
139 | | - cls = getattr(catcore, class_name) |
140 | | - return cls(**result) |
141 | | - |
142 | | - elif parent_key == 'simulation_method': |
143 | | - class_name = explicit_type or infer_class_type(result, SIMULATION_METHOD_MAP) |
144 | | - if class_name: |
145 | | - cls = getattr(catcore, class_name) |
146 | | - return cls(**result) |
147 | | - |
148 | | - elif parent_key == 'calculated_property': |
149 | | - class_name = explicit_type or infer_class_type(result, CALCULATED_PROPERTY_MAP) |
150 | | - if class_name: |
151 | | - cls = getattr(catcore, class_name) |
152 | | - return cls(**result) |
153 | | - |
154 | | - elif parent_key == 'reactor_design_type': |
155 | | - class_name = explicit_type or infer_class_type(result, REACTOR_DESIGN_MAP, |
156 | | - DEFAULT_CLASSES.get('reactor_design_type')) |
157 | | - if class_name: |
158 | | - cls = getattr(catcore, class_name) |
159 | | - return cls(**result) |
160 | | - |
161 | | - elif parent_key == 'product_identification_method': |
162 | | - # For now, use GCMS as default if no specific fields found |
163 | | - class_name = explicit_type or DEFAULT_CLASSES.get('product_identification_method') |
164 | | - if class_name: |
165 | | - cls = getattr(catcore, class_name) |
166 | | - return cls(**result) |
167 | | - |
168 | | - return result |
169 | | - |
170 | 16 |
|
171 | 17 | @pytest.mark.parametrize("filepath", VALID_EXAMPLE_FILES) |
172 | 18 | def test_valid_data_files(filepath): |
173 | 19 | """Test loading of all valid data files.""" |
174 | 20 | target_class_name = Path(filepath).stem.split("-")[0] |
175 | | - tgt_class = getattr(catcore, target_class_name) |
176 | | - |
177 | | - # Load the YAML content |
178 | | - with open(filepath, 'r') as f: |
179 | | - data_dict = yaml.safe_load(f) |
180 | | - |
181 | | - # Handle polymorphic fields by instantiating concrete classes |
182 | | - poly_fields = [ |
183 | | - 'characterization_technique', |
184 | | - 'preparation_method', |
185 | | - 'simulation_method', |
186 | | - 'calculated_property', |
187 | | - 'reactor_design_type', |
188 | | - 'product_identification_method', |
189 | | - 'operation_parameters', |
190 | | - 'precursor' |
191 | | - ] |
192 | | - |
193 | | - for poly_field in poly_fields: |
194 | | - if poly_field in data_dict and data_dict[poly_field]: |
195 | | - data_dict[poly_field] = instantiate_polymorphic_objects( |
196 | | - data_dict[poly_field], |
197 | | - poly_field |
198 | | - ) |
199 | | - |
200 | | - # Instantiate the target class |
201 | | - obj = tgt_class(**data_dict) |
202 | | - |
| 21 | + tgt_class = getattr( |
| 22 | + src.catcore.datamodel.catcore, |
| 23 | + target_class_name, |
| 24 | + ) |
| 25 | + obj = yaml_loader.load(filepath, target_class=tgt_class) |
203 | 26 | assert obj |
0 commit comments