Data

def last_datapoint(self) -> dict[str, Any]:
    """Return last measured data point."""
    return {
        'index': self.index,
        'x': self.x_array[-1],
        'y': self.y_array[-1],
    }

def new_datapoints(self) -> Generator[dict[str, Any]]:
    """Return new data points since last callback."""
    for i in range(self.start, self.index + 1):
        yield {
            'x': self.x_array[i],
            'y': self.y_array[i],
            'index': i,
        }

def last_datapoint(self) -> dict[str, float]:
    """Return last measured data point."""
    ret = {
        array.name: array[self.index]
        for array in self.data.arrays()
        if not array.is_derived
    }
    ret['index'] = self.index
    return ret

def new_datapoints(self) -> Generator[dict[str, float]]:
    """Return new data points since last callback."""
    for i in range(self.start, self.index + 1):
        ret = {array.name: array[i] for array in self.data.values() if not array.is_derived}
        ret['index'] = i
        yield ret

def copy(self) -> DataArray:
    """Return a copy of the array."""
    return DataArray._wrap(self._inner.Clone())

def current(self) -> list[float]:
    """Current in µA."""
    # Work-around for mIDC bug
    if self.type == 'miDC':
        return [implementation(val).Value for val in self._inner]
    return self.to_list()

def current_in_range(self) -> list[float]:
    """Raw current value expressed in the active current range.

    `current` = `current_in_range` * CR, e.g. 0.2 * 100uA = 2.0 uA
    """
    return [implementation(val).ValueInRange for val in self._inner]

def current_range(self) -> list[AllowedCurrentRanges]:
    """Return current range as list of strings."""
    return [cr_enum_to_string(implementation(val).CurrentRange) for val in self._inner]

def current_reading(self) -> list[CurrentReading]:
    """Return as list of potential reading objects."""
    return [CurrentReading._from_psobject(implementation(val)) for val in self._inner]

def max(self) -> float:
    """Return max value."""
    return self._inner.MaxValue

def min(self) -> float:
    """Return min value."""
    return self._inner.MinValue

def normalize(self) -> DataArray:
    """Normalize values in array to the 0 - 1 range.

    Values are scaled with ``(value - min) / (max - min)``,
    where min and max are the min and max values.

    The unit of the returned array is inherited from the source array.

    Returns
    -------
    new_array : DataArray
        Data array with normalized values.
    """
    new_values = PSMath.NormalizeArray(self._inner.GetValues(), self.min(), self.max())
    new_array = PSDataArray(
        self._inner.Description, self._inner.Unit, self._inner.ArrayType
    )
    new_array.AddRange(new_values)
    return type(self)._wrap(new_array)

def reading_status(self) -> list[AllowedReadingStatus]:
    """Return reading status as list of strings."""
    return [str(implementation(val).ReadingStatus) for val in self._inner]  # type:ignore

def savitsky_golay(self, window_size: int = 3) -> DataArray:
    """Smooth the array using a Savitsky-Golay filter with the window size.

    (i.e. window size 2 will filter points based on the values of the next/previous 2 points)

    Parameters
    ----------
    window_size : int
        Size of the window

    Returns
    -------
    new_array : DataArray
        Smoothed DataArray
    """
    new = self.copy()
    success = new._inner.Smooth(window_size, False)
    if not success:
        raise ValueError('Something went wrong.')
    return new

def timing_status(self) -> list[AllowedTimingStatus]:
    """Return timing status as list of strings."""
    return [str(implementation(val).TimingStatus) for val in self._inner]  # type:ignore

def to_dataframe(self) -> pd.DataFrame:
    """Return array as pandas DataFrome.

    Requires pandas to be installed.

    Returns
    -------
    df : pd.DataFrame
        Dataframe with current readings
    """
    import pandas as pd

    return pd.DataFrame(self.to_dict())

def to_dict(self) -> dict[str, list[Any]]:
    """Return array as key/value mapping.

    The mapping can be used to create a pandas or polars dataframe.

    For example:

        array = measurement.dataset['Current']
        df = pd.DataFrame(array.to_dict())

    Returns
    -------
    dict[str, list[float | str]
        Dictionary with current readings
    """
    return {
        'Current': self.current(),
        'CurrentInRange': self.current_in_range(),
        'CR': self.current_range(),
        'TimingStatus': self.timing_status(),
        'ReadingStatus': self.reading_status(),
    }

def to_list(self) -> list[float]:
    """Export data array to list."""
    return list(self._inner.GetValues())

def to_numpy(self) -> np.ndarray:
    """Export data array to numpy."""
    return np.array(self._inner.GetValues())

def update(self, values: Sequence[float]):
    """Update all values in-place.

    Parameters
    ----------
    values : Iterable[float]
        Values to store in the array.

    Raises
    ------
    ValueError:
        If `data` length differs from current length.
    """
    if len(values) != len(self._inner):
        raise ValueError(
            f'Replacement data must have same length ({len(self._inner)}), '
            f'got {len(values)}'
        )

    for val, new in zip(self._inner, values):
        val.Value = new

def clear_peaks(self):
    """Clear peaks stored on object."""
    self._inner.ClearPeaks()

def concat(self, other: Curve) -> Self:
    """Concatenate the x and y arrays of this curve with another curve.

    Parameters
    ----------
    other: Curve
        Curve whose points are appended after this curve's points.

    Returns
    -------
    Curve
        New curve with the concatenated x and y arrays.
    """
    new_curve = PSMath.AppendCurves(self._inner, other._inner)

    return type(self)._wrap(new_curve)

def copy(self) -> Curve:
    """Return a copy of this curve."""
    return Curve._wrap(PSCurve(self._inner, cloneData=True))

def find_peaks(
    self,
    min_peak_width: float = 0.1,
    min_peak_height: float = 0.0,
    peak_shoulders: bool = False,
    merge_overlapping_peaks: bool = True,
) -> list[Peak]:
    """Find peaks in a curve in all directions.

    CV can have 1 or 2 direction changes.

    Parameters
    ----------
    min_peak_width : float
        Minimum width of the peak in V
    min_peak_height : float
        Minimum height of the peak in uA
    peak_shoulders : bool, optional
        Use alternative peak search algorithm optimized for finding peaks on slopes
    merge_overlapping_peaks : bool, optional
        Two or more peaks that overlap will be identified as a single
        base peak and as shoulder peaks on the base peak.

    Returns
    -------
    peak_list : list[Peak]
    """
    pspeaks = self._inner.FindPeaks(
        minPeakWidth=min_peak_width,
        minPeakHeight=min_peak_height,
        peakShoulders=peak_shoulders,
        mergeOverlappingPeaks=merge_overlapping_peaks,
    )

    peaks_list = [Peak._wrap(peak) for peak in pspeaks]

    return peaks_list

def find_peaks_semiderivative(
    self,
    min_peak_height: float = 0.0,
) -> list[Peak]:
    """
    Find peaks in a curve using the semi-derivative algorithm.

    Used for detecting non-overlapping peaks in LSV and CV curves.
    The peaks are also assigned to the curve, updating `Curve.peaks`.
    Existing peaks are overwritten.

    For more info, see this
    [Wikipedia page](https://en.wikipedia.org/wiki/Neopolarogram).

    Parameters
    ----------
    min_peak_height : float
        Minimum height of the peak in uA

    Returns
    -------
    peak_list : list[Peak]
    """
    dct = System.Collections.Generic.Dictionary[PSCurve, System.Double]()
    dct[self._inner] = min_peak_height

    pd = PSAnalysis.SemiDerivativePeakDetection()
    pd.GetNonOverlappingPeaks(dct)

    return self.peaks

def linear_slope(
    self, start: None | int = None, stop: None | int = None
) -> tuple[float, float, float]:
    """Calculate linear line parameters for this curve between two indexes.

    current = a + b * x

    Parameters
    ----------
    start : int, optional
        begin index
    stop : int, optional
        end index

    Returns
    -------
    a : float
    b : float
    coefdet : float
        Coefficient of determination (R2)
    """
    if start and stop:
        return self._inner.LLS(start, stop)
    else:
        return self._inner.LLS()

def metadata(self) -> CurveMetadata:
    """Generate curve metadata as dataclass."""
    return CurveMetadata(
        title=self.title,
        columns=['x', 'y'],
        units=[self.x_unit, self.y_unit],
        labels=[self.x_label, self.y_label],
        id=self.id,
    )

def metadata_json(self) -> bytes:
    """Generate curve metadata as json."""
    return TypeAdapter(CurveMetadata).dump_json(self.metadata())

def plot(
    self,
    ax: None | axes.Axes = None,
    legend: bool = True,
    **plot_kwargs,
) -> figure.Figure | figure.SubFigure:
    """Generate simple plot for this curve using matplotlib.

    Parameters
    ----------
    ax : Optional[axes.Axes]
        Add plot to this ax if specified.
    legend : bool
        If True, add legend.
    plot_kwargs
        These keyword arguments are passed to `ax.plot`.

    Returns
    -------
    fig : fig.Figure
        Matplotlib figure. Use `fig.show()` to render plot.
    """
    import matplotlib.pyplot as plt

    if not ax:
        _, ax = plt.subplots()

    _ = ax.plot(self.x_array, self.y_array, label=self.title, **plot_kwargs)
    _ = ax.set_xlabel(f'{self.x_label} ({self.x_unit})')
    _ = ax.set_ylabel(f'{self.y_label} ({self.y_unit})')

    if peaks := self.peaks:
        x, y = list(zip(*((peak.x, peak.y) for peak in peaks)))
        _ = ax.scatter(x, y, label='Peaks')

    if legend:
        _ = ax.legend()

    return ax.figure

def remove_baseline(
    self,
    max_sweeps: int = 1001,
    window_size: int = 2,
    mode: Literal['moving-average'] = 'moving-average',
) -> tuple[Curve, Curve]:
    """Perform the moving average baseline correction on a curve.

    This method calculates and applies a baseline correction using a specified
    moving average window size and a maximum number of sweeps allowed.

    Parameters
    ----------
    max_sweeps : int, optional
        The maximum number of sweeps allowed during the baseline correction process.
        Defaults to 1001.
    window_size : int, optional
        The size of the moving average window used for calculating the baseline.
        Defaults to 2.
    mode : str, optional
        The method used for baseline correction (e.g., 'moving-average').
        Defaults to 'moving-average'.

    Returns
    -------
    BaselineResult
        A named tuple containing:

        - `corrected` : The baseline-corrected curve.
        - `baseline` : The calculated baseline curve that was subtracted.

    Examples
    --------
    Unpacking as a tuple:
    >>> corrected, baseline = curve.remove_baseline(max_sweeps=500, window_size=2)

    Attribute access:
    >>> result = curve.remove_baseline(max_sweeps=500, window_size=2)
    >>> result.corrected, result.baseline
    """

    assert mode == 'moving-average'

    class BaselineResult(NamedTuple):
        corrected: Curve
        baseline: Curve

    _corrected = PSAnalysis.BaselineCorrection.GetMovingAverageBaselineCorrected(
        self._inner, nWindowSize=window_size, maxNSweeps=max_sweeps, baseline=False
    )
    _corrected.Title = self.title + ' (corrected)'
    _baseline = PSAnalysis.BaselineCorrection.GetMovingAverageBaselineCorrected(
        self._inner, nWindowSize=window_size, maxNSweeps=max_sweeps, baseline=True
    )
    _baseline.Title = self.title + ' (baseline)'

    return BaselineResult(
        corrected=Curve._wrap(_corrected), baseline=Curve._wrap(_baseline)
    )

def savitsky_golay(self, window_size: int = 3):
    """Smooth the .y_array using a Savitsky-Golay filter with the specified window
    size.

    (i.e. window size 2 will filter points based on the values of the next/previous 2 points)

    Parameters
    ----------
    window_size : int
        Size of the window
    """
    self._inner.SavitskyGolay(windowSize=window_size)

def smooth(self, smooth_level: int = 0):
    """Smooth the .y_array using a Savitsky-Golay filter with the specified smooth
    level.

    Parameters
    ----------
    smooth_level : int
        The smooth level to be used. -1 = none, 0 = no smooth (spike rejection only),
        1 = 5 points, 2 = 9 points, 3 = 15 points, 4 = 25 points
    """
    success = self._inner.Smooth(smoothLevel=smooth_level)
    if not success:
        raise ValueError('Something went wrong.')

def copy(self) -> DataArray:
    """Return a copy of the array."""
    return DataArray._wrap(self._inner.Clone())

def max(self) -> float:
    """Return max value."""
    return self._inner.MaxValue

def min(self) -> float:
    """Return min value."""
    return self._inner.MinValue

def normalize(self) -> DataArray:
    """Normalize values in array to the 0 - 1 range.

    Values are scaled with ``(value - min) / (max - min)``,
    where min and max are the min and max values.

    The unit of the returned array is inherited from the source array.

    Returns
    -------
    new_array : DataArray
        Data array with normalized values.
    """
    new_values = PSMath.NormalizeArray(self._inner.GetValues(), self.min(), self.max())
    new_array = PSDataArray(
        self._inner.Description, self._inner.Unit, self._inner.ArrayType
    )
    new_array.AddRange(new_values)
    return type(self)._wrap(new_array)

def savitsky_golay(self, window_size: int = 3) -> DataArray:
    """Smooth the array using a Savitsky-Golay filter with the window size.

    (i.e. window size 2 will filter points based on the values of the next/previous 2 points)

    Parameters
    ----------
    window_size : int
        Size of the window

    Returns
    -------
    new_array : DataArray
        Smoothed DataArray
    """
    new = self.copy()
    success = new._inner.Smooth(window_size, False)
    if not success:
        raise ValueError('Something went wrong.')
    return new

def to_list(self) -> list[float]:
    """Export data array to list."""
    return list(self._inner.GetValues())

def to_numpy(self) -> np.ndarray:
    """Export data array to numpy."""
    return np.array(self._inner.GetValues())

def update(self, values: Sequence[float]):
    """Update all values in-place.

    Parameters
    ----------
    values : Iterable[float]
        Values to store in the array.

    Raises
    ------
    ValueError:
        If `data` length differs from current length.
    """
    if len(values) != len(self._inner):
        raise ValueError(
            f'Replacement data must have same length ({len(self._inner)}), '
            f'got {len(values)}'
        )

    for val, new in zip(self._inner, values):
        val.Value = new

def arrays(
    self,
    type: AllowedArrayTypes | None = None,
    name: str | None = None,
    quantity: str | None = None,
    hidden: bool = False,
) -> Sequence[DataArray]:
    """Return list of all arrays.

    By default, return all arrays.
    Only one filter can be active at the time.

    Parameters
    ----------
    type : str
        Get arrays by type of the array, e.g. 'Current', 'Frequency', 'AuxInput'.
        Use `.array_types()` to get a list of possible values.
    name : str
        Get arrays with given name, e.g. 'scan1', 'time'.
        Use `.array_names()` to get a list of possible values.
    quantity : str
        Get arrays by quantity, e.g. 'Charge', '-Phase'.
        Use `.array_quantities()` to get a list of possible values.
    hidden : bool
        Return hidden arrays used for debugging.

    Returns
    -------
    arrays : Sequence[DataArray]
        List of arrays.
    """
    if type:
        return self._filter(key=lambda array: array.type == type)
    elif name:
        return self._filter(key=lambda array: array.name == name)
    elif quantity:
        return self._filter(key=lambda array: array.quantity == quantity)
    elif hidden:
        return [DataArray._wrap(psarray) for psarray in self._inner if psarray.Hidden]
    else:
        return list(self.values())

def curve(self, x: str, y: str, title: str | None = None) -> Curve:
    """Construct a custom curve from x and y keys.

    Parameters
    ----------
    x : str
        Key identifying the x array
    y : str
        Key identifying the y array
    title : str
        Set the title. If None, use the $x-$y as title

    Returns
    -------
    curve : Curve
        New Curve with plotting x against y
    """
    xarray = self[x]
    yarray = self[y]

    if not title:
        title = f'{x}-{y}'

    return Curve(xarray, yarray, title=title)

def to_dataframe(self) -> pd.DataFrame:
    """Return dataset as pandas DataFrame.
    Requires pandas to be installed.

    Returns
    -------
    df : pd.DataFrame
        Dataframe with all arrays in dataset.
    """
    import pandas as pd

    dct = self.to_dict()
    return pd.DataFrame.from_dict(dct, orient='index').T

def to_dict(self) -> dict[str, list[Any]]:
    """Return dataset as key/value mapping.

    The mapping can be used to create a pandas or polars dataframe.

    For example:

        df = pd.DataFrame(dataset.to_dict())

    Returns
    -------
    dct : dict[str, list[Any]]
        Dictionary with all arrays in dataset.
    """
    dct: dict[str, Any] = {key: arr.to_list() for key, arr in self.items() if len(arr)}

    if 'Current' in self.array_types:
        *_, current = self.arrays(type='Current')

        if isinstance(current, CurrentArray):
            dct['CR'] = current.current_range()
            dct['ReadingStatus'] = current.reading_status()

    return dct