Cluster estimation integration test #251
Conversation
| import time | ||
| import multiprocessing as mp | ||
| from typing import List | ||
| import numpy as np | ||
| from utilities.workers import queue_proxy_wrapper, worker_controller | ||
| from modules.detection_in_world import DetectionInWorld | ||
| from modules.cluster_estimation.cluster_estimation_worker import cluster_estimation_worker |
There was a problem hiding this comment.
Ensure that your imports follow the styling convention: https://uwarg-docs.atlassian.net/wiki/spaces/CV/pages/2253226033/Python+Style+Convention
| args=( | ||
| 3, | ||
| 0, | ||
| 3, | ||
| 0, |
There was a problem hiding this comment.
Try to store numerical constants like these at the top (ie RANDOM_STATE = 0, see other integration tests as an example).
| worker_process.start() | ||
| time.sleep(1) | ||
|
|
||
| output_results: List[List[DetectionInWorld]] = output_queue.queue.get() |
There was a problem hiding this comment.
We want to check to see if the output result is a list of ObjectInWorld, otherwise the worker wouldnt be doing anything
| input_queue.queue.put(test_data_2) | ||
| time.sleep(1) | ||
|
|
||
| output_results: List[List[DetectionInWorld]] = output_queue.queue.get() |
…G/computer-vision-python into cluster-estimation-integration-test
| output_results: List[List[DetectionInWorld]] = output_queue.queue.get() | ||
|
|
||
| assert output_results is not None | ||
| assert isinstance(output_results, list) | ||
| assert len(output_results) == 1 | ||
| assert all(isinstance(obj, ObjectInWorld) for obj in output_results) |
There was a problem hiding this comment.
Sorry maybe I should've been more specific. On line 121 and 135 where you're using the .get() method to remove a list from the queue, you want the variable to expect a List[ObjectInWorld], not a List[List[DetectionInWorld]]. The get method returns a single element from the queue, and in this case, each element should be a List[ObjectInWorld].
There was a problem hiding this comment.
ah ok mb, simple fix then
| output_results: List[DetectionInWorld] = output_queue.queue.get() | ||
|
|
||
| assert output_results is not None | ||
| assert isinstance(output_results, list) | ||
| assert len(output_results) == 2 | ||
| assert all(isinstance(obj, ObjectInWorld) for obj in output_results) |
There was a problem hiding this comment.
It might be better to create a loop where you extract each list from the queue and check whether they are the types youre expecting. You'd remove the len(output_results) assertion and just wrap the other ones inside the loop, exiting if they fail. I would probably try to do this with test_data_1 as well.
HermanG05
left a comment
HermanG05
left a comment
There was a problem hiding this comment.
Conditionally Approved
| output_results: List[DetectionInWorld] = output_queue.queue.get() | ||
|
|
||
| while not output_queue.queue.empty(): | ||
| output_results = output_queue.queue.get() |
There was a problem hiding this comment.
Move output_results: List[DetectionInWorld] = output_queue.queue.get() to line 30 and remove it from line 27.
No description provided.