Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
15 changes: 8 additions & 7 deletions src/ruptures/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
"""

import abc
from typing_extensions import Self
from ruptures.utils import pairwise


Expand All @@ -18,17 +19,17 @@ class BaseEstimator(metaclass=abc.ABCMeta):
"""

@abc.abstractmethod
def fit(self, *args, **kwargs):
def fit(self, *args, **kwargs) -> Self:
"""To call the segmentation algorithm."""
pass

@abc.abstractmethod
def predict(self, *args, **kwargs):
def predict(self, *args, **kwargs) -> list[int]:
"""To call the segmentation algorithm."""
pass

@abc.abstractmethod
def fit_predict(self, *args, **kwargs):
def fit_predict(self, *args, **kwargs) -> list[int]:
"""To call the segmentation algorithm."""
pass

Expand All @@ -43,17 +44,17 @@ class BaseCost(object, metaclass=abc.ABCMeta):
"""

@abc.abstractmethod
def fit(self, *args, **kwargs):
def fit(self, *args, **kwargs) -> Self:
"""Set the parameters of the cost function, for instance the Gram
matrix, etc."""
pass

@abc.abstractmethod
def error(self, start, end):
def error(self, start: int, end: int) -> float:
"""Returns the cost on segment [start:end]."""
pass

def sum_of_costs(self, bkps):
def sum_of_costs(self, bkps: list[int]) -> float:
"""Returns the sum of segments cost for the given segmentation.

Args:
Expand All @@ -67,5 +68,5 @@ def sum_of_costs(self, bkps):

@property
@abc.abstractmethod
def model(self):
def model(self) -> str:
pass
8 changes: 5 additions & 3 deletions src/ruptures/costs/costautoregressive.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import numpy as np
from numpy.lib.stride_tricks import as_strided
from numpy.linalg import lstsq
from numpy.typing import NDArray
from copy import deepcopy
from typing_extensions import Self

from ruptures.base import BaseCost
from ruptures.costs import NotEnoughPoints
Expand All @@ -12,7 +14,7 @@ class CostAR(BaseCost):

model = "ar"

def __init__(self, order=4):
def __init__(self, order: int = 4) -> None:
"""Initialize the object.

Args:
Expand All @@ -23,7 +25,7 @@ def __init__(self, order=4):
self.min_size = max(5, order + 1)
self.order = order

def fit(self, signal):
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance. The signal must be 1D.

Args:
Expand All @@ -49,7 +51,7 @@ def fit(self, signal):
self.signal[: self.order] = self.signal[self.order]
return self

def error(self, start, end):
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
8 changes: 5 additions & 3 deletions src/ruptures/costs/costclinear.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
r"""Continuous linear change."""

import numpy as np
from numpy.typing import NDArray
from typing_extensions import Self

from ruptures.base import BaseCost
from ruptures.costs import NotEnoughPoints
Expand All @@ -11,12 +13,12 @@ class CostCLinear(BaseCost):

model = "clinear"

def __init__(self):
def __init__(self) -> None:
"""Initialize the object."""
self.signal = None
self.min_size = 3

def fit(self, signal) -> "CostCLinear":
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance.

Args:
Expand All @@ -32,7 +34,7 @@ def fit(self, signal) -> "CostCLinear":

return self

def error(self, start, end) -> float:
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
10 changes: 6 additions & 4 deletions src/ruptures/costs/costcosine.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
r"""CostCosine (kernel change point detection with the cosine similarity)"""

from typing_extensions import Self
import numpy as np
from numpy.typing import NDArray
from ruptures.base import BaseCost
from ruptures.costs import NotEnoughPoints
from scipy.spatial.distance import pdist, squareform
Expand All @@ -11,14 +13,14 @@ class CostCosine(BaseCost):

model = "cosine"

def __init__(self):
def __init__(self) -> None:
"""Initialize the object."""
self.signal = None
self.min_size = 1
self._gram = None

@property
def gram(self):
def gram(self) -> NDArray[np.number]:
"""Generate the gram matrix (lazy loading).

Only access this function after a `.fit()` (otherwise
Expand All @@ -28,7 +30,7 @@ def gram(self):
self._gram = squareform(1 - pdist(self.signal, metric="cosine"))
return self._gram

def fit(self, signal) -> "CostCosine":
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance.

Args:
Expand All @@ -43,7 +45,7 @@ def fit(self, signal) -> "CostCosine":
self.signal = signal
return self

def error(self, start, end) -> float:
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
7 changes: 5 additions & 2 deletions src/ruptures/costs/costl1.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
r"""CostL1 (least absolute deviation)"""

from typing_extensions import Self

import numpy as np
from numpy.typing import NDArray

from ruptures.base import BaseCost
from ruptures.costs import NotEnoughPoints
Expand All @@ -16,7 +19,7 @@ def __init__(self) -> None:
self.signal = None
self.min_size = 2

def fit(self, signal) -> "CostL1":
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance.

Args:
Expand All @@ -32,7 +35,7 @@ def fit(self, signal) -> "CostL1":

return self

def error(self, start, end) -> float:
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
11 changes: 8 additions & 3 deletions src/ruptures/costs/costl2.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
r"""CostL2 (least squared deviation)"""

from numpy.typing import NDArray
import numpy as np
from typing_extensions import Self


from ruptures.costs import NotEnoughPoints

from ruptures.base import BaseCost
Expand All @@ -10,12 +15,12 @@ class CostL2(BaseCost):

model = "l2"

def __init__(self):
def __init__(self) -> None:
"""Initialize the object."""
self.signal = None
self.min_size = 1

def fit(self, signal) -> "CostL2":
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance.

Args:
Expand All @@ -31,7 +36,7 @@ def fit(self, signal) -> "CostL2":

return self

def error(self, start, end) -> float:
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
10 changes: 7 additions & 3 deletions src/ruptures/costs/costlinear.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,10 @@
r"""Linear model change."""

from typing_extensions import Self

import numpy as np
from numpy.linalg import lstsq
from numpy.typing import NDArray

from ruptures.base import BaseCost
from ruptures.costs import NotEnoughPoints
Expand All @@ -11,13 +15,13 @@ class CostLinear(BaseCost):

model = "linear"

def __init__(self):
def __init__(self) -> None:
"""Initialize the object."""
self.signal = None
self.covar = None
self.min_size = 2

def fit(self, signal) -> "CostLinear":
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance. The first column contains the
observed variable. The other columns contains the covariates.

Expand All @@ -33,7 +37,7 @@ def fit(self, signal) -> "CostLinear":
self.covar = signal[:, 1:]
return self

def error(self, start, end) -> float:
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
10 changes: 7 additions & 3 deletions src/ruptures/costs/costml.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,11 @@
r"""Change detection with a Mahalanobis-type metric."""

import numpy as np
from numpy.typing import NDArray

from numpy.linalg import inv
from typing import Optional
from typing_extensions import Self

from ruptures.base import BaseCost
from ruptures.exceptions import NotEnoughPoints
Expand All @@ -12,7 +16,7 @@ class CostMl(BaseCost):

model = "mahalanobis"

def __init__(self, metric=None):
def __init__(self, metric: Optional[NDArray[np.number]] = None) -> None:
"""Create a new instance.

Args:
Expand All @@ -25,7 +29,7 @@ def __init__(self, metric=None):
self.gram = None
self.min_size = 2

def fit(self, signal) -> "CostMl":
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance.

Args:
Expand All @@ -47,7 +51,7 @@ def fit(self, signal) -> "CostMl":

return self

def error(self, start, end):
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
9 changes: 6 additions & 3 deletions src/ruptures/costs/costnormal.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
r"""Gaussian process changes (CostNormal)"""

from typing import Optional
from numpy.typing import NDArray
from typing_extensions import Self
import warnings

import numpy as np
Expand All @@ -13,7 +16,7 @@ class CostNormal(BaseCost):

model = "normal"

def __init__(self, add_small_diag=True):
def __init__(self, add_small_diag: Optional[bool] = True) -> None:
"""Initialize the object.

Args:
Expand All @@ -32,7 +35,7 @@ def __init__(self, add_small_diag=True):
UserWarning,
)

def fit(self, signal) -> "CostNormal":
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance.

Args:
Expand All @@ -49,7 +52,7 @@ def fit(self, signal) -> "CostNormal":
self.n_samples, self.n_dims = self.signal.shape
return self

def error(self, start, end) -> float:
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
8 changes: 5 additions & 3 deletions src/ruptures/costs/costrank.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
r"""Rank-based cost function (CostRank)"""

from typing_extensions import Self
import numpy as np
from numpy.linalg import pinv, LinAlgError
from numpy.typing import NDArray
from scipy.stats.mstats import rankdata

from ruptures.base import BaseCost
Expand All @@ -13,13 +15,13 @@ class CostRank(BaseCost):

model = "rank"

def __init__(self):
def __init__(self) -> None:
"""Initialize the object."""
self.inv_cov = None
self.ranks = None
self.min_size = 2

def fit(self, signal) -> "CostRank":
def fit(self, signal: NDArray[np.number]) -> Self:
"""Set parameters of the instance.

Args:
Expand Down Expand Up @@ -55,7 +57,7 @@ def fit(self, signal) -> "CostRank":

return self

def error(self, start, end):
def error(self, start: int, end: int) -> float:
"""Return the approximation cost on the segment [start:end].

Args:
Expand Down
Loading
Loading