Add unittests

README.md

@@ -20,13 +20,13 @@ pip install git+ssh://git@github.com/seetacloud/codewithgpu.git
 
 ## Quick Start
 
-### Image Inference
+### Deploy Image Inference Application
 
-See [Example: Simple image inference](examples/simple_image_inference.py).
+See [Example: Image Inference](examples/image_inference.py).
 
 ### Use Record Dataset To Accelerate Data Loading
 
-See [Example: Use the record dataset](examples/use_record_dataset.py).
+See [Example: Record Dataset](examples/record_dataset.py).
 
 ### Model Benchmarks
 

codewithgpu/data/reader.py

@@ -116,7 +116,7 @@ def __init__(
         self._shuffle = shuffle
         self._initial_fill = initial_fill
         self._seed = seed
-        self._stick_to_stick_to_partition = stick_to_partition
+        self._stick_to_partition = stick_to_partition
         self._first, self._current, self._last = 0, 0, 0
         self._partition_size = 0
         self._dataset_size = 0
@@ -135,9 +135,9 @@ def next_example(self):
         return self._dataset.read()
 
     def reset(self):
-        """Reset the environment of dataset."""
+        """Reset the dataset."""
         # Redirect to the adjacent part if available.
-        if not self._stick_to_stick_to_partition:
+        if not self._stick_to_partition:
             self._partition_id = (self._partition_id + 1) % self._num_partitions
         self._first = self._partition_id * self._partition_size
         self._last = min(self._first + self._partition_size, self._dataset_size)
@@ -147,43 +147,46 @@ def reset(self):
         counter = self._buffer_bounds[-1].end
         self._buffer_bounds.append(self.BufferBound(counter, counter))
 
+    def push_example(self):
+        """Push an example into the output queue."""
+        # Pop the depleted buffer if necessary.
+        if self._buffer_bounds[0].is_depleted:
+            self._buffer_bounds.pop(0)
+        pop_bound = self._buffer_bounds[0]
+        push_bound = self._buffer_bounds[-1]
+        pop_offset = 0
+        if self._shuffle:
+            # Sample a random offset.
+            pop_range = pop_bound.end - pop_bound.start
+            pop_offset = np.random.randint(0, pop_range)
+        # Pop an example from the buffer.
+        i = pop_bound.start % len(self._buffer_seq)
+        j = (pop_bound.start + pop_offset) % len(self._buffer_seq)
+        self._output_queue.put(self._buffer_seq[j])
+        self._buffer_seq[j] = self._buffer_seq[i]
+        # Push an example into the buffer.
+        k = push_bound.end % len(self._buffer_seq)
+        self._buffer_seq[k] = self.next_example()
+        # Increase the buffer boundary.
+        push_bound.end += 1
+        pop_bound.start += 1
+        # Reset the cursor if necessary.
+        if self._current >= self._last:
+            self.reset()
+
     def run(self):
         """Start the process."""
         self._init_dataset()
-        # Persist a loop to read examples.
+        # Persist a loop to push examples.
         while True:
-            # Pop the depleted buffer if necessary.
-            if self._buffer_bounds[0].is_depleted:
-                self._buffer_bounds.pop(0)
-            pop_bound = self._buffer_bounds[0]
-            push_bound = self._buffer_bounds[-1]
-            pop_offset = 0
-            if self._shuffle:
-                # Sample a random offset.
-                pop_range = pop_bound.end - pop_bound.start
-                pop_offset = np.random.randint(0, pop_range)
-            # Pop an example from the buffer.
-            i = pop_bound.start % len(self._buffer_seq)
-            j = (pop_bound.start + pop_offset) % len(self._buffer_seq)
-            self._output_queue.put(self._buffer_seq[j])
-            self._buffer_seq[j] = self._buffer_seq[i]
-            # Push an example into the buffer.
-            k = push_bound.end % len(self._buffer_seq)
-            self._buffer_seq[k] = self.next_example()
-            # Increase the buffer boundary.
-            push_bound.end += 1
-            pop_bound.start += 1
-            # Reset the cursor if necessary.
-            if self._current >= self._last:
-                self.reset()
+            self.push_example()
 
     def _init_dataset(self):
         """Initialize the dataset."""
         np.random.seed(self._seed)
         # Instantiate the dataset here to avoid a fork of process.
-        # Fork will somehow fail if dataset is implemented in C/C++.
         self._dataset = self._dataset_getter(path=self._path)
-        # Determine the part specification.
+        # Compute the partitions.
         self._dataset_size = self._dataset.size
         self._partition_size = (self._dataset_size +
                                 self._num_partitions - 1) // self._num_partitions

codewithgpu/inference/module.py

@@ -48,13 +48,4 @@ def get_results(self, inputs):
             The result of each example in the batch.
 
         """
-
-    def get_time_diffs(self):
-        """Return the time differences.
-
-        Returns
-        -------
-        Dict[str, number]
-            The time differences.
-
-        """
+        return inputs

codewithgpu/utils/unittest_util.py

@@ -0,0 +1,36 @@
+# ------------------------------------------------------------------------
+# Copyright (c) 2022-present, SeetaCloud, Co.,Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ------------------------------------------------------------------------
+"""Unittest utilities."""
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import sys
+import unittest
+
+import argparse
+
+# The global argument parser
+parser = argparse.ArgumentParser(add_help=False)
+
+
+def run_tests(argv=None):
+    """Run tests under the current ``__main__``."""
+    if argv is None:
+        _, remaining = parser.parse_known_args()
+        argv = [sys.argv[0]] + remaining
+    unittest.main(argv=argv)

examples/simple_image_inference.py → examples/image_inference.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ------------------------------------------------------------------------
-"""Simple image inference example."""
+"""Image inference example."""
 
 from __future__ import absolute_import
 from __future__ import division

examples/use_record_dataset.py → examples/record_dataset.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ------------------------------------------------------------------------
-"""Use the record dataset."""
+"""Record dataset example."""
 
 from __future__ import absolute_import
 from __future__ import division

requirements.txt

@@ -1,6 +1,6 @@
 # Python dependencies required for development.
-flask
-gradio
-opencv-python
 numpy
 protobuf
+opencv-python
+flask
+gradio

setup.py

@@ -43,7 +43,7 @@ def parse_args():
     if os.path.exists('.git'):
         try:
             git_version = subprocess.check_output(
-                ['git', 'rev-parse', 'HEAD'], cwd=args.source + './')
+                ['git', 'rev-parse', 'HEAD'], cwd='./')
             args.git_version = git_version.decode('ascii').strip()
         except (OSError, subprocess.CalledProcessError):
             pass

test/codewithgpu/test_data.py

@@ -0,0 +1,82 @@
+# ------------------------------------------------------------------------
+# Copyright (c) 2022-present, SeetaCloud, Co.,Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ------------------------------------------------------------------------
+"""Test data module."""
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import os
+import queue
+import shutil
+import tempfile
+import unittest
+
+import codewithgpu
+from codewithgpu.utils.unittest_util import run_tests
+
+
+class TestRecord(unittest.TestCase):
+    """Test record components."""
+
+    def test_writer_and_reader(self):
+        path = tempfile.gettempdir() + '/test_record'
+        features = {'a': ['float'],
+                    'b': {'bb': ['int']},
+                    'c': [['bytes']],
+                    'd': 'string',
+                    'e': [{'ee': 'int'}]}
+        data = {'a': [1., 2., 3.],
+                'b': {'bb': [4, 5, 6]},
+                'c': [[b'7', b'8', b'9']],
+                'd': 'data',
+                'e': [{'ee': 1}, {'ee': 2}]}
+        if os.path.exists(path):
+            shutil.rmtree(path)
+        os.makedirs(path)
+        with codewithgpu.RecordWriter(path, features, max_examples=1) as writer:
+            for _ in range(5):
+                writer.write(data)
+        try:
+            writer.write(data)
+        except RuntimeError:
+            pass
+        dataset = codewithgpu.RecordDataset(path)
+        self.assertEqual(dataset._features, writer._features)
+        self.assertEqual(dataset.size, 5)
+        self.assertEqual(len(dataset), 5)
+        dataset.seek(0)
+        self.assertEqual(data, dataset.read())
+        dataset.reset()
+        for data in dataset:
+            pass
+        self.assertEqual(data, dataset[0])
+        self.assertEqual(data, dataset[3])
+        self.assertEqual(dataset.tell(), 4)
+        dataset.close()
+        output_queue = queue.Queue(10)
+        for shuffle, initial_fill in [(False, 1), (True, 1), (True, 1024)]:
+            reader = codewithgpu.DatasetReader(
+                path, output_queue, shuffle=shuffle, initial_fill=initial_fill)
+            reader._init_dataset()
+            for _ in range(2):
+                reader.push_example()
+            reader._dataset.close()
+        self.assertEqual(data, output_queue.get())
+
+
+if __name__ == '__main__':
+    run_tests()

test/codewithgpu/test_inference.py

@@ -0,0 +1,47 @@
+# ------------------------------------------------------------------------
+# Copyright (c) 2022-present, SeetaCloud, Co.,Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ------------------------------------------------------------------------
+"""Test inference module."""
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import queue
+import unittest
+
+import codewithgpu
+from codewithgpu.utils.unittest_util import run_tests
+
+
+class TestCommand(unittest.TestCase):
+    """Test command.."""
+
+    def test_inference_command(self):
+        input_queue = queue.Queue(10)
+        output_queue = queue.Queue(10)
+        command = codewithgpu.InferenceCommand(
+            input_queue, output_queue, batch_size=2, batch_timeout=0.01)
+        input_queue.put((0, 'data1'))
+        input_queue.put((-1, None))
+        command.run()
+
+    def test_serving_command(self):
+        command = codewithgpu.ServingCommand()
+        command.run()
+
+
+if __name__ == '__main__':
+    run_tests()