tests in separate process

Author: David Erb

src/chimpflow_lib/chimp_adapter.py

@@ -1,7 +1,4 @@
-import importlib
-import inspect
 import logging
-import os
 import warnings
 from pathlib import Path
 from typing import Dict
@@ -37,7 +34,6 @@ def __init__(self, specification: Dict):
             model_path: filename of the pytorch file
             num_classes: input to chimp detect, normally always 3
         """
-        self.__specification = specification
 
         self.__model_path = require(
             "specification",
@@ -50,7 +46,7 @@ def __init__(self, specification: Dict):
             "num_classes",
         )
 
-    async def process(
+    def detect(
         self, crystal_well_model: CrystalWellModel
     ) -> CrystalWellAutolocationModel:
         """
@@ -66,6 +62,9 @@ async def process(
         # Filename is full path to the input filename.
         filename: Path = Path(crystal_well_model.filename)
 
+        # Make a detector object.
+        # TODO: Arrange ChimpDetector internals so that we only have to load
+        # the torch model once per server, instead of once per detection request.
         detector = ChimpDetector(
             self.__model_path,
             [str(filename)],

src/chimpflow_lib/miners/direct_poll.py

@@ -158,7 +158,7 @@ async def query_and_chimp(self) -> None:
         for well_model in well_models:
             # Do the chimp processing.
             logger.info(f"starting chimp process for {well_model.filename}")
-            autolocation_model = await self.__chimp_adapter.process(well_model)
+            autolocation_model = self.__chimp_adapter.detect(well_model)
 
             # Add to the list for uploading.
             autolocation_models.append(autolocation_model)

tests/test_chimp_adapter.py

@@ -1,4 +1,5 @@
 import logging
+import multiprocessing
 import uuid
 import warnings
 
@@ -40,13 +41,30 @@ async def _main_coroutine(self, constants, output_directory):
         """ """
 
         # Make a specification for the chimp adapter.
-        specification = {
+        self.__specification = {
             "model_path": constants["model_path"],
             "num_classes": 3,
         }
 
-        # Make the adapter object which computes the autolocation information.
-        chimp_adapter = ChimpAdapter(specification)
+        # Do the work in a separate process.
+        # TODO: Figure out how to release resources from torchvision in process.
+        p = multiprocessing.Process(target=self.__process)
+        p.start()
+        p.join()
+
+        # Do the same thing, but in a separate process.
+        p = multiprocessing.Process(target=self.__process)
+        p.start()
+        p.join()
+
+    # ----------------------------------------------------------------------------------------
+    def __process(self):
+        chimp_adapter = ChimpAdapter(self.__specification)
+
+        self.__run(chimp_adapter)
+
+    # ----------------------------------------------------------------------------------------
+    def __run(self, chimp_adapter):
 
         # Make a well model to serve as the input to the chimp adapter process method.
         well_model = CrystalWellModel(
@@ -55,8 +73,8 @@ async def _main_coroutine(self, constants, output_directory):
         )
 
         # Process the well image and get the resulting autolocation information.
-        well_model_autolocation: CrystalWellAutolocationModel = (
-            await chimp_adapter.process(well_model)
+        well_model_autolocation: CrystalWellAutolocationModel = chimp_adapter.detect(
+            well_model
         )
 
         assert well_model_autolocation.drop_detected

tests/test_miner.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import multiprocessing
 import os
 import time
 
@@ -34,6 +35,19 @@ def test(self, constants, logging_setup, output_directory):
         # Configuration file to use.
         configuration_file = "tests/configurations/direct_poll.yaml"
 
+        # Do the work in a separate process so that the Service test
+        # can also be run in the same pytest invocation.
+        # TODO: Figure out how to release resources from torchvision in process.
+        p = multiprocessing.Process(
+            target=self.__process,
+            args=[constants, configuration_file, output_directory],
+        )
+        p.start()
+        p.join()
+
+    # ----------------------------------------------------------------------------------------
+    def __process(self, constants, configuration_file, output_directory):
+
         MinerTester().main(constants, configuration_file, output_directory)