Skip to content

fix: fixed the indep coord linkage in EDD strainanalysis #387

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rolfverberg merged 2 commits into from
Oct 20, 2026
Merged

fix: fixed the indep coord linkage in EDD strainanalysis #387

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
82a5e3d
fix: fixed the indep coord linkage in EDD strainanalysis
rolfverberg Oct 20, 2026
a78cd0a
doc: makes nxprocess role in StrainAnalysisProcessor._get_nxprocess c…
rolfverberg Oct 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 57 additions & 84 deletions CHAP/edd/processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2651,15 +2651,13 @@ def validate_strainanalysisprocessor_before(cls, data):
return data

@staticmethod
def add_points(nxroot, points, logger=None):
def add_points(nxprocess, points, logger=None):
"""Add or update the strain analysis for a set of map points
in a
`NXroot <https://manual.nexusformat.org/classes/base_classes/NXroot.html#nxroot>`__
object.
in a NeXus `NXprocess` object.

:param nxroot: Strain analysis object to add/update the
:param nxprocess: NeXus `NXprocess` object to add/update the
points to.
:type nxroot: nexusformat.nexus.NXroot
:type nxprocess: nexusformat.nexus.NXprocess
:param points: Strain analysis results for a set of points.
:type points: list[dict[str, object]
"""
Expand All @@ -2674,14 +2672,6 @@ def add_points(nxroot, points, logger=None):
# Local modules
from CHAP.common.map_utils import get_axes

nxprocess = None
for nxobject in nxroot.values():
if isinstance(nxobject, NXprocess):
nxprocess = nxobject
break
if nxprocess is None:
raise RuntimeError('Unable to find the strainanalysis object')

nxdata_detectors = []
for nxobject in nxprocess.values():
if isinstance(nxobject, NXdetector):
Expand Down Expand Up @@ -2743,6 +2733,9 @@ def process(self, data):
:rtype: list[dict[str, object]] or
nexusformat.nexus.NXroot, PipelineData, PipelineData
"""
# Third party modules
from nexusformat.nexus import NXprocess

# Local modules
from CHAP.utils.general import list_to_string

Expand All @@ -2759,11 +2752,23 @@ def process(self, data):
self._animation = []

# Load the pipeline input data
nxroot = self.get_nxroot(self.get_data(data))
nxprocess = NXprocess()
if self.setup and not self.standalone:
# Third party modules
from nexusformat.nexus import NXprocess

# Local modules
from CHAP.utils.general import nxcopy

nxroot = nxcopy(self.get_data(data))
nxentry = self.get_default_nxentry(nxroot)

# Initialize a NXprocess under NXroot for link consistency
nxroot[f'{nxentry.nxname}_strainanalysis'] = nxprocess
else:
nxentry = self.get_default_nxentry(self.get_data(data))

# Load the detector data
# FIX set rel_height_cutoff
nxentry = self.get_default_nxentry(nxroot)
nxdata = nxentry[nxentry.default]

# Load the validated calibration configuration
Expand Down Expand Up @@ -2884,27 +2889,24 @@ def process(self, data):
if self.update and not no_raw_data:
results = self._strain_analysis()
if self.setup:
if self.standalone:
nxsetup = self._get_nxprocess(nxentry, calibration_config)
else:
nxsetup = self._get_nxroot(nxentry, calibration_config)
self._get_nxprocess(calibration_config, nxprocess)
if not self.standalone:
points = self._get_points(results)
if points:
self.logger.info(f'Adding {len(points)} points')
self.add_points(nxsetup, points, logger=self.logger)
self.add_points(nxprocess, points, logger=self.logger)
self.logger.info(f'... done')
else:
self.logger.warning('Skip adding points')
if self.standalone and self.update:
# Return nxprocess structure and values separately for
# writer
values = self._get_values(results)
ret = [nxsetup, values]
ret = [nxprocess, values]
else:
if not (self._figures or self._animation):
return nxsetup
ret = [nxsetup]
return nxroot
ret = [nxroot]
else:
result = self._get_values(results) if self.standalone \
else self._get_points(results)
Expand Down Expand Up @@ -3178,20 +3180,17 @@ def _get_values(self, results):
"""
return [{'data': v, 'path': k} for k, v in results.items()]

def _get_nxprocess(self, nxentry, calibration_config, name):
def _get_nxprocess(self, calibration_config, nxprocess):
"""Return a standalone NXprocess for the strain analysis
results & metadata.

:param nxentry: Strain analysis map, including the raw
MCA data.
:type nxentry: nexusformat.nexus.NXentry
:param calibration_config: 2&theta calibration configuration.
:type calibration_config:
CHAP.edd.models.MCATthCalibrationConfig
:param name: NXprocess name.
:type name: str
:return: Strain analysis results & associated metadata.
:rtype: nexusformat.nexus.NXprocess
:param nxprocess: NeXus style `nexusformat.nexus.NXprocess`
(wrapped in an `nexusformat.nexus.NXroot` for
non-standalone analysis)
:type nxprocess: nexusformat.nexus.NXprocess
"""
# Third party modules
# pylint: disable=no-name-in-module
Expand All @@ -3200,8 +3199,6 @@ def _get_nxprocess(self, nxentry, calibration_config, name):
NXdata,
NXdetector,
NXfield,
NXparameters,
NXprocess,
)
# pylint: enable=no-name-in-module

Expand All @@ -3218,7 +3215,6 @@ def _get_nxprocess(self, nxentry, calibration_config, name):
'StrainAnalysis Configuration, or re-run the pipeline with '
'the --interactive flag.')

nxprocess = NXprocess(name=name)
nxprocess.calibration_config = \
calibration_config.model_dump_json()
nxprocess.strain_analysis_config = \
Expand Down Expand Up @@ -3322,7 +3318,7 @@ def _get_nxprocess(self, nxentry, calibration_config, name):
ddet_nxdata.success = NXfield(
shape=[shape[0]], dtype='bool',
maxshape=(None,), chunks=(1,))

self._linkdims(ddet_nxdata, det_nxdata)

# Setup the NXcollection objects for the per material peak
# fitting results
Expand Down Expand Up @@ -3366,6 +3362,9 @@ def _get_nxprocess(self, nxentry, calibration_config, name):
'Standard deviations in strain from peak '
'fit (\u03B5)'},
maxshape=(None,), chunks=(1,)))
self._linkdims(
nxcollection['unconstrained'].strains, det_nxdata,
skip_field_dims=['energy'])

# Add the peak fit results
self._add_fit_nxcollection(
Expand All @@ -3374,32 +3373,6 @@ def _get_nxprocess(self, nxentry, calibration_config, name):
# Add the strain fields
det_nxdata.tth.nxdata = detector.get_tth_map((num_points,))

return nxprocess

def _get_nxroot(self, nxentry, calibration_config):
"""Return a `nexusformat.nexus.NXroot` object initialized for
the strain analysis.

:param nxentry: Strain analysis map, including the raw
MCA data.
:type nxentry: nexusformat.nexus.NXentry
:param calibration_config: 2&theta calibration configuration.
:type calibration_config:
CHAP.edd.models.MCATthCalibrationConfig
:return: Strain analysis results & associated metadata.
:rtype: nexusformat.nexus.NXroot
"""
# Third party modules
# pylint: disable=no-name-in-module
from nexusformat.nexus import NXroot
# pylint: enable=no-name-in-module

nxroot = NXroot()
nxroot[nxentry.nxname] = nxentry
nxprocess = self._get_nxprocess(
nxentry, calibration_config, f'{nxentry.nxname}_strainanalysis')
nxroot[f'{nxentry.nxname}_strainanalysis'] = nxprocess
return nxroot

def _linkdims(
self, nxgroup, nxdata_source, *, add_field_dims=None,
Expand Down Expand Up @@ -3630,14 +3603,15 @@ def _store_peak_fit_results(
name, i = component_peak_indices[j]
assert name == material_name
prefix = f'{name}_peak{i+1}_'
centers_errors_j = \
fit_results['best_errors'][f'{prefix}center']
results.update({
f'{path}/amplitudes/values':
fit_results['best_values'][f'{prefix}amplitude'],
f'{path}/amplitudes/errors':
fit_results['best_errors'][f'{prefix}amplitude'],
f'{path}/centers/values': centers[j],
f'{path}/centers/errors':
fit_results['best_errors'][f'{prefix}center'],
f'{path}/centers/errors': centers_errors_j,
f'{path}/sigmas/values':
fit_results['best_values'][f'{prefix}sigma'],
f'{path}/sigmas/errors':
Expand All @@ -3651,25 +3625,24 @@ def _store_peak_fit_results(
fit_results['best_errors'][f'{prefix}fraction'],
})
# Strain values: NaN where unconstrained center is zero
# has_center = centers[j].astype(bool)
# centers_errors_j = \
# fit_results['best_errors'][f'{prefix}center']
# strain_errors_raw = np.asarray([
# (e / c if e is not None else np.nan)
# for e, c in zip(centers_errors_j, centers[j])
# ])
# results.update({
# f'{path}/strains/values':
# np.where(has_center, strains[j], np.nan),
# f'{path}/strains/errors':
# np.where(has_center, strain_errors_raw, np.nan),
# f'{path}/strains/residuals':
# np.where(
# has_center,
# (component_strains[material_name]['mean'] -
# strains[j]),
# np.nan),
# })
if fit_type == 'unconstrained':
has_center = centers[j].astype(bool)
strain_errors_raw = np.asarray([
(e / c if e is not None else np.nan)
for e, c in zip(centers_errors_j, centers[j])
])
results.update({
f'{path}/strains/values':
np.where(has_center, strains[j], np.nan),
f'{path}/strains/errors':
np.where(has_center, strain_errors_raw, np.nan),
f'{path}/strains/residuals':
np.where(
has_center,
(component_strains[material_name]['mean'] -
strains[j]),
np.nan),
})

# Get and subtract the detector baselines
self._subtract_baselines()
Expand Down
Loading