add model and preprocessor code

birdnet_custom_v2.4/model.py

@@ -0,0 +1,215 @@
+from pathlib import Path
+import numpy as np
+
+try:
+    import tflite_runtime.interpreter as tflite
+except Exception:
+    from tensorflow import lite as tflite
+
+from birdnetlib.analyzer import AnalyzerConfigurationError
+
+from iSparrow.sparrow_model_base import ModelBase
+from iSparrow import utils
+
+
+class Model(ModelBase):
+
+    def _check_classifier_path_integrity(
+        self, classifier_model_path: str, classifier_labels_path: str
+    ):
+        """checks if custom classifier/labels are both given if one is present and the files they point to exist"""
+
+        if (classifier_model_path is not None and classifier_labels_path is None) or (
+            classifier_model_path is None and classifier_labels_path is not None
+        ):
+            raise AnalyzerConfigurationError(
+                "Model and label file paths must be specified to use a custom classifier"
+            )
+
+        if (
+            classifier_model_path is not None
+            and Path(classifier_model_path).exists() is False
+        ):
+            raise AnalyzerConfigurationError(
+                f"Custom classifier model could not be found at the provided path {classifier_model_path}"
+            )
+
+        if (
+            classifier_model_path is not None
+            and Path(classifier_labels_path).exists() is False
+        ):
+            raise AnalyzerConfigurationError(
+                f"Custom classifier labels could not be found at the provided path {classifier_labels_path}"
+            )
+
+    #
+    def __init__(
+        self,
+        default_model_path: str = None,
+        model_path: str = None,
+        sigmoid_sensitivity: float = 1.0,
+        num_threads: int = 1,
+    ):
+
+        self.default_model_path = str(Path(default_model_path) / "model.tflite")
+        self.default_labels_path = str(Path(default_model_path) / "labels.txt")
+
+        classifier_model_path = str(Path(model_path) / "model.tflite")
+        classifier_labels_path = str(Path(model_path) / "labels.txt")
+
+        self.sensitivity = sigmoid_sensitivity
+
+        # check custom classifier paths through function due to higher complexity
+        self._check_classifier_path_integrity(
+            classifier_model_path, classifier_labels_path
+        )
+        # need to call this custom because the super class has no prefix..
+        self.custom_classifier = None
+        self.custom_input_layer_index = None
+        self.custom_output_layer_index = None
+
+        self.input_layer_index = None
+        self.output_layer_index = None
+
+        # use the super class for handling the default models and load the custom ones in this one
+        super().__init__(
+            "birdnet_custom",
+            model_path=classifier_model_path,
+            labels_path=classifier_labels_path,
+            num_threads=num_threads,
+            sensitivity=sigmoid_sensitivity,
+        )
+
+    def load_model(self):
+        """
+        load_model Load the default model for making feature embeddings and the custom classifier for classifying them into species.
+
+        """
+
+        # this overrides the base method because we need to load the default models to provide
+        # the feature embeddings and the custom classifier to apply to them to get the actual
+        # classification
+
+        # load the default model
+        self.model = utils.load_model_from_file_tflite(
+            self.default_model_path, num_threads=self.num_threads
+        )
+        self.model.allocate_tensors()
+
+        # Get input and output tensors.
+        input_details = self.model.get_input_details()
+        output_details = self.model.get_output_details()
+
+        # Get input tensor index
+        self.input_layer_index = input_details[0]["index"]
+
+        # Get  feature embeddings
+        self.output_layer_index = output_details[0]["index"] - 1
+        print("Default classifier loaded")
+
+        # now load the custom classifier
+        self.custom_classifier = tflite.Interpreter(
+            model_path=str(self.model_path), num_threads=self.num_threads
+        )
+        self.custom_classifier.allocate_tensors()
+
+        # Get input and output tensors.
+        custom_input_details = self.custom_classifier.get_input_details()
+        custom_output_details = self.custom_classifier.get_output_details()
+
+        self.custom_input_layer_index = custom_input_details[0]["index"]
+        self.custom_output_layer_index = custom_output_details[0]["index"]
+
+        print("Custom classifier loaded")
+
+    def load_species_list(self):
+        # TODO
+        pass
+
+    def get_embeddings(self, data: np.array) -> np.array:
+        """
+        get_embeddings Extract feature embedding from audio file without immediatelly classifying the species.
+                       These can in a second step be used with a custom classifier to find species not
+                       included in the default training data.
+
+        Args:
+            data (np.array): Preprocessed audio snippet to extract features from
+
+        Returns:
+            np.array: Feature embedding produces by the default birdnet CNN.
+        """
+        print(" get embeddings")
+        self.model.resize_tensor_input(
+            self.input_layer_index, [len(data), *data[0].shape]
+        )
+
+        self.model.allocate_tensors()
+
+        # Extract feature embeddings
+        self.model.set_tensor(self.input_layer_index, np.array(data, dtype="float32"))
+
+        self.model.invoke()
+
+        features = self.model.get_tensor(self.output_layer_index)
+
+        return features
+
+    def predict(self, sample: np.array) -> np.array:
+        """
+        predict Make inference about the bird species for the preprocessed data passed to this function as arguments.
+
+        Args:
+            data (np.array): list of preprocessed data chunks
+        Returns:
+            list: List of (label, inferred_probability)
+        """
+        data = np.array([sample], dtype="float32")
+
+        input_details = self.custom_classifier.get_input_details()
+
+        input_size = input_details[0]["shape"][-1]
+
+        feature_vector = self.get_embeddings(data) if input_size != 144000 else data
+
+        self.custom_classifier.resize_tensor_input(
+            self.custom_input_layer_index,
+            [len(feature_vector), *feature_vector[0].shape],
+        )
+
+        self.custom_classifier.allocate_tensors()
+
+        # Make a prediction
+        self.custom_classifier.set_tensor(
+            self.custom_input_layer_index, np.array(feature_vector, dtype="float32")
+        )
+
+        self.custom_classifier.invoke()
+
+        prediction = self.custom_classifier.get_tensor(self.custom_output_layer_index)
+
+        # map to probabilities
+        confidence = self._sigmoid(np.array(prediction), -self.sensitivity)
+
+        return confidence
+
+    @classmethod
+    def from_cfg(cls, sparrow_dir: str, cfg: dict):
+        """
+        from_cfg Create a new instance from a dictionary containing keyword arguments. Usually loaded from a config file.
+
+        Args:
+            sparrow_dir (str): Installation directory of the Sparrow package
+            cfg (dict): Dictionary containing the keyword arguments
+
+        Returns:
+            Model: New model instance created with the supplied kwargs.
+        """
+
+        # preprocess config because we need two models here
+        cfg["default_model_path"] = str(
+            Path(sparrow_dir) / Path("models") / Path("birdnet_default")
+        )
+        cfg["model_path"] = str(
+            Path(sparrow_dir) / Path("models") / Path(cfg["model_path"])
+        )
+        return cls(**cfg)

birdnet_custom_v2.4/preprocessor.py

@@ -0,0 +1,100 @@
+import numpy as np
+import iSparrow.preprocessor_base as ppb
+
+
+class Preprocessor(ppb.PreprocessorBase):
+    """
+    Preprocessor Preprocess audio data into resampled chunks for analysis.
+
+    """
+
+    def __init__(
+        self,
+        sample_rate: int = 48000,
+        overlap: float = 0.0,
+        sample_secs: int = 3.0,
+        resample_type: str = "kaiser_fast",
+    ):
+        """
+        __init__ Construct a new preprocesssor for custom birdnet classifiers from given parameters, and use defaults for the ones not present.
+
+        Args:
+            sample_rate (int, optional): The sample rate used to resample the read audio file. Defaults to 48000.
+            overlap (float, optional): Overlap between chunks to be analyzed. Defaults to 0.0.
+            sample_secs (int, optional): Length of chunks to be analyzed at once. Defaults to 3.0.
+            resample_type (str, optional): Resampling method used when reading from file. Defaults to "kaiser_fast".
+        """
+
+        super().__init__(
+            "birdnet_custom",
+            sample_rate=sample_rate,
+            overlap=overlap,
+            sample_secs=sample_secs,
+            resample_type=resample_type,
+        )
+
+    def process_audio_data(self, rawdata: np.ndarray) -> list:
+        """
+        process_audio_data Process raw, resampled audio data into chunks that then can be analyzed
+
+        Args:
+            data (np.ndarray): raw, resampled audio data as returned from 'read_audio'
+
+        Returns:
+            list: chunked audio data
+        """
+        print("process audio data custom")
+        seconds = self.sample_secs
+        minlen = 1.5
+
+        self.chunks = []
+
+        for i in range(
+            0, len(rawdata), int((seconds - self.overlap) * self.sample_rate)
+        ):
+
+            split = rawdata[i : (i + int(seconds * self.actual_sampling_rate))]
+
+            # End of signal?
+            if len(split) < int(minlen * self.actual_sampling_rate):
+                break
+
+            # Signal chunk too short? Fill with zeros.
+            if len(split) < int(self.actual_sampling_rate * seconds):
+                temp = np.zeros((int(self.actual_sampling_rate * seconds)))
+                temp[: len(split)] = split
+                split = temp
+
+            self.chunks.append(split)
+
+        print(
+            "process audio data custom: complete, read ",
+            str(len(self.chunks)),
+            "chunks.",
+        )
+
+        return self.chunks
+
+    @classmethod
+    def from_cfg(cls, cfg: dict):
+        """
+        from_cfg Construct a new preprocessor from a given dictionary. This represents typically a config node read from a YAML file.
+
+        Args:
+            cfg (dict): Config node read from a YAML file
+
+        Returns:  new preprocessor instance
+        """
+        allowed = [
+            "sample_rate",
+            "overlap",
+            "sample_secs",
+            "resample_type",
+            "duration",
+            "actual_sampling_rate",
+        ]
+
+        if len([key for key in cfg if key not in allowed]) > 0:
+            raise RuntimeError("Erroneous keyword arguments in preprocessor config")
+
+        return cls(**cfg)

birdnet_default_v2.4/model.py

@@ -0,0 +1,105 @@
+from pathlib import Path
+import numpy as np
+
+from iSparrow.sparrow_model_base import ModelBase
+
+# from iSparrow import utils
+
+
+class Model(ModelBase):
+    """
+    Model Model class representing the the default birdnet model. Does currently not support custom species list or species prediction based on coordinates.
+
+    """
+
+    def __init__(
+        self,
+        model_path: str,
+        num_threads: int = 1,
+        sigmoid_sensitivity: float = 1.0,
+        species_list_file: str = None,
+    ):
+        """
+        __init__ Create a new model instance that uses birdnet-analyzer models for bird species classification
+
+        Args:
+            model_path (str): Path to the location of the model file to be loaded
+            num_threads (int, optional): Number of threads used for inference. Defaults to 1.
+            sigmoid_sensitivity (float, optional): Parameter of the sigmoid function used to compute probabilities. Defaults to 1.0.
+
+        Raises:
+            AnalyzerConfigurationError: The model file 'model.tflite' doesn't exist at the given path.
+            AnalyzerConfigurationError: The labels file 'labels.txt' doesn't exist at the given path.
+        """
+
+        labels_path = str(Path(model_path) / "labels.txt")
+
+        model_path = str(Path(model_path) / "model.tflite")
+
+        # base class loads the model and labels
+        super().__init__(
+            "birdnet_default",
+            model_path,
+            labels_path,
+            num_threads=num_threads,
+            sensitivity=sigmoid_sensitivity,
+        )
+
+        # store input and output index to not have to retrieve them each time an inference is made
+        input_details = self.model.get_input_details()
+
+        output_details = self.model.get_output_details()
+
+        # Get input tensor index
+        self.input_layer_index = input_details[0]["index"]
+
+        # Get classification output or feature embeddings as output, depending on presence fo custom classifier
+        self.output_layer_index = output_details[0]["index"]
+
+    def load_species_list(self):
+        # TODO
+        pass
+
+    def predict(self, sample: np.array) -> np.array:
+        """
+        predict Make inference about the bird species for the preprocessed data passed to this function as arguments.
+
+        Args:
+            data (np.array): list of preprocessed data chunks
+        Returns:
+            list: List of (label, inferred_probability)
+        """
+        data = np.array([sample], dtype="float32")
+
+        self.model.resize_tensor_input(
+            self.input_layer_index, [len(data), *data[0].shape]
+        )
+        self.model.allocate_tensors()
+
+        # Make a prediction (Audio only for now)
+        self.model.set_tensor(self.input_layer_index, np.array(data, dtype="float32"))
+        self.model.invoke()
+
+        prediction = self.model.get_tensor(self.output_layer_index)
+
+        confidence = self._sigmoid(np.array(prediction), sensitivity=-self.sensitivity)
+
+        return confidence
+
+    @classmethod
+    def from_cfg(cls, sparrow_folder: str, cfg: dict):
+        """
+        from_cfg Create a new instance from a dictionary containing keyword arguments. Usually loaded from a config file.
+
+        Args:
+            sparrow_dir (str): Installation directory of the Sparrow package
+            cfg (dict): Dictionary containing the keyword arguments
+
+        Returns:
+            Model: New model instance created with the supplied kwargs.
+        """
+        cfg["model_path"] = str(
+            Path(sparrow_folder) / Path("models") / cfg["model_path"]
+        )
+
+        return cls(**cfg)

birdnet_default_v2.4/preprocessor.py

@@ -0,0 +1,100 @@
+import numpy as np
+import iSparrow.preprocessor_base as ppb
+
+
+class Preprocessor(ppb.PreprocessorBase):
+    """
+    Preprocessor Preprocess audio data into resampled chunks for analysis.
+
+    """
+
+    def __init__(
+        self,
+        sample_rate: int = 48000,
+        overlap: float = 0.0,
+        sample_secs: int = 3.0,
+        resample_type: str = "kaiser_fast",
+    ):
+        """
+        __init__ Construct a new preprocesssor for custom birdnet classifiers from given parameters, and use defaults for the ones not present.
+
+        Args:
+            sample_rate (int, optional): The sample rate used to resample the read audio file. Defaults to 48000.
+            overlap (float, optional): Overlap between chunks to be analyzed. Defaults to 0.0.
+            sample_secs (int, optional): Length of chunks to be analyzed at once. Defaults to 3.0.
+            resample_type (str, optional): Resampling method used when reading from file. Defaults to "kaiser_fast".
+        """
+
+        super().__init__(
+            "birdnet_default",
+            sample_rate=sample_rate,
+            overlap=overlap,
+            sample_secs=sample_secs,
+            resample_type=resample_type,
+        )
+
+    def process_audio_data(self, rawdata: np.ndarray) -> list:
+        """
+        process_audio_data Process raw, resampled audio data into chunks that then can be analyzed
+
+        Args:
+            data (np.ndarray): raw, resampled audio data as returned from 'read_audio'
+
+        Returns:
+            list: chunked audio data
+        """
+        print("process audio data default")
+        seconds = self.sample_secs
+        minlen = 1.5
+
+        self.chunks = []
+
+        for i in range(
+            0, len(rawdata), int((seconds - self.overlap) * self.sample_rate)
+        ):
+
+            split = rawdata[i : (i + int(seconds * self.actual_sampling_rate))]
+
+            # End of signal?
+            if len(split) < int(minlen * self.actual_sampling_rate):
+                break
+
+            # Signal chunk too short? Fill with zeros.
+            if len(split) < int(self.actual_sampling_rate * seconds):
+                temp = np.zeros((int(self.actual_sampling_rate * seconds)))
+                temp[: len(split)] = split
+                split = temp
+
+            self.chunks.append(split)
+
+        print(
+            "process audio data default: complete, read ",
+            str(len(self.chunks)),
+            "chunks.",
+        )
+
+        return self.chunks
+
+    @classmethod
+    def from_cfg(cls, cfg: dict):
+        """
+        from_cfg Construct a new preprocessor from a given dictionary. This represents typically a config node read from a YAML file.
+
+        Args:
+            cfg (dict): Config node read from a YAML file
+
+        Returns:  new preprocessor instance
+        """
+        allowed = [
+            "sample_rate",
+            "overlap",
+            "sample_secs",
+            "resample_type",
+            "duration",
+            "actual_sampling_rate",
+        ]
+
+        if len([key for key in cfg if key not in allowed]) > 0:
+            raise RuntimeError("Erroneous keyword arguments in preprocessor config")
+
+        return cls(**cfg)

google_bird_classification/model.py

@@ -0,0 +1,73 @@
+from pathlib import Path
+import numpy as np
+import tensorflow as tf
+from iSparrow.sparrow_model_base import ModelBase
+import pandas as pd
+
+
+class Model(ModelBase):
+
+    def __init__(self, model_path: str, num_threads: int = 1, species_list_file=None):
+        """
+        __init__ Create a new Model instance using the google perch model.
+
+        Args:
+            model_path (str): Path to the model file to load from disk
+            num_threads (int): The number of threads used for inference. Currently not used for this model.
+        """
+        labels_path = str(Path(model_path) / "labels.txt")
+
+        model_path = str(Path(model_path) / "saved_model.pb")
+
+        self.class_mask = None  # used later
+
+        super().__init__(
+            "google_perch",
+            model_path,
+            labels_path,
+            num_threads=num_threads,
+            # sensitivity kwarg doesn't exist here
+        )  # num_threads doesn't do anything here.
+
+    def predict(self, data: np.array):
+        """
+        predict Make inference about the bird species for the preprocessed data passed to this function as arguments.
+
+        Args:
+            data (np.array): list of preprocessed data chunks
+        Returns:
+            list: List of (label, inferred_probability)
+        """
+
+        results = self.labels.copy()
+
+        # README: this should be parallelized??
+        logits, embeddings = self.model.infer_tf(
+            np.array(
+                [
+                    data,
+                ]
+            )
+        )
+
+        results = tf.nn.softmax(logits).numpy()
+        return results
+
+    @classmethod
+    def from_cfg(cls, sparrow_folder: str, cfg: dict):
+        """
+        from_cfg Create a new instance from a dictionary containing keyword arguments. Usually loaded from a config file.
+
+        Args:
+            sparrow_dir (str): Installation directory of the Sparrow package
+            cfg (dict): Dictionary containing the keyword arguments
+
+        Returns:
+            Model: New model instance created with the supplied kwargs.
+        """
+
+        cfg["model_path"] = str(
+            Path(sparrow_folder) / Path("models") / Path(cfg["model_path"])
+        )
+
+        return cls(**cfg)

google_bird_classification/preprocessor.py

@@ -0,0 +1,88 @@
+
+import numpy as np
+
+from tensorflow.signal import frame as tf_split_signal_into_chunks
+
+from iSparrow import preprocessor_base as ppb
+
+
+# README: work in progress - will be completed in separate issue
+class Preprocessor(ppb.PreprocessorBase):
+    """
+    Preprocessor Preprocess audio data into resampled chunks for analysis.
+
+    """
+
+    def __init__(
+        self,
+        sample_rate: int = 32000,
+        sample_secs: float = 5.0,
+        resample_type: str = "kaiser_fast",
+    ):
+        """
+        __init__ Construct a new preprocesssor for custom birdnet classifiers from given parameters, and use defaults for the ones not present.
+
+        Args:
+            sample_rate (int, optional): The sample rate used to resample the read audio file. Defaults to 48000.
+            sample_secs (float, optional): Length of chunks to be analyzed at once. Defaults to 3.0.
+            resample_type (str, optional): Resampling method used when reading from file. Defaults to "kaiser_fast".
+        """
+        # README: this class does not have an overlap attribute because the model it works with does not want it.
+        super().__init__(
+            "google_perch",
+            sample_rate=sample_rate,
+            sample_secs=sample_secs,
+            resample_type=resample_type,
+        )
+
+    def process_audio_data(self, rawdata: np.array) -> np.array:
+        """
+        process_audio_data Process raw, resampled audio data into chunks that then can be analyzed
+
+        Args:
+            data (np.ndarray): raw, resampled audio data as returned from 'read_audio'
+
+        Returns:
+            list: chunked audio data
+        """
+        print("process audio data custom ")
+
+        self.chunks = []
+
+        # raise when sampling rate is unequal.
+        if self.actual_sampling_rate != self.sample_rate:
+            raise RuntimeError(
+                "Sampling rate is not the desired one. Desired sampling rate: {self.sample_rate}, actual sampling rate: {self.actual_sampling_rate}"
+            )
+
+        frame_length = int(self.sample_secs * self.sample_rate)
+        step_length = int(self.sample_secs - self.overlap) * self.sample_rate
+
+        self.chunks = tf_split_signal_into_chunks(
+            rawdata, frame_length, step_length, pad_end=True
+        ).numpy()
+
+        print(
+            "process audio data google: complete, read ",
+            str(len(self.chunks)),
+            "chunks.",
+        )
+
+        return self.chunks
+
+    @classmethod
+    def from_cfg(cls, cfg: dict):
+
+        # make sure there are no more than the allowed keyword arguments in the cfg
+        allowed = [
+            "sample_rate",
+            "sample_secs",
+            "resample_type",
+            "duration",
+            "actual_sampling_rate",
+        ]
+
+        if len([key for key in cfg if key not in allowed]) > 0:
+            raise RuntimeError("Erroneous keyword arguments in preprocessor config")
+
+        return cls(**cfg)