AdjustedMatrix.py

import numpy as np
from obspy import UTCDateTime
from pydantic import BaseModel
from typing import Any, List, Optional

from .. import ChannelConverter
from .. import pydantic_utcdatetime
from .Metric import Metric
from ..residual.Reading import Reading, get_absolutes_xyz, get_ordinates


class AdjustedMatrix(BaseModel):
    """Attributes pertaining to adjusted(affine) matrices, applied by the AdjustedAlgorithm

    Attributes
    ----------
    matrix: affine matrix generated by Affine's calculate method
    pier_correction: pier correction generated by Affine's calculate method
    starttime: beginning of interval that matrix is valid for
    endtime: end of interval that matrix is valid for
    NOTE: valid intervals are only generated when bad data is encountered.
    Matrix is non-constrained otherwise
    """

    matrix: Optional[Any] = None
    pier_correction: Optional[float] = None
    metrics: Optional[List[Metric]] = None
    starttime: Optional[UTCDateTime] = None
    endtime: Optional[UTCDateTime] = None
    time: Optional[UTCDateTime] = None

    def process(self, values: List[List[float]], outchannels=["X", "Y", "Z", "F"]):
        """ Apply matrix to raw data. Apply pier correction to F when necessary """
        data = np.vstack([values[0:3]] + [np.ones_like(values[0])])
        adjusted = self.matrix @ data
        if "F" in outchannels:
            f = values[-1] + self.pier_correction
            adjusted = np.vstack([adjusted[0 : len(outchannels) - 1]] + [f])
        else:
            adjusted = adjusted[0 : len(outchannels)]
        return adjusted

    def get_metrics(self, readings: List[Reading]) -> List[Metric]:
        """Computes mean absolute error and standard deviation for X, Y, Z, and dF between expected and predicted values.

        Attributes
        ----------
        readings: list of Readings
        matrix: composed matrix

        Outputs
        -------
        metrics: list of Metric objects
        """
        absolutes = get_absolutes_xyz(readings=readings)
        ordinates = get_ordinates(readings=readings)
        stacked_ordinates = np.vstack((ordinates, np.ones_like(ordinates[0])))
        predicted = self.matrix @ stacked_ordinates
        metrics = []
        elements = ["X", "Y", "Z", "dF"]
        expected = absolutes + tuple(
            ChannelConverter.get_computed_f_using_squares(*absolutes)
        )
        predicted = predicted[0:3] + tuple(
            ChannelConverter.get_computed_f_using_squares(*predicted[0:3])
        )
        for i in range(len(elements) - 1):
            diff = expected[i] - predicted[i]
            metrics.append(
                Metric(
                    element=elements[i],
                    absmean=abs(np.nanmean(diff)),
                    stddev=np.std(diff),
                )
            )
        return metrics