<filename>ethologger/mapping/windows.py
import numpy as np
import warnings

from sklearn.feature_extraction.text import TfidfTransformer
from sklearn.preprocessing import MultiLabelBinarizer
from scipy.signal.windows import get_window
from sklearn.preprocessing import normalize

import ethologger.utils.auxiliary as aux


class BehavioralWindows:
    def __init__(self, params_cfg):
        self.cfg = params_cfg["windows"]
        self.wintype = self.cfg.get("wintype", "boxcar")
        self.winsize = self.cfg.get("winsize", 30)
        self.stepsize = self.cfg.get("stepsize", 60)
        self.method = self.cfg.get("method", "tf-idf")
        self.norm = self.cfg.get("norm", None)

        if self.stepsize > self.winsize:
            warnings.warn("It is better to set stepsize smaller than winsize.")
        if self.method == "tf-idf" and self.wintype != "boxcar":
            raise ValueError(
                "tf-idf representations can only be used with boxcar wintype."
            )
        if self.norm is None and self.method == "tf-idf":
            warnings.warn(
                "A normalization method must be specifed for tf-idf representations.\n\
                Setting self.norm to 'l1'."
            )

    def get_representations(self, BhvSeq):
        if self.method == "tf-idf":
            win_repr = self.get_tfidf_vectors(
                BhvSeq,
                winsize=self.winsize,
                stepsize=self.stepsize,
                norm=self.norm,
            )
        elif self.method == "binary":
            win_repr = self.get_binary_vectors(
                BhvSeq,
                winsize=self.winsize,
                stepsize=self.stepsize,
            )
        elif self.method == "majority":
            win_repr = self.get_majority_labels(
                BhvSeq,
                wintype=self.wintype,
                winsize=self.winsize,
                stepsize=self.stepsize,
            )
        elif self.method == "count" and self.norm is None:
            win_repr = self.get_behavior_counts(
                BhvSeq,
                wintype=self.wintype,
                winsize=self.winsize,
                stepsize=self.stepsize,
            )
        elif self.method == "count" and self.norm in ["l1", "l2"]:
            win_repr = self.get_count_vectors(
                BhvSeq,
                wintype=self.wintype,
                winsize=self.winsize,
                stepsize=self.stepsize,
                norm=self.norm,
            )
        else:
            raise ValueError(
                f"Undefined (method, normalization) pair: ({self.method},{self.norm}) is given."
            )
        return win_repr

    @staticmethod
    def _get_bhv_windows(BhvSeq, winsize, stepsize):
        halfsize = winsize // 2
        Bh_win = [w for w in aux.sliding_window(BhvSeq, n=halfsize, s=stepsize)]
        return Bh_win

    @staticmethod
    def get_behavior_counts(BhvSeq, wintype="boxcar", winsize=60, stepsize=1):
        ## from sklearn.feature_extraction.text import CountVectorizer
        # Using CountVectorizer is another alternative, maybe faster?
        assert isinstance(winsize, int)
        num_bhv = np.max(BhvSeq) + 1

        Bh_win = BehavioralWindows._get_bhv_windows(BhvSeq, winsize, stepsize)
        window_arr = get_window(wintype, winsize)

        Bh_count = np.zeros((len(Bh_win), num_bhv), dtype=np.float)

        for i, bhv in enumerate(Bh_win):
            for j, bhv_j in enumerate(bhv):
                Bh_count[i, bhv_j] += window_arr[j]
        return Bh_count

    @staticmethod
    def get_tfidf_vectors(BhvSeq, winsize=60, stepsize=1, norm="l1"):
        Bh_count = BehavioralWindows.get_behavior_counts(
            BhvSeq, winsize=winsize, stepsize=stepsize
        )
        vectorizer = TfidfTransformer(norm=norm)
        Bh_tfidf = vectorizer.fit_transform(Bh_count).toarray()
        return Bh_tfidf

    @staticmethod
    def get_binary_vectors(BhvSeq, winsize=60, stepsize=1):
        halfsize = winsize // 2
        Bh_win = aux.sliding_window(BhvSeq, n=halfsize, s=stepsize)
        mlb_vectorizer = MultiLabelBinarizer()
        Bh_binary = mlb_vectorizer.fit_transform(Bh_win)
        return Bh_binary

    @staticmethod
    def get_count_vectors(BhvSeq, wintype="boxcar", winsize=60, stepsize=1, norm="l1"):
        Bh_count = BehavioralWindows.get_behavior_counts(
            BhvSeq, wintype=wintype, winsize=winsize, stepsize=stepsize
        )
        Bh_count_n = normalize(Bh_count, norm=norm)
        return Bh_count_n

    @staticmethod
    def _get_majority_labels(BhvSeq, winsize=10, stepsize=5):
        halfsize = winsize // 2
        Bh_win = aux.sliding_window(BhvSeq, n=halfsize, s=stepsize)
        Bh_majority = [aux.most_common(Bh_win[i]) for i in range(len(Bh_win))]
        return Bh_majority

    @staticmethod
    def get_majority_labels(BhvSeq, wintype="boxcar", winsize=60, stepsize=1):
        # Ignore '0' label as much as possible.
        Bh_count = BehavioralWindows.get_behavior_counts(
            BhvSeq, wintype=wintype, winsize=winsize, stepsize=stepsize
        )
        Bh_majority = [
            aux.most_common(Bh_count[i, 1:])
            if sum(Bh_count[i, 1:]) > 0
            else aux.most_common(Bh_count[i, :])
            for i in range(len(Bh_count))
        ]
        return Bh_majority

<filename>ethologger/mapping/windows.py
import numpy as np
import warnings

from sklearn.feature_extraction.text import TfidfTransformer
from sklearn.preprocessing import MultiLabelBinarizer
from scipy.signal.windows import get_window
from sklearn.preprocessing import normalize

import ethologger.utils.auxiliary as aux


class BehavioralWindows:
    def __init__(self, params_cfg):
        self.cfg = params_cfg["windows"]
        self.wintype = self.cfg.get("wintype", "boxcar")
        self.winsize = self.cfg.get("winsize", 30)
        self.stepsize = self.cfg.get("stepsize", 60)
        self.method = self.cfg.get("method", "tf-idf")
        self.norm = self.cfg.get("norm", None)

        if self.stepsize > self.winsize:
            warnings.warn("It is better to set stepsize smaller than winsize.")
        if self.method == "tf-idf" and self.wintype != "boxcar":
            raise ValueError(
                "tf-idf representations can only be used with boxcar wintype."
            )
        if self.norm is None and self.method == "tf-idf":
            warnings.warn(
                "A normalization method must be specifed for tf-idf representations.\n\
                Setting self.norm to 'l1'."
            )

    def get_representations(self, BhvSeq):
        if self.method == "tf-idf":
            win_repr = self.get_tfidf_vectors(
                BhvSeq,
                winsize=self.winsize,
                stepsize=self.stepsize,
                norm=self.norm,
            )
        elif self.method == "binary":
            win_repr = self.get_binary_vectors(
                BhvSeq,
                winsize=self.winsize,
                stepsize=self.stepsize,
            )
        elif self.method == "majority":
            win_repr = self.get_majority_labels(
                BhvSeq,
                wintype=self.wintype,
                winsize=self.winsize,
                stepsize=self.stepsize,
            )
        elif self.method == "count" and self.norm is None:
            win_repr = self.get_behavior_counts(
                BhvSeq,
                wintype=self.wintype,
                winsize=self.winsize,
                stepsize=self.stepsize,
            )
        elif self.method == "count" and self.norm in ["l1", "l2"]:
            win_repr = self.get_count_vectors(
                BhvSeq,
                wintype=self.wintype,
                winsize=self.winsize,
                stepsize=self.stepsize,
                norm=self.norm,
            )
        else:
            raise ValueError(
                f"Undefined (method, normalization) pair: ({self.method},{self.norm}) is given."
            )
        return win_repr

    @staticmethod
    def _get_bhv_windows(BhvSeq, winsize, stepsize):
        halfsize = winsize // 2
        Bh_win = [w for w in aux.sliding_window(BhvSeq, n=halfsize, s=stepsize)]
        return Bh_win

    @staticmethod
    def get_behavior_counts(BhvSeq, wintype="boxcar", winsize=60, stepsize=1):
        ## from sklearn.feature_extraction.text import CountVectorizer
        # Using CountVectorizer is another alternative, maybe faster?
        assert isinstance(winsize, int)
        num_bhv = np.max(BhvSeq) + 1

        Bh_win = BehavioralWindows._get_bhv_windows(BhvSeq, winsize, stepsize)
        window_arr = get_window(wintype, winsize)

        Bh_count = np.zeros((len(Bh_win), num_bhv), dtype=np.float)

        for i, bhv in enumerate(Bh_win):
            for j, bhv_j in enumerate(bhv):
                Bh_count[i, bhv_j] += window_arr[j]
        return Bh_count

    @staticmethod
    def get_tfidf_vectors(BhvSeq, winsize=60, stepsize=1, norm="l1"):
        Bh_count = BehavioralWindows.get_behavior_counts(
            BhvSeq, winsize=winsize, stepsize=stepsize
        )
        vectorizer = TfidfTransformer(norm=norm)
        Bh_tfidf = vectorizer.fit_transform(Bh_count).toarray()
        return Bh_tfidf

    @staticmethod
    def get_binary_vectors(BhvSeq, winsize=60, stepsize=1):
        halfsize = winsize // 2
        Bh_win = aux.sliding_window(BhvSeq, n=halfsize, s=stepsize)
        mlb_vectorizer = MultiLabelBinarizer()
        Bh_binary = mlb_vectorizer.fit_transform(Bh_win)
        return Bh_binary

    @staticmethod
    def get_count_vectors(BhvSeq, wintype="boxcar", winsize=60, stepsize=1, norm="l1"):
        Bh_count = BehavioralWindows.get_behavior_counts(
            BhvSeq, wintype=wintype, winsize=winsize, stepsize=stepsize
        )
        Bh_count_n = normalize(Bh_count, norm=norm)
        return Bh_count_n

    @staticmethod
    def _get_majority_labels(BhvSeq, winsize=10, stepsize=5):
        halfsize = winsize // 2
        Bh_win = aux.sliding_window(BhvSeq, n=halfsize, s=stepsize)
        Bh_majority = [aux.most_common(Bh_win[i]) for i in range(len(Bh_win))]
        return Bh_majority

    @staticmethod
    def get_majority_labels(BhvSeq, wintype="boxcar", winsize=60, stepsize=1):
        # Ignore '0' label as much as possible.
        Bh_count = BehavioralWindows.get_behavior_counts(
            BhvSeq, wintype=wintype, winsize=winsize, stepsize=stepsize
        )
        Bh_majority = [
            aux.most_common(Bh_count[i, 1:])
            if sum(Bh_count[i, 1:]) > 0
            else aux.most_common(Bh_count[i, :])
            for i in range(len(Bh_count))
        ]
        return Bh_majority

## from sklearn.feature_extraction.text import CountVectorizer # Using CountVectorizer is another alternative, maybe faster? # Ignore '0' label as much as possible.