test: update tests

aip_trainer/lambdas/lambdaTTS.py

@@ -5,7 +5,7 @@ from pathlib import Path
 from aip_trainer import app_logger
 
 
-def get_tts(text: str, language: str):
+def get_tts(text: str, language: str, tmp_prefix="audio_", tmp_suffix=".wav") -> str:
     from aip_trainer.models import models
 
     if text is None or len(text) == 0:
@@ -24,7 +24,7 @@ def get_tts(text: str, language: str):
         )
     app_logger.info(f"model speaker #0: {speaker} ...")
 
-    with tempfile.NamedTemporaryFile(prefix="audio_", suffix=".wav", delete=False) as tmp_audio_file:
+    with tempfile.NamedTemporaryFile(prefix=tmp_prefix, suffix=tmp_suffix, delete=False) as tmp_audio_file:
         app_logger.info(f"tmp_audio_file output: {tmp_audio_file.name} ...")
         audio_paths = model.save_wav(text=text, speaker=speaker, sample_rate=sample_rate, audio_path=str(tmp_audio_file.name))
         app_logger.info(f"audio_paths output: {audio_paths} ...")

aip_trainer/utils/serialize.py

@@ -0,0 +1,83 @@
+"""Serialize objects"""
+from typing import Mapping
+
+from aip_trainer import app_logger
+
+
+def serialize(obj: any, include_none: bool = False):
+    """
+    Return the input object into a serializable one
+
+    Args:
+        obj: Object to serialize
+        include_none: bool to indicate if include also keys with None values during dict serialization
+
+    Returns:
+        serialized object
+    """
+    return _serialize(obj, include_none)
+
+
+def _serialize(obj: any, include_none: bool):
+    from numpy import ndarray as np_ndarray, floating as np_floating, integer as np_integer
+
+    primitive = (int, float, str, bool)
+    # print(type(obj))
+    try:
+        if obj is None:
+            return None
+        elif isinstance(obj, np_integer):
+            return int(obj)
+        elif isinstance(obj, np_floating):
+            return float(obj)
+        elif isinstance(obj, np_ndarray):
+            return obj.tolist()
+        elif isinstance(obj, primitive):
+            return obj
+        elif type(obj) is list:
+            return _serialize_list(obj, include_none)
+        elif type(obj) is tuple:
+            return list(obj)
+        elif type(obj) is bytes:
+            return _serialize_bytes(obj)
+        elif isinstance(obj, Exception):
+            return _serialize_exception(obj)
+        # elif isinstance(obj, object):
+        #     return _serialize_object(obj, include_none)
+        else:
+            return _serialize_object(obj, include_none)
+    except Exception as e_serialize:
+        app_logger.error(f"e_serialize::{e_serialize}, type_obj:{type(obj)}, obj:{obj}.")
+        return f"object_name:{str(obj)}__object_type_str:{str(type(obj))}."
+
+
+def _serialize_object(obj: Mapping[any, object], include_none: bool) -> dict[any]:
+    from bson import ObjectId
+
+    res = {}
+    if type(obj) is not dict:
+        keys = [i for i in obj.__dict__.keys() if (getattr(obj, i) is not None) or include_none]
+    else:
+        keys = [i for i in obj.keys() if (obj[i] is not None) or include_none]
+    for key in keys:
+        if type(obj) is not dict:
+            res[key] = _serialize(getattr(obj, key), include_none)
+        elif isinstance(obj[key], ObjectId):
+            continue
+        else:
+            res[key] = _serialize(obj[key], include_none)
+    return res
+
+
+def _serialize_list(ls: list, include_none: bool) -> list:
+    return [_serialize(elem, include_none) for elem in ls]
+
+
+def _serialize_bytes(b: bytes) -> dict[str, str]:
+    import base64
+    encoded = base64.b64encode(b)
+    return {"value": encoded.decode('ascii'), "type": "bytes"}
+
+
+def _serialize_exception(e: Exception) -> dict[str, str]:
+    return {"msg": str(e), "type": str(type(e)), **e.__dict__}

aip_trainer/utils/utilities.py

@@ -0,0 +1,57 @@
+"""Various utilities (logger, time benchmark, args dump, numerical and stats info)"""
+
+from copy import deepcopy
+from aip_trainer import app_logger
+from aip_trainer.utils.serialize import serialize
+
+
+def hash_calculate(arr_or_path, is_file: bool, read_mode: str = "rb") -> str | bytes:
+    """
+    Return computed hash from input variable (typically a numpy array).
+
+    Args:
+        arr: input variable
+
+    Returns:
+        computed hash from input variable
+    """
+    from hashlib import sha256
+    from base64 import b64encode
+    from numpy import ndarray as np_ndarray
+
+    if is_file:
+        with open(arr_or_path, read_mode) as file_to_check:
+            # read contents of the file
+            arr_or_path = file_to_check.read()
+            # # pipe contents of the file through
+            # try:
+            #     return hashlib.sha256(data).hexdigest()
+            # except TypeError:
+            #     app_logger.warning(
+            #         f"TypeError, re-try encoding arg:{arr_or_path},type:{type(arr_or_path)}."
+            #     )
+            #     return hashlib.sha256(data.encode("utf-8")).hexdigest()
+
+    if isinstance(arr_or_path, np_ndarray):
+        hash_fn = sha256(arr_or_path.data)
+    elif isinstance(arr_or_path, dict):
+        import json
+
+        serialized = serialize(arr_or_path)
+        variable_to_hash = json.dumps(serialized, sort_keys=True).encode("utf-8")
+        hash_fn = sha256(variable_to_hash)
+    elif isinstance(arr_or_path, str):
+        try:
+            hash_fn = sha256(arr_or_path)
+        except TypeError:
+            app_logger.warning(
+                f"TypeError, re-try encoding arg:{arr_or_path},type:{type(arr_or_path)}."
+            )
+            hash_fn = sha256(arr_or_path.encode("utf-8"))
+    elif isinstance(arr_or_path, bytes):
+        hash_fn = sha256(arr_or_path)
+    else:
+        raise ValueError(
+            f"variable 'arr':{arr_or_path} of type '{type(arr_or_path)}' not yet handled."
+        )
+    return b64encode(hash_fn.digest())

tests/test_lambdaTTS.py

@@ -0,0 +1,63 @@
+import os
+import unittest
+
+from aip_trainer import app_logger
+
+
+def exec_test_lambda_tts(text, language, expected_hash):
+    import random
+    from aip_trainer.lambdas import lambdaTTS
+    from aip_trainer.utils import utilities
+
+    tmp_rnd = str(random.random())
+    tmp_prefix = f"test_lambdaTTS_{language}_ok_{tmp_rnd}_"
+    tmp_suffix = ".wav"
+    output = lambdaTTS.get_tts(
+        text, language, tmp_prefix=tmp_prefix, tmp_suffix=tmp_suffix
+    )
+    assert tmp_prefix in output
+    assert tmp_suffix in output
+    assert os.path.exists(output) and os.path.isfile(output)
+    output_hash = utilities.hash_calculate(output, is_file=True, read_mode="rb")
+    app_logger.info(f"output_hash '{text}', '{language}' => {output_hash}")
+    assert expected_hash == output_hash
+    os.unlink(output)
+
+
+def assert_raises_get_tts(
+    self, real_text, language, exc, error_message
+):
+    from aip_trainer.lambdas import lambdaTTS
+
+    with self.assertRaises(exc):
+        try:
+            lambdaTTS.get_tts(real_text, language)
+        except exc as e:
+            self.assertEqual(str(e), error_message)
+            raise e
+
+
+class TestLambdaTTS(unittest.TestCase):
+    def test_lambdaTTS_en_ok(self):
+        exec_test_lambda_tts(
+            "Hi there, how are you?",
+            "en",
+            b'vf1QWORrGWlvCEQuvI0fajGG7iz2Zqp83p8dVH8pZtY='
+        )
+
+    def test_lambdaTTS_de_ok(self):
+        exec_test_lambda_tts(
+            "Ich bin Alex!",
+            "de",
+            b'jkvM+0Whlb1nwf9eiuDyoQZ1ekzb46k3DcW2glYh0YY='
+        )
+
+    def test_lambdaTTS_empty_text(self):
+        assert_raises_get_tts(self, "", "fake language", ValueError, "cannot read an empty/None text: ''...")
+
+    def test_lambdaTTS_empty_language(self):
+        assert_raises_get_tts(self, "fake text", "", NotImplementedError, "Not tested/supported with '' language...")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_serialize.py

@@ -0,0 +1,97 @@
+import unittest
+
+import numpy as np
+
+from aip_trainer.utils.serialize import serialize
+
+
+test_dict_list_dict = {
+    "type": "FeatureCollection",
+    "name": "volcanoes",
+    "crs": {"type": "name", "properties": {"name": "urn:ogc:def:crs:OGC:1.3:CRS84"}},
+    "features": [
+        {"type": "Feature", "properties": {"Volcano_Number": 283010, "Volcano_Name": "Izu-Tobu", "prop_none": None},
+         "geometry": {"type": "Point", "coordinates": [139.098, 34.9]}},
+        {"type": "Feature",
+         "properties": {"Volcano_Number": 283020, "Volcano_Name": "Hakoneyama", "ndarray": np.array([1])},
+         "geometry": {"type": "Point", "coordinates": [139.021, 35.233]}}
+    ]
+}
+
+
+class TestSerialize(unittest.TestCase):
+    def test_serialize(self):
+        from bson import ObjectId
+
+        # remove keys with values as bson.ObjectId
+        d1 = {"_id": ObjectId()}
+        self.assertDictEqual(serialize(d1), dict())
+
+        # test: serialize nd.float*, number as key => str
+        np_int_4 = np.asarray([87], dtype=np.integer)[0]
+        d2 = {"b": np.float32(45.0), 3: 33, 1.56: np_int_4, 3.5: 44.0, "d": "b", "tuple": (1, 2)}
+        expected_d2 = {
+            'b': 45.0,
+            3: 33,
+            1.56: 87,
+            3.5: 44.0,
+            'd': 'b',
+            "tuple": [1, 2]
+        }
+        serialized_d2 = serialize(d2)
+        self.assertDictEqual(serialized_d2, expected_d2)
+
+        # # nested dict of list of dict, serialize nd.array
+        d3 = {"e": [{"q": 123}, {"q": 456}], "a": np.arange(1.1, 16.88).reshape(4, 4)}
+        expected_d3 = {
+            "e": [{"q": 123}, {"q": 456}],
+            'a': [[1.1, 2.1, 3.1, 4.1], [5.1, 6.1, 7.1, 8.1], [9.1, 10.1, 11.1, 12.1], [13.1, 14.1, 15.1, 16.1]]
+        }
+        self.assertDictEqual(serialize(d3), expected_d3)
+
+    def test_serialize_dict_exception(self):
+        from json import JSONDecodeError
+
+        e = JSONDecodeError(msg="x", doc="what we are?", pos=111)
+        exception = serialize({"k": e})
+        self.assertDictEqual(
+            exception,
+            {'k': {'msg': 'x', 'type': "<class 'json.decoder.JSONDecodeError'>", 'doc': 'what we are?', 'pos': 111,
+                   'lineno': 1, 'colno': 112}}
+        )
+
+    def test_serialize_bytes(self):
+        self.assertDictEqual(
+            serialize({"k": b"x"}),
+            {'k': {'value': 'eA==', 'type': 'bytes'}}
+        )
+
+    def test_serialize_dict_list_dict(self):
+        serialized_dict_no_none = serialize(test_dict_list_dict, include_none=False)
+        self.assertDictEqual(serialized_dict_no_none, {
+            'type': 'FeatureCollection',
+            'name': 'volcanoes',
+            'crs': {'type': 'name', 'properties': {'name': 'urn:ogc:def:crs:OGC:1.3:CRS84'}},
+            'features': [
+                {'type': 'Feature', 'properties': {'Volcano_Number': 283010, 'Volcano_Name': 'Izu-Tobu'},
+                 'geometry': {'type': 'Point', 'coordinates': [139.098, 34.9]}},
+                {'type': 'Feature',
+                 'properties': {'Volcano_Number': 283020, 'Volcano_Name': 'Hakoneyama', 'ndarray': [1]},
+                 'geometry': {'type': 'Point', 'coordinates': [139.021, 35.233]}}
+            ]
+        })
+
+        serialized_dict_wiht_none = serialize(test_dict_list_dict, include_none=True)
+        self.assertDictEqual(serialized_dict_wiht_none, {
+            'type': 'FeatureCollection',
+            'name': 'volcanoes',
+            'crs': {'type': 'name', 'properties': {'name': 'urn:ogc:def:crs:OGC:1.3:CRS84'}},
+            'features': [
+                {'type': 'Feature',
+                 'properties': {'Volcano_Number': 283010, 'Volcano_Name': 'Izu-Tobu', 'prop_none': None},
+                 'geometry': {'type': 'Point', 'coordinates': [139.098, 34.9]}},
+                {'type': 'Feature',
+                 'properties': {'Volcano_Number': 283020, 'Volcano_Name': 'Hakoneyama', 'ndarray': [1]},
+                 'geometry': {'type': 'Point', 'coordinates': [139.021, 35.233]}}
+            ]
+        })

tests/test_utilities.py

@@ -0,0 +1,45 @@
+import unittest
+import numpy as np
+
+from tests import EVENTS_FOLDER
+from aip_trainer import app_logger
+
+
+class TestUtilities(unittest.TestCase):
+    def test_hash_calculate_not_file(self):
+        from aip_trainer.utils.utilities import hash_calculate
+
+        size = 5
+        input_arr = np.arange(size**2).reshape((size, size))
+        hash_output = hash_calculate(input_arr, is_file=False)
+        self.assertEqual(hash_output, b'KgoWp86FwhH2tuinWOfsCfn9d+Iw6B10wwqFfdUeLeY=')
+
+        hash_output = hash_calculate({"arr": input_arr}, is_file=False)
+        self.assertEqual(hash_output, b'M/EYsBPRQLVP9T299xHyOrtT7bdCkIDaMmW2hslMays=')
+
+        hash_output = hash_calculate("a test string...", is_file=False)
+        self.assertEqual(hash_output, b'29a8JwujQklQ6MKQhPyix6G1i/7Pp0uUg5wFybKuCC0=')
+
+        hash_output = hash_calculate("123123123", is_file=False)
+        self.assertEqual(hash_output, b'ky88G1YlfOhTmsJp16q0JVDaz4gY0HXwvfGZBWKq4+8=')
+
+        hash_output = hash_calculate(b"a byte test string...", is_file=False)
+        self.assertEqual(hash_output, b'dgSt/jiqLk0HJ09Xqe/BWzMvnYiOqzWlcSCCfN767zA=')
+
+        with self.assertRaises(ValueError):
+            try:
+                hash_calculate(1, is_file=False)
+            except ValueError as ve:
+                self.assertEqual(str(ve), "variable 'arr':1 of type '<class 'int'>' not yet handled.")
+                raise ve
+
+    def test_hash_calculate_is_file(self):
+        from aip_trainer.utils.utilities import hash_calculate
+
+        output_hash = hash_calculate(EVENTS_FOLDER / "test_en.wav", is_file=True, read_mode="rb")
+        app_logger.info(f"output_hash test_en: {output_hash}")
+        assert b'Dsvmm+mj/opHnmKLT7wIqyhqMLeIuVP4hTWi+DAXS8Y=' == output_hash
+
+        output_hash = hash_calculate(EVENTS_FOLDER / "GetAccuracyFromRecordedAudio.json", is_file=True)
+        app_logger.info(f"output_hash json: {output_hash}")
+        assert b'i83jKpwzfcPitZsrHsnhyFt8xbc+DStpns9rb3vfigw=' == output_hash