Coverage for dynasor/post_processing/weights.py: 100%
48 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-10 08:27 +0000
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-10 08:27 +0000
1from typing import Any, Optional
2from pandas import DataFrame
5class Weights:
6 """
7 Class holding weights and support functions for weighting of samples.
9 Parameters
10 ----------
11 weights_coh
12 A dict with keys and values representing the atom types and their corresponding
13 coherent scattering length, ``{'A': b_A }``.
14 weights_incoh
15 A dict with keys and values representing the atom types and their corresponding
16 incoherent scattering length, ``{'A': b_A }``.
17 supports_currents
18 Whether or not the coherent weights should be applied to current-correlation functions.
19 """
21 def __init__(
22 self,
23 weights_coh: dict[str, Any],
24 weights_incoh: Optional[dict[str, Any]] = None,
25 supports_currents: Optional[bool] = True,
26 ):
27 if weights_incoh is not None:
28 if set(weights_coh.keys()) != set(weights_incoh.keys()):
29 raise ValueError(
30 'Incoherent weights keys do not match coherent weights keys: '
31 f'{sorted(weights_incoh.keys())} vs {sorted(weights_coh.keys())}'
32 )
33 self._weights_coh = weights_coh
34 self._weights_incoh = weights_incoh
35 self._supports_currents = supports_currents
37 def get_weight_coh(self, atom_type: str, q_norm: float = None) -> float:
38 """Get the coherent weight for a given atom type and q-vector norm."""
39 if atom_type not in self._weights_coh.keys():
40 raise ValueError(f'Coherent weights for {atom_type} have not been specified')
41 return self._weights_coh[atom_type]
43 def get_weight_incoh(self, atom_type: str, q_norm: float = None) -> float:
44 """Get the incoherent weight for a given atom type and q-vector norm."""
45 if self._weights_incoh is None:
46 return None
47 if atom_type not in self._weights_incoh.keys():
48 raise ValueError(f'Incoherent weights for {atom_type} have not been specified')
49 return self._weights_incoh[atom_type]
51 @property
52 def parameters(self) -> DataFrame:
53 """Parameters used for weighting.
54 """
55 atom_types = set(self._weights_coh.keys())
56 if self._weights_incoh is not None:
57 atom_types |= set(self._weights_incoh.keys())
58 data = []
59 for s in atom_types:
60 record = dict(
61 atom_type=s,
62 coherent=self._weights_coh.get(s, None),
63 )
64 data.append(record)
65 if self._weights_incoh is not None:
66 record.update(dict(incoherent=self._weights_incoh.get(s, None)))
67 return DataFrame(data).sort_values('atom_type', ignore_index=True)
69 @property
70 def supports_currents(self) -> bool:
71 """
72 Whether or not this :class:`Weights` object supports weighting of current correlations.
73 """
74 return self._supports_currents
76 @property
77 def supports_incoherent(self) -> bool:
78 """
79 Whether or not this :class:`Weights` object supports weighting of incoherent
80 correlation functions.
81 """
82 return self._weights_incoh is not None
84 def __str__(self):
85 s = ['weights coherent:']
86 for key, val in self._weights_coh.items():
87 s.append(f' {key}: {val}')
89 # Return early if incoherent weights
90 # are None
91 if self._weights_incoh is None:
92 return '\n'.join(s)
94 s.append('weights incoherent:')
95 for key, val in self._weights_incoh.items():
96 s.append(f' {key}: {val}')
97 return '\n'.join(s)