diff --git a/sasdata/metadata.py b/sasdata/metadata.py index d53c3102c..304446a22 100644 --- a/sasdata/metadata.py +++ b/sasdata/metadata.py @@ -1,7 +1,7 @@ """ Contains classes describing the metadata for a scattering run -The metadata is structures around the CANSas format version 1.1, found at +The metadata is structured around the CANSas format version 1.1, found at https://www.cansas.org/formats/canSAS1d/1.1/doc/specification.html Metadata from other file formats should be massaged to fit into the data classes presented here. @@ -142,6 +142,7 @@ def as_h5(self, group: h5py.Group): if self.slit_length: self.slit_length.as_h5(group, "slit_length") + @dataclass(kw_only=True) class Aperture: distance: Quantity[float] | None @@ -168,7 +169,6 @@ def from_json(obj): type_=obj["type"], ) - def as_h5(self, group: h5py.Group): """Export data onto an HDF5 group""" if self.distance is not None: @@ -184,7 +184,6 @@ def as_h5(self, group: h5py.Group): size_group.attrs["name"] = self.size_name - @dataclass(kw_only=True) class Collimation: """ @@ -281,9 +280,6 @@ def as_h5(self, group: h5py.Group): self.wavelength_spread.as_h5(group, "wavelength_spread") - - - @dataclass(kw_only=True) class Sample: """ @@ -458,14 +454,13 @@ def to_string(self, header=""): ) else: attributes = "" - if self.contents: - if type(self.contents) is str: - children = f"\n{header} {self.contents}" - else: + match self.contents: + case list() | tuple(): children = "".join([n.to_string(header + " ") for n in self.contents]) - else: - children = "" - + case None: + children = "" + case _: + children = f"\n{header} {self.contents}" return f"\n{header}{self.name}:{attributes}{children}" def filter(self, name: str) -> list[ndarray | Quantity | str]: @@ -573,7 +568,7 @@ def id_header(self): title = "" if self.title is not None: title = self.title - return f"{title}:{",".join(self.run)}" + return f"{title}:{','.join(self.run)}" def as_h5(self, f: h5py.Group): """Export data onto an HDF5 group""" diff --git a/sasdata/trend.py b/sasdata/trend.py index 9b1a371a4..2c8f7ae10 100644 --- a/sasdata/trend.py +++ b/sasdata/trend.py @@ -1,48 +1,182 @@ +import logging from dataclasses import dataclass import numpy as np from sasdata.data import SasData -from sasdata.data_backing import Dataset, Group from sasdata.quantities.quantity import Quantity from sasdata.transforms.rebinning import calculate_interpolation_matrix_1d +logger = logging.getLogger(__name__) + # Axis strs refer to the name of their associated NamedQuantity. -# TODO: This probably shouldn't be here but will keep it here for now. + +# TODO: This probably shouldn't be here but will keep it here for now. --> In sasdta/data.py? +# TODO: Similarity/relation to __getitem__ in SasData class? +# TODO: Or a method of Metadata class? # TODO: Not sure how to type hint the return. def get_metadatum_from_path(data: SasData, metadata_path: list[str]): - current_group = data._raw_metadata + current_node = data.metadata.raw for path_item in metadata_path: - current_item = current_group.children.get(path_item, None) - if current_item is None or (isinstance(current_item, Dataset) and path_item != metadata_path[-1]): - raise ValueError('Path does not lead to valid a metadatum.') - elif isinstance(current_item, Group): - current_group = current_item - else: - return current_item.data - raise ValueError('End of path without finding a dataset.') + current_item = None + + if isinstance(current_node.contents, list): + # Search through list of MetaNodes + for node in current_node.contents: + if node.name == path_item: + current_item = node + break + + # If we did not find the item (either not a list or not found in list) + if current_item is None: + raise ValueError("Path does not lead to a valid metadatum.") + + # Check if we're at the end of the path + if path_item == metadata_path[-1]: + return current_item.contents + + current_node = current_item + raise ValueError("End of path without finding a dataset.") @dataclass class Trend: data: list[SasData] - # This is going to be a path to a specific metadatum. - # - # TODO: But what if the trend axis will be a particular NamedQuantity? Will probably need to think on this. - trend_axis: list[str] - - # Designed to take in a particular value of the trend axis, and return the SasData object that matches it. - # TODO: Not exaclty sure what item's type will be. It could depend on where it is pointing to. - def __getitem__(self, item) -> SasData: - for datum in self.data: - metadatum = get_metadatum_from_path(datum, self.trend_axis) - if metadatum == item: - return datum - raise KeyError() + trend_axes: dict[str, list[str] | list] # Path or manual values + + def __post_init__(self): + + # First, filter out invalid data items + self._filter_and_validate_data() + + # Validate data length matches manual value lists + self._validate_manual_values() + + # Validate metadata paths + self._validate_metadata_paths() + + def _filter_and_validate_data(self): + """Filter out non-SasData objects and validate data integrity""" + valid_data = [] + invalid_indices = [] + + for i, datum in enumerate(self.data): + if not isinstance(datum, SasData): + invalid_indices.append(i) + continue + + # Check if datum has metadata + if not hasattr(datum, "metadata") or datum.metadata is None: + invalid_indices.append(i) + continue + + # Check if datum has raw metadata + if not hasattr(datum.metadata, "raw") or datum.metadata.raw is None: + invalid_indices.append(i) + continue + + valid_data.append(datum) + + # Update data with only valid items + self.data = valid_data + + # Warn about filtered items + if invalid_indices: + logger.warning( + f"Warning: Removed data items at indices {invalid_indices} - not SasData objects or missing/invalid metadata" + ) + + # Additional validation + if not self.data: + raise ValueError("No valid data items remain after filtering") + + if len(self.data) < 2: + logger.warning(f"Only {len(self.data)} valid data items remain") + + # TODO: Decide if these limitations are ok or not (e.g. Should the user be able + # to specify manual values that are not numbers? Or have a different number of + # manual values than data items? How to assign the values then?, etc.) + def _validate_manual_values(self): + """Ensure manual value lists are valid and match data length""" + + for axis_name, axis_config in self.trend_axes.items(): + # Only validate if this is a manual value axis (not a metadata path) + if isinstance(axis_config, list) and len(axis_config) > 0 and isinstance(axis_config[0], str): + # This is a metadata path, skip manual value validation + continue + + if not isinstance(axis_config, list): + raise ValueError( + f"Manual values for axis '{axis_name}' should be passed as a list, got {type(axis_config).__name__}" + ) + + if len(axis_config) == 0: + raise ValueError(f"Manual values for axis '{axis_name}' must not be empty") + + if not all(isinstance(v, (int, float)) for v in axis_config): + raise ValueError(f"All values for axis '{axis_name}' must be numbers (int or float)") + + if len(axis_config) != len(self.data): + raise ValueError( + f"Manual values for axis '{axis_name}' must have same length as data " + f"({len(self.data)} items, got {len(axis_config)})" + ) + + def _validate_metadata_paths(self): + """Validate metadata paths""" + for axis_name, axis_config in self.trend_axes.items(): + if isinstance(axis_config, list) and len(axis_config) > 0 and isinstance(axis_config[0], str): + # This is a metadata path + for i, datum in enumerate(self.data): + try: + get_metadatum_from_path(datum, axis_config) + except ValueError as e: + raise ValueError(f"trend_axes['{axis_name}'] path {axis_config} invalid for data item {i}: {e}") + + def get_trend_values(self, axis_name: str) -> list: + """Get values for a named trend axis""" + if axis_name not in self.trend_axes: + raise KeyError(f"Axis '{axis_name}' not found") + + axis_config = self.trend_axes[axis_name] + + if isinstance(axis_config[0], str): + # Metadata path - extract from data + return [get_metadatum_from_path(datum, axis_config) for datum in self.data] + else: + # Manual values - return as-is + return axis_config.copy() # Return copy to prevent modification + + def add_manual_axis(self, axis_name: str, values: list): + """Add a new manual trend axis""" + if len(values) != len(self.data): + raise ValueError(f"Manual values must have same length as data ({len(self.data)} items, got {len(values)})") + + self.trend_axes[axis_name] = values.copy() + + def add_metadata_axis(self, axis_name: str, path: list[str]): + """Add a new metadata trend axis""" + # Validate the path first + for i, datum in enumerate(self.data): + try: + get_metadatum_from_path(datum, path) + except ValueError as e: + raise ValueError(f"Path {path} invalid for data item {i}: {e}") + + self.trend_axes[axis_name] = path + @property - def trend_axes(self) -> list[float]: - return [get_metadatum_from_path(datum, self.trend_axis) for datum in self.data] + def axis_names(self) -> list[str]: + return list(self.trend_axes.keys()) + + def is_manual_axis(self, axis_name: str) -> bool: + """Check if an axis uses manual values or metadata path""" + if axis_name not in self.trend_axes: + raise KeyError(f"Axis '{axis_name}' not found") + + axis_config = self.trend_axes[axis_name] + return not (isinstance(axis_config, list) and len(axis_config) > 0 and isinstance(axis_config[0], str)) # TODO: Assumes there are at least 2 items in data. Is this reasonable to assume? Should there be error handling for # situations where this may not be the case? @@ -84,6 +218,5 @@ def interpolate(self, axis: str) -> "Trend": metadata=datum.metadata, ) new_data.append(new_datum) - new_trend = Trend(new_data, - self.trend_axis) + new_trend = Trend(new_data, self.trend_axes) return new_trend diff --git a/test/trend_test_data/hdf_test_files/nxcansas_1Dand2D_multisasentry.h5 b/test/trend_test_data/hdf_test_files/nxcansas_1Dand2D_multisasentry.h5 new file mode 100644 index 000000000..fdce32035 Binary files /dev/null and b/test/trend_test_data/hdf_test_files/nxcansas_1Dand2D_multisasentry.h5 differ diff --git a/test/trend_test_data/xml_test_files/cansas1d_notitle.xml b/test/trend_test_data/xml_test_files/cansas1d_notitle.xml new file mode 100644 index 000000000..2419676b5 --- /dev/null +++ b/test/trend_test_data/xml_test_files/cansas1d_notitle.xml @@ -0,0 +1,140 @@ + + + + + 1234 + + + 0.02 + 1000 + 3 + 0.01 + + + + + 0.03 + 1001 + 4 + 0.02 + + + + + + SI600-new-long + 1.03 + 0.327 + 0.0000 + + 10.00 + 0.00 + + + 22.5 + 0.020 + +
+ + http://chemtools.chem.soton.ac.uk/projects/blog/blogs.php/bit_id/2720 +
+
+ Some text here +
+ + + + canSAS instrument + + neutron + + 12.00 + 13.00 + + disc + 6.00 + 0.22 + 1.00 + + 14.3 + + + + 123.0 + + + 50 + + 11.000 + + + + 1 + + + + + fictional hybrid + + + 4.150 + + + 1 + 2 + + + 1.00 + 0.00 + 0.00 + + + 322.64 + 327.68 + + + 5.00 + 5.00 + + + + + spol + 04-Sep-2007 18:35:02 + 10.000 + 180.0 + 0.0 + USER:MASK.COM + + AvA1 0.0000E+00 AsA2 1.0000E+00 XvA3 1.0526E+03 XsA4 + 5.2200E-02 XfA5 0.0000E+00 + + + S... 13597 0 2.26E+02 2A 5mM 0%D2O Sbak 13594 0 1.13E+02 + H2O Buffer + + V... 13552 3 1.00E+00 H2O5m + + + NCNR-IGOR + 03-SEP-2006 11:42:47 + + Circular + SEP06064.SA3_AJJ_L205 + SEP06064.SA3_AJJ_L205 + SEP06064.SA3_AJJ_L205 + SEP06064.SA3_AJJ_L205 + SEP06064.SA3_AJJ_L205 + 1 + 1 + 230.09 + 1 + No Information + + + + diff --git a/test/utest_trend.py b/test/utest_trend.py index b079bf53c..8be92227e 100644 --- a/test/utest_trend.py +++ b/test/utest_trend.py @@ -1,69 +1,151 @@ from os import listdir, path +import numpy as np import pytest import sasdata.temp_ascii_reader as ascii_reader from sasdata.ascii_reader_metadata import AsciiMetadataCategory from sasdata.quantities.units import per_angstrom, per_nanometer from sasdata.temp_ascii_reader import AsciiReaderParams +from sasdata.temp_hdf5_reader import load_data as hdf_load_data +from sasdata.temp_xml_reader import load_data as xml_load_data from sasdata.trend import Trend mumag_test_directories = [ - 'FeNiB_perpendicular_Bersweiler_et_al', - 'Nanoperm_perpendicular_Honecker_et_al', - 'NdFeB_parallel_Bick_et_al' + "FeNiB_perpendicular_Bersweiler_et_al", + "Nanoperm_perpendicular_Honecker_et_al", + "NdFeB_parallel_Bick_et_al", ] -custom_test_directory = 'custom_test' +xml_file = path.join(path.dirname(__file__), "trend_test_data", "xml_test_files", "cansas1d_notitle.xml") + +hdf_file = path.join(path.dirname(__file__), "trend_test_data", "hdf_test_files", "nxcansas_1Dand2D_multisasentry.h5") + + +custom_test_directory = "custom_test" + def get_files_to_load(directory_name: str) -> list[str]: - load_from = path.join(path.dirname(__file__), 'trend_test_data', directory_name) + load_from = path.join(path.dirname(__file__), "trend_test_data", directory_name) base_filenames_to_load = listdir(load_from) files_to_load = [path.join(load_from, basename) for basename in base_filenames_to_load] return files_to_load -@pytest.mark.parametrize('directory_name', mumag_test_directories) -def test_trend_build_interpolate(directory_name: str): - """Try to build a trend object on the MuMag datasets""" + +def test_trend_build_from_xml(): + """ + Try to build a trend object from an XML file. + The loader returns a dict of SasData objects (here only one, named 'SasData01') + and currently a single Metadata (no tree of MetaNodes), but this should be + corrected in the future in the XML reader, I guess! + """ + data = xml_load_data(xml_file) # dict of SasData objects + trend = Trend(data=[data["SasData01"]], trend_axes={"entry": ["SASentry"]}) + assert not trend.is_manual_axis("entry") + assert len(trend.get_trend_values("entry")) == 1 + + +def test_trend_build_from_hdf5_with_multiple_axes(): + """ + Try to build a trend object with more than one axisfrom an HDF5 file. + The loader returns a dict of SasData objects (here two, named 'sasentry01' and 'sasentry02') + Need to think how to compare the Quantity objects + """ + data = hdf_load_data(hdf_file) # dict of SasData objects + trend = Trend( + data=list(data.values()), + trend_axes={ + "run_number": ["run"], + "title": ["title"], + "SDD1": ["sasinstrument", "sasdetector01", "SDD"], + "SDD2": ["sasinstrument", "sasdetector02", "SDD"], + "transmission": ["sastransmission_spectrum01", "T"], + "wavelength": ["sastransmission_spectrum01", "lambda"], + }, + ) + assert not trend.is_manual_axis("run_number") + assert trend.get_trend_values("run_number") == ["33837", "33837"] + assert trend.get_trend_values("title") == ["MH4_5deg_16T_SLOW", "MH4_5deg_16T_SLOW"] + assert np.allclose(trend.get_trend_values("transmission")[0], trend.get_trend_values("transmission")[1]) + assert np.allclose(trend.get_trend_values("wavelength")[0], trend.get_trend_values("wavelength")[1]) + + +@pytest.mark.parametrize("directory_name", [mumag_test_directories[1]]) +def test_trend_build_from_ascii_with_manual_axis(directory_name: str): + """ + Try to build a trend object from an ASCII file including a manual axis. + """ files_to_load = get_files_to_load(directory_name) params = AsciiReaderParams( filenames=files_to_load, - columns=[('Q', per_nanometer), ('I', per_nanometer), ('dI', per_nanometer)], + columns=[("Q", per_nanometer), ("I", per_nanometer), ("dI", per_nanometer)], ) - params.separator_dict['Whitespace'] = True - params.metadata.master_metadata['magnetic'] = AsciiMetadataCategory( + params.separator_dict["Whitespace"] = True + params.metadata.master_metadata["magnetic"] = AsciiMetadataCategory( values={ - 'counting_index': 0, - 'applied_magnetic_field': 1, - 'saturation_magnetization': 2, - 'demagnetizing_field': 3 + "counting_index": 0, + "applied_magnetic_field": 1, + "saturation_magnetization": 2, + "demagnetizing_field": 3, } ) data = ascii_reader.load_data(params) trend = Trend( data=data, - trend_axis=['magnetic', 'applied_magnetic_field'] + trend_axes={ + "index": ["magnetic", "counting_index"], + "field": ["magnetic", "applied_magnetic_field"], + "manual_temp": np.linspace(300, 350, len(data)).tolist(), + }, + ) + assert not trend.is_manual_axis("index") + assert trend.is_manual_axis("manual_temp") + + +@pytest.mark.parametrize("directory_name", mumag_test_directories) +def test_trend_build_interpolate(directory_name: str): + """ + Try to build a trend object on the MuMag datasets. + and interpolates the data to match the Q axes. + Maybe confusing to have here data axes ('Q', 'I', 'dI') + and trend axes ('field')? + """ + files_to_load = get_files_to_load(directory_name) + params = AsciiReaderParams( + filenames=files_to_load, + columns=[("Q", per_nanometer), ("I", per_nanometer), ("dI", per_nanometer)], ) - # Initially, the q axes in this date don't exactly match - to_interpolate_on = 'Q' + params.separator_dict["Whitespace"] = True + params.metadata.master_metadata["magnetic"] = AsciiMetadataCategory( + values={ + "counting_index": 0, + "applied_magnetic_field": 1, + "saturation_magnetization": 2, + "demagnetizing_field": 3, + } + ) + data = ascii_reader.load_data(params) + trend = Trend(data=data, trend_axes={"field": ["magnetic", "applied_magnetic_field"]}) + # Initially, the q axes in this data don't exactly match + to_interpolate_on = "Q" assert not trend.all_axis_match(to_interpolate_on) interpolated_trend = trend.interpolate(to_interpolate_on) assert interpolated_trend.all_axis_match(to_interpolate_on) + def test_trend_q_axis_match(): + """ + Try to build a trend object on the custom test dataset + and check if the Q axes match. + But the file contents are skipped, so 'Q' and 'I' are zeroes! + """ files_to_load = get_files_to_load(custom_test_directory) - params = AsciiReaderParams( - filenames=files_to_load, - columns=[('Q', per_angstrom), ('I', per_angstrom)] - ) - params.metadata.master_metadata['magnetic'] = AsciiMetadataCategory( + params = AsciiReaderParams(filenames=files_to_load, columns=[("Q", per_angstrom), ("I", per_angstrom)]) + params.metadata.master_metadata["magnetic"] = AsciiMetadataCategory( values={ - 'counting_index': 0, + "counting_index": 0, } ) data = ascii_reader.load_data(params) - trend = Trend( - data=data, - trend_axis=['magnetic', 'counting_index'] - ) - assert trend.all_axis_match('Q') + trend = Trend(data=data, trend_axes={"index": ["magnetic", "counting_index"]}) + assert trend.all_axis_match("Q")