finish implementation of google_perch tflite stuff

google_perch_tflite/model.py

@@ -46,12 +46,6 @@ class Model(ModelBase):
         self.output_layer_index = output_details[1]["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.
@@ -68,7 +62,7 @@ class Model(ModelBase):
         )
         self.model.allocate_tensors()
 
-        # Make a prediction (Audio only for now)
+        # Make a prediction
         self.model.set_tensor(self.input_layer_index, data)
         self.model.invoke()
 

google_perch_tflite/preprocessor.py

@@ -0,0 +1,62 @@
+import numpy as np 
+import iSparrow.preprocessor_base as ppb
+
+
+class Preprocessor(ppb.PreprocessorBase):
+
+    def __init__(self,
+        sample_rate: int = 32000,
+        sample_secs: float = 5.0,
+        resample_type: str = "kaiser_fast",
+        **kwargs ):
+
+        super().__init__(
+            "google_perch_tflite", 
+            sample_rate=sample_rate,
+            sample_secs=sample_secs,
+            resample_type=resample_type,
+            **kwargs
+        )
+
+    def process_audio_data(self, rawdata: np.array)->np.array: 
+        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.",
+            flush=True
+        )
+
+        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)