Skip to content

links

GridSearchLinearModel dataclass

Bases: Parameters, LinkModelInterface

Search parameter grid space for the maxima.

Implements grid search for best fitting parameters. We don't solve a bunch of MCF problems like Pepe and Lanari (2006) but instead solve this link-by-link.

Max: |exp(j * (A * x - b))|^2 s.t: ranges[i][0] <= x_i <= ranges[i][1]

Source code in src/spurt/links/_grid_search.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class GridSearchLinearModel(Parameters, LinkModelInterface):
    """Search parameter grid space for the maxima.

    Implements grid search for best fitting parameters.
    We don't solve a bunch of MCF problems like Pepe and Lanari (2006) but
    instead solve this link-by-link.

    Max: |exp(j * (A * x - b))|^2
    s.t: ranges[i][0] <= x_i <= ranges[i][1]
    """

    @property
    def nobs(self) -> int:
        return self.matrix.shape[0]

    @property
    def ndim(self) -> int:
        return self.matrix.shape[1]

    def fwd_model(self, x: np.ndarray) -> np.ndarray:
        return np.dot(self.matrix, x)

    def estimate_model(
        self,
        wrapdata: np.ndarray,
        weights: np.ndarray | float | None = None,
    ) -> tuple[np.ndarray, float]:
        """Fit model parameters via grid search followed by Nelder-Mead optimization.

        Parameters
        ----------
        wrapdata: np.ndarray
            Real-valued array of wrapped phase gradient
        weights: np.ndarray | None
            Real-valued weights - assumed to be normalized to 1.

        Returns
        -------
        params: np.ndarray
            1D array of length ndim
        coh: float
            Temporal coherence
        """
        if wrapdata.ndim != 1:
            errmsg = f"Input data must be a 1D array. Got {wrapdata.shape}."
            raise ValueError(errmsg)

        if weights is None:
            weights = 1.0 / self.nobs

        if isinstance(weights, np.ndarray) and (weights.shape != wrapdata.shape):
            errmsg = f"Weights shape mismatch: {weights.shape} vs {wrapdata.shape}"
            raise ValueError(errmsg)

        return solve(self.matrix, self.ranges, wrapdata, weights)

    def estimate_model_many(
        self,
        wrapdata: np.ndarray,
        weights: np.ndarray | float | None = None,
        worker_count: int | None = None,
    ) -> tuple[np.ndarray, np.ndarray]:
        """Grid search followed by fmin in parallel."""
        if (worker_count is None) or (worker_count <= 0):
            worker_count = max(1, get_cpu_count() - 1)

        if wrapdata.ndim != 2:
            errmsg = f"Input data must be a 2D array. Got {wrapdata.shape}."
            raise ValueError(errmsg)

        if wrapdata.shape[0] != self.nobs:
            errmsg = f"Input shape mismatch. Got {wrapdata.shape} vs {self.nobs}"
            raise ValueError(errmsg)

        const_weights: bool = True
        if weights is None:
            weights = 1.0 / self.nobs
        elif isinstance(weights, np.ndarray):
            if weights.ndim == 2 and (weights.shape != wrapdata.shape):
                errmsg = (
                    f"Weights shape mismatch. Got {weights.shape} vs {wrapdata.shape}"
                )
                raise ValueError(errmsg)
                const_weights = False
                arr_weights: np.ndarray = weights

            if weights.shape[0] != self.nobs:
                errmsg = f"Weights shape mismatch. Got {weights.shape} vs {self.nobs}"
                raise ValueError(errmsg)

        # Return arrays
        nruns: int = wrapdata.shape[1]
        params: np.ndarray = np.zeros((self.ndim, nruns))
        tcoh: np.ndarray = np.zeros(nruns)

        # Run sequentially when only 1 worker available
        if worker_count == 1:
            for ii in range(nruns):
                wts = weights if const_weights else arr_weights[:, ii]
                res = self.estimate_model(
                    wrapdata[:, ii],
                    wts,
                )
                params[:, ii] = res[0]
                tcoh[ii] = res[1]
        else:
            logger.info(f"Modeling batch of {nruns} with {worker_count} threads")

            def inv_inputs(idxs):
                for ii in idxs:
                    wts = weights if const_weights else arr_weights[:, ii]
                    yield (
                        ii,
                        self.matrix,
                        self.ranges,
                        wrapdata[:, ii],
                        wts,
                    )

            # Create a pool and dispatch
            with get_context("fork").Pool(processes=worker_count) as p:
                mp_tasks = p.imap_unordered(wrap_solve, inv_inputs(range(nruns)))

                # Gather results
                for res in mp_tasks:  # type: ignore[assignment]
                    params[:, res[0]] = res[1]
                    tcoh[res[0]] = res[2]  # type: ignore[misc]

        return params, tcoh

estimate_model(wrapdata, weights=None)

Fit model parameters via grid search followed by Nelder-Mead optimization.

Parameters:

Name Type Description Default
wrapdata ndarray

Real-valued array of wrapped phase gradient

 required
weights ndarray | float | None

Real-valued weights - assumed to be normalized to 1.

 None

Returns:

Name Type Description
params ndarray

1D array of length ndim
coh float

Temporal coherence
Source code in src/spurt/links/_grid_search.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def estimate_model(
    self,
    wrapdata: np.ndarray,
    weights: np.ndarray | float | None = None,
) -> tuple[np.ndarray, float]:
    """Fit model parameters via grid search followed by Nelder-Mead optimization.

    Parameters
    ----------
    wrapdata: np.ndarray
        Real-valued array of wrapped phase gradient
    weights: np.ndarray | None
        Real-valued weights - assumed to be normalized to 1.

    Returns
    -------
    params: np.ndarray
        1D array of length ndim
    coh: float
        Temporal coherence
    """
    if wrapdata.ndim != 1:
        errmsg = f"Input data must be a 1D array. Got {wrapdata.shape}."
        raise ValueError(errmsg)

    if weights is None:
        weights = 1.0 / self.nobs

    if isinstance(weights, np.ndarray) and (weights.shape != wrapdata.shape):
        errmsg = f"Weights shape mismatch: {weights.shape} vs {wrapdata.shape}"
        raise ValueError(errmsg)

    return solve(self.matrix, self.ranges, wrapdata, weights)

estimate_model_many(wrapdata, weights=None, worker_count=None)

Grid search followed by fmin in parallel.

Source code in src/spurt/links/_grid_search.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def estimate_model_many(
    self,
    wrapdata: np.ndarray,
    weights: np.ndarray | float | None = None,
    worker_count: int | None = None,
) -> tuple[np.ndarray, np.ndarray]:
    """Grid search followed by fmin in parallel."""
    if (worker_count is None) or (worker_count <= 0):
        worker_count = max(1, get_cpu_count() - 1)

    if wrapdata.ndim != 2:
        errmsg = f"Input data must be a 2D array. Got {wrapdata.shape}."
        raise ValueError(errmsg)

    if wrapdata.shape[0] != self.nobs:
        errmsg = f"Input shape mismatch. Got {wrapdata.shape} vs {self.nobs}"
        raise ValueError(errmsg)

    const_weights: bool = True
    if weights is None:
        weights = 1.0 / self.nobs
    elif isinstance(weights, np.ndarray):
        if weights.ndim == 2 and (weights.shape != wrapdata.shape):
            errmsg = (
                f"Weights shape mismatch. Got {weights.shape} vs {wrapdata.shape}"
            )
            raise ValueError(errmsg)
            const_weights = False
            arr_weights: np.ndarray = weights

        if weights.shape[0] != self.nobs:
            errmsg = f"Weights shape mismatch. Got {weights.shape} vs {self.nobs}"
            raise ValueError(errmsg)

    # Return arrays
    nruns: int = wrapdata.shape[1]
    params: np.ndarray = np.zeros((self.ndim, nruns))
    tcoh: np.ndarray = np.zeros(nruns)

    # Run sequentially when only 1 worker available
    if worker_count == 1:
        for ii in range(nruns):
            wts = weights if const_weights else arr_weights[:, ii]
            res = self.estimate_model(
                wrapdata[:, ii],
                wts,
            )
            params[:, ii] = res[0]
            tcoh[ii] = res[1]
    else:
        logger.info(f"Modeling batch of {nruns} with {worker_count} threads")

        def inv_inputs(idxs):
            for ii in idxs:
                wts = weights if const_weights else arr_weights[:, ii]
                yield (
                    ii,
                    self.matrix,
                    self.ranges,
                    wrapdata[:, ii],
                    wts,
                )

        # Create a pool and dispatch
        with get_context("fork").Pool(processes=worker_count) as p:
            mp_tasks = p.imap_unordered(wrap_solve, inv_inputs(range(nruns)))

            # Gather results
            for res in mp_tasks:  # type: ignore[assignment]
                params[:, res[0]] = res[1]
                tcoh[res[0]] = res[2]  # type: ignore[misc]

    return params, tcoh

LinkModelInterface

Bases: Protocol

Interface to correcting link gradients.

Such objects should provide access to estimated model parameters and reconstructed model.

Attributes:

Name Type Description
nobs int

Number of observations. Can be epochs or interferograms based on context.
ndim int

Number of model parameters.
Source code in src/spurt/links/_interface.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@runtime_checkable
class LinkModelInterface(Protocol):
    """
    Interface to correcting link gradients.

    Such objects should provide access to estimated model
    parameters and reconstructed model.

    Attributes
    ----------
    nobs: int
        Number of observations. Can be epochs or interferograms based on
        context.
    ndim: int
        Number of model parameters.
    """

    @property
    def nobs(self) -> int:
        """Return number of observations in model."""

    @property
    def ndim(self) -> int:
        """Return number of model parameters."""

    def fwd_model(self, x: np.ndarray) -> np.ndarray:
        """Return model evaluated at x."""

    def estimate_model(
        self, wrapdata: np.ndarray, weights: np.ndarray | float | None = None
    ) -> tuple[np.ndarray, float]:
        """Return estimated model parameters and quality metric.

        Parameters
        ----------
        wrapdata: np.ndarray
            1D complex array or real valued array in radians.
        weights: np.ndarray
            1D array of same length as wrapdata with weights.
            If scalar, all observations are equally weighted.

        Returns
        -------
        params: np.ndarray
              1D array of length ndim with the estimated model parameters.
        quality: float
              Temporal coherence or a similar quality metric indicating model
              fit.
        """

    def estimate_model_many(
        self,
        wrapdata: np.ndarray,
        weights: np.ndarray | float | None = None,
        worker_count: int = 0,
    ) -> tuple[np.ndarray, np.ndarray]:
        """Estimate model for many inputs.

        Parameters
        ----------
        wrapdata: np.ndarray
            2D complex array or real valued array in radians.
        weights: np.ndarray
            1D / 2D array of same length as wrapdata with weights.
            If 1D array, same weights are reused for all inputs.
         worker_count: int
            Number of workers to use if inputs can be handled in parallel. The
            implementation determines default in case a number <=0 is provided.

        Returns
        -------
        params: np.ndarray
            2D array of shape (ninputs, ndim) with the estimated model parameters.
        quality: np.ndarray
            1D array of length ninputs representing temporal coherence or a similar
            quality metric indicating model fit.
        """

ndim property

Return number of model parameters.

nobs property

Return number of observations in model.

estimate_model(wrapdata, weights=None)

Return estimated model parameters and quality metric.

Parameters:

Name Type Description Default
wrapdata ndarray

1D complex array or real valued array in radians.

 required
weights ndarray | float | None

1D array of same length as wrapdata with weights. If scalar, all observations are equally weighted.

 None

Returns:

Name Type Description
params ndarray

1D array of length ndim with the estimated model parameters.
quality float

Temporal coherence or a similar quality metric indicating model fit.
Source code in src/spurt/links/_interface.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def estimate_model(
    self, wrapdata: np.ndarray, weights: np.ndarray | float | None = None
) -> tuple[np.ndarray, float]:
    """Return estimated model parameters and quality metric.

    Parameters
    ----------
    wrapdata: np.ndarray
        1D complex array or real valued array in radians.
    weights: np.ndarray
        1D array of same length as wrapdata with weights.
        If scalar, all observations are equally weighted.

    Returns
    -------
    params: np.ndarray
          1D array of length ndim with the estimated model parameters.
    quality: float
          Temporal coherence or a similar quality metric indicating model
          fit.
    """

estimate_model_many(wrapdata, weights=None, worker_count=0)

Estimate model for many inputs.

Parameters:

Name Type Description Default
wrapdata ndarray

2D complex array or real valued array in radians.

 required
weights ndarray | float | None

1D / 2D array of same length as wrapdata with weights. If 1D array, same weights are reused for all inputs. worker_count: int Number of workers to use if inputs can be handled in parallel. The implementation determines default in case a number <=0 is provided.

 None

Returns:

Name Type Description
params ndarray

2D array of shape (ninputs, ndim) with the estimated model parameters.
quality ndarray

1D array of length ninputs representing temporal coherence or a similar quality metric indicating model fit.
Source code in src/spurt/links/_interface.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def estimate_model_many(
    self,
    wrapdata: np.ndarray,
    weights: np.ndarray | float | None = None,
    worker_count: int = 0,
) -> tuple[np.ndarray, np.ndarray]:
    """Estimate model for many inputs.

    Parameters
    ----------
    wrapdata: np.ndarray
        2D complex array or real valued array in radians.
    weights: np.ndarray
        1D / 2D array of same length as wrapdata with weights.
        If 1D array, same weights are reused for all inputs.
     worker_count: int
        Number of workers to use if inputs can be handled in parallel. The
        implementation determines default in case a number <=0 is provided.

    Returns
    -------
    params: np.ndarray
        2D array of shape (ninputs, ndim) with the estimated model parameters.
    quality: np.ndarray
        1D array of length ninputs representing temporal coherence or a similar
        quality metric indicating model fit.
    """

fwd_model(x)

Return model evaluated at x.

Source code in src/spurt/links/_interface.py 
43
44
def fwd_model(self, x: np.ndarray) -> np.ndarray:
    """Return model evaluated at x."""