Add similarity measures for time-series data

.github/workflows/ci.yml

@@ -11,7 +11,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: DeterminateSystems/nix-installer-action@v10
       - uses: DeterminateSystems/magic-nix-cache-action@v4
-      - run: nix flake check
+      - run: nix flake check --show-trace
   release:
     runs-on: ubuntu-latest
     needs: test
@@ -90,7 +90,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           ref: ${{ needs.release.outputs.git-head }}
-      - uses: actions/configure-pages@v4
+      - uses: actions/configure-pages@v5
       - uses: DeterminateSystems/nix-installer-action@v10
       - uses: DeterminateSystems/magic-nix-cache-action@v4
       - run: nix build .#docs

.pre-commit-config.yaml

@@ -2,11 +2,11 @@ default_language_version:
   python: python3.11
 repos:
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.1.3
+    rev: v0.3.4
     hooks:
       - id: ruff
       - id: ruff-format
   - repo: https://github.com/python-poetry/poetry
-    rev: 1.6.1
+    rev: 1.8.2
     hooks:
       - id: poetry-check

cbrkit/__init__.py

@@ -5,7 +5,6 @@
 
 """
 
-
 from . import helpers, loaders, retrieval, sim, typing
 
 __all__ = [

cbrkit/cli.py

@@ -2,7 +2,6 @@
 .. include:: ../cli.md
 """
 
-
 from pathlib import Path
 
 try:

cbrkit/sim/collections.py

@@ -1,10 +1,12 @@
-from collections.abc import Collection, Set
+from collections.abc import Collection, Sequence, Set
 from typing import Any
 
 from cbrkit.helpers import dist2sim
 from cbrkit.typing import SimPairFunc
 
-__all__ = ["jaccard"]
+Number = float | int
+
+__all__ = ["jaccard", "smith_waterman", "dtw"]
 
 
 def jaccard() -> SimPairFunc[Collection[Any], float]:
@@ -26,3 +28,64 @@ def wrapped_func(x: Collection[Any], y: Collection[Any]) -> float:
         return dist2sim(jaccard_distance(x, y))
 
     return wrapped_func
+
+
+def smith_waterman(
+    match_score: int = 2, mismatch_penalty: int = -1, gap_penalty: int = -1
+) -> SimPairFunc[Sequence[Any], float]:
+    """
+    Performs the Smith-Waterman alignment with configurable scoring parameters. If no element matches it returns 0.0.
+
+    Args:
+        match_score: Score for matching characters. Defaults to 2.
+        mismatch_penalty: Penalty for mismatching characters. Defaults to -1.
+        gap_penalty: Penalty for gaps. Defaults to -1.
+
+    Example:
+        >>> sim = smith_waterman()
+        >>> sim("abcde", "fghe")
+        2
+    """
+    from minineedle import core, smith
+
+    def wrapped_func(x: Sequence[Any], y: Sequence[Any]) -> float:
+        try:
+            alignment = smith.SmithWaterman(x, y)
+            alignment.change_matrix(
+                core.ScoreMatrix(
+                    match=match_score, miss=mismatch_penalty, gap=gap_penalty
+                )
+            )
+            alignment.align()
+
+            return alignment.get_score()
+        except ZeroDivisionError:
+            return 0.0
+
+    return wrapped_func
+
+
+def dtw() -> SimPairFunc[Collection[int], float]:
+    """Dynamic Time Warping similarity function.
+
+    Examples:
+        >>> sim = dtw()
+        >>> sim([1, 2, 3], [1, 2, 3, 4])
+        0.5
+    """
+    import numpy as np
+    from dtaidistance import dtw
+
+    def wrapped_func(
+        x: Collection[Number] | np.ndarray, y: Collection[Number] | np.ndarray
+    ) -> float:
+        if not isinstance(x, np.ndarray):
+            x = np.array(x)
+        if not isinstance(y, np.ndarray):
+            y = np.array(y)
+
+        distance = dtw.distance(x, y)
+
+        return dist2sim(distance)
+
+    return wrapped_func

cbrkit/typing.py

@@ -32,20 +32,17 @@ class FloatProtocol(Protocol):
 class SimMapFunc(Protocol[KeyType, ValueType_contra, SimType_cov]):
     def __call__(
         self, x_map: Mapping[KeyType, ValueType_contra], y: ValueType_contra
-    ) -> SimMap[KeyType, SimType_cov]:
-        ...
+    ) -> SimMap[KeyType, SimType_cov]: ...
 
 
 class SimSeqFunc(Protocol[ValueType_contra, SimType_cov]):
     def __call__(
         self, pairs: Sequence[tuple[ValueType_contra, ValueType_contra]], /
-    ) -> SimSeq[SimType_cov]:
-        ...
+    ) -> SimSeq[SimType_cov]: ...
 
 
 class SimPairFunc(Protocol[ValueType_contra, SimType_cov]):
-    def __call__(self, x: ValueType_contra, y: ValueType_contra, /) -> SimType_cov:
-        ...
+    def __call__(self, x: ValueType_contra, y: ValueType_contra, /) -> SimType_cov: ...
 
 
 AnySimFunc = (
@@ -60,14 +57,12 @@ def __call__(
         self,
         similarities: SimSeqOrMap[KeyType, SimType_contra],
         /,
-    ) -> float:
-        ...
+    ) -> float: ...
 
 
 class PoolingFunc(Protocol):
     def __call__(
         self,
         similarities: SimSeq[float],
         /,
-    ) -> float:
-        ...
+    ) -> float: ...

flake.nix

@@ -47,6 +47,7 @@
             extras = [
               "cli"
               "nlp"
+              "timeseries"
             ];
           };
         in