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
33 changes: 20 additions & 13 deletions aeon/transformations/collection/dictionary_based/_sfa.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
"""

__maintainer__ = []
__all__ = ["SFA"]
__all__ = ["SFA", "get_chars"]

import math
import os
Expand Down Expand Up @@ -476,10 +476,15 @@ def get_words(self):
-------
Array of words
"""
words = np.squeeze(self.words)
return np.array(
[_get_chars(word, self.word_length, self.alphabet_size) for word in words]
)
if self.save_words:
return np.array(
[
get_chars(word, self.word_length, self.alphabet_size)
for word in self.words
]
)

raise ValueError("save_words must be set to True when initializing SFA.")

def _binning(self, X, y=None):
num_windows_per_inst = math.ceil(self.n_timepoints / self.window_size)
Expand Down Expand Up @@ -1178,16 +1183,18 @@ def _get_test_params(cls, parameter_set="default"):


@njit(cache=True, fastmath=True)
def _get_chars(word, word_length, alphabet_size):
chars = np.zeros(word_length, dtype=np.uint32)
def get_chars(words, word_length, alphabet_size):
chars = np.zeros((len(words), word_length), dtype=np.uint32)
letter_bits = int(np.log2(alphabet_size))
mask = (1 << letter_bits) - 1
for i in range(word_length):
# Extract the last bits
char = word & mask
chars[-i - 1] = char
for j in range(chars.shape[0]):
word = words[j] # local copy
for i in range(word_length):
# Extract the last bits
char = word & mask
chars[j][-i - 1] = char

# Right shift by to move to the next group of bits
word >>= letter_bits
# Right shift by to move to the next group of bits
word >>= letter_bits # only changes local copy

return chars
29 changes: 10 additions & 19 deletions aeon/transformations/collection/dictionary_based/_sfa_fast.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
from sklearn.utils import check_random_state

from aeon.transformations.collection import BaseCollectionTransformer
from aeon.transformations.collection.dictionary_based._sfa import get_chars
from aeon.utils.numba.general import AEON_NUMBA_STD_THRESHOLD
from aeon.utils.validation import check_n_jobs

Expand Down Expand Up @@ -775,10 +776,15 @@ def get_words(self):
-------
Array of words
"""
words = np.squeeze(self.words)
return np.array(
[_get_chars(word, self.word_length, self.letter_bits) for word in words]
)
if self.save_words:
return np.array(
[
get_chars(word, self.word_length, self.alphabet_size)
for word in self.words
]
)

raise ValueError("save_words must be set to True when initializing SFA_FAST.")

def transform_words(self, X):
"""Return the words and dft coefficients generated for each series.
Expand Down Expand Up @@ -865,21 +871,6 @@ def __setstate__(self, state):
self.relevant_features = typed_dict


@njit(cache=True, fastmath=True)
def _get_chars(word, word_length, letter_bits):
chars = np.zeros(word_length, dtype=np.uint32)
for i in range(word_length):
# Extract the last bits
mask = (1 << letter_bits[i]) - 1
char = word & mask
chars[-i - 1] = char

# Right shift by to move to the next group of bits
word >>= letter_bits[i]

return chars


@njit(fastmath=True, cache=True, parallel=True)
def _binning_dft(
X,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from aeon.datasets import load_unit_test
from aeon.transformations.collection.dictionary_based import SFA, SFAFast, SFAWhole
from aeon.transformations.collection.dictionary_based._sfa import get_chars


@pytest.mark.parametrize(
Expand Down Expand Up @@ -286,3 +287,46 @@ def test_sfa_dynamic(alphabet_allocation_method):

# Check if the budget is correctly allocated
assert np.mean(np.log2(p.alphabet_sizes)) <= np.log2(alphabet_size)


def test_sfa_get_words_from_sliding_window():
"""Test get_words with sliding window (multiple windows per case)."""
X = np.random.rand(3, 1, 20)
y = np.array([0, 1, 0])

# Sliding window: window_size < n_timepoints
word_length = 4
window_size = 10
alphabet_size = 8

p = SFA(
word_length=word_length,
alphabet_size=alphabet_size,
window_size=window_size,
save_words=True,
)
p.fit_transform(X, y)

words = p.get_words()

# Should return array of shape (n_cases, n_windows, word_length)
assert len(words) == X.shape[0] # n_cases
assert len(words[0]) == X.shape[-1] - window_size + 1 # sliding windows
assert len(words[0][0]) == word_length


def test_get_chars_does_not_modify_input():
"""Test that get_chars does not modify the input words array."""
words = np.array(
[
np.uint32(0b110101),
np.uint32(0b101011),
np.uint32(0b011110),
],
dtype=np.uint32,
)
words_original = words.copy()

_ = get_chars(words, word_length=3, alphabet_size=4)

np.testing.assert_array_equal(words, words_original)
Loading