Spaces:
Running
Running
File size: 4,062 Bytes
c1390d7 baf2993 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
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]}}
]
})
if __name__ == '__main__':
unittest.main()
|