fix: test coverage and linting

.theia/settings.json

@@ -8,5 +8,6 @@
     "editor.autoSave": "on",
     "python.linting.enabled": true,
     "python.linting.pylintEnabled": false,
-    "python.linting.pylamaEnabled": true
+    "python.linting.pylamaEnabled": true,
+    "python.pythonPath": "/usr/bin/python"
 }

src/ambianic/logger.py

@@ -3,7 +3,6 @@
 import logging.handlers
 import os
 import pathlib
-import time
 
 DEFAULT_FILE_LOG_LEVEL = logging.INFO
 DEFAULT_CONSOLE_LOG_LEVEL = logging.WARN

src/ambianic/pipeline/ai/fall_detect.py

@@ -1,10 +1,12 @@
 """Fall detection pipe element."""
 from ambianic.pipeline.ai.tf_detect import TFDetectionModel
 from ambianic.pipeline.ai.pose_engine import PoseEngine
+from ambianic import DEFAULT_DATA_DIR
 import logging
 import math
 import time
 from PIL import Image, ImageDraw
+from pathlib import Path
 
 log = logging.getLogger(__name__)
 
@@ -32,6 +34,12 @@ def __init__(self,
                          confidence_threshold=confidence_threshold,
                          **kwargs)
 
+        if self.context:
+            self._sys_data_dir = self.context.data_dir
+        else:
+            self._sys_data_dir = DEFAULT_DATA_DIR
+        self._sys_data_dir = Path(self._sys_data_dir)
+
         # previous pose detection information for frame at time t-1 and t-2 \
         # to compare pose changes against
         self._prev_data = [None] * 2
@@ -56,7 +64,7 @@ def __init__(self,
         # self._prev_data[1] : store data of frame at t-1
         self._prev_data[0] = self._prev_data[1] = _dix
 
-        self._pose_engine = PoseEngine(self._tfengine)
+        self._pose_engine = PoseEngine(self._tfengine, context=self.context)
         self._fall_factor = 60
         self.confidence_threshold = confidence_threshold
         log.debug(f"Initializing FallDetector with conficence threshold: {self.confidence_threshold}")
@@ -191,7 +199,7 @@ def find_keypoints(self, image):
                     tmp_swap = keypoint.yx[0]
                     keypoint.yx[0] = keypoint.yx[1]
                     keypoint.yx[1] = height-tmp_swap
-            # we could not detect a pose with sufficient confidence
+            # we could not detexct a pose with sufficient confidence
             log.info(f"""A pose detected with spinal_vector_score={spinal_vector_score} >= {min_score} confidence threshold.
                 Pose keypoints: {pose_dix}"
                 """
@@ -268,9 +276,11 @@ def draw_lines(self, thumbnail, pose_dix, score):
 
         # save a thumbnail for debugging
         timestr = int(time.monotonic()*1000)
-        path = f'tmp-fall-detect-thumbnail-{timestr}-score-{score}.jpg'
-        thumbnail.save(path, format='JPEG')
-
+        debug_image_file_name = \
+            f'tmp-fall-detect-thumbnail-{timestr}-score-{score}.jpg'
+        thumbnail.save(
+                       Path(self._sys_data_dir, debug_image_file_name),
+                       format='JPEG')
         return body_lines_drawn
 
     def get_line_angles_with_yaxis(self, pose_dix):

src/ambianic/pipeline/ai/pose_engine.py

@@ -1,8 +1,10 @@
 from ambianic.pipeline.ai.tf_detect import TFDetectionModel
+from ambianic import DEFAULT_DATA_DIR
 import logging
 import time
 import numpy as np
-from PIL import Image, ImageDraw
+from PIL import ImageDraw
+from pathlib import Path
 
 log = logging.getLogger(__name__)
 
@@ -52,12 +54,17 @@ def __repr__(self):
 
 class PoseEngine:
     """Engine used for pose tasks."""
-    def __init__(self, tfengine=None):
+    def __init__(self, tfengine=None, context=None):
         """Creates a PoseEngine wrapper around an initialized tfengine.
         """
+        if context:
+            self._sys_data_dir = self.context.data_dir
+        else:
+            self._sys_data_dir = DEFAULT_DATA_DIR
+        self._sys_data_dir = Path(self._sys_data_dir)
         assert tfengine is not None
         self._tfengine = tfengine
-        
+
         self._input_tensor_shape = self.get_input_tensor_shape()
         _, self._tensor_image_height, self._tensor_image_width, self._tensor_image_depth = \
                                                 self.get_input_tensor_shape()
@@ -166,7 +173,7 @@ def detect_poses(self, img):
 
             keypoint = Keypoint(KEYPOINTS[point_i], [x, y], prob)            
             keypoint_dict[KEYPOINTS[point_i]] = keypoint
-        
+
         # overall pose score is calculated as the average of all individual keypoint scores
         pose_score = cnt/keypoint_count
         log.debug(f"Overall pose score (keypoint score average): {pose_score}")
@@ -175,7 +182,11 @@ def detect_poses(self, img):
             # save template_image for debugging
             timestr = int(time.monotonic()*1000)
             log.debug(f"Detected a pose with {cnt} keypoints that score over the minimum confidence threshold of {self.confidence_threshold}.")
-            debug_image_file_name = f'tmp-pose-detect-image-time-{timestr}-keypoints-{cnt}.jpg'
-            template_image.save(debug_image_file_name, format='JPEG')
+            debug_image_file_name = \
+                f'tmp-pose-detect-image-time-{timestr}-keypoints-{cnt}.jpg'
+            template_image.save(
+                                Path(self._sys_data_dir,
+                                     debug_image_file_name),
+                                format='JPEG')
             log.debug(f"Debug image saved: {debug_image_file_name}")
         return poses, thumbnail, pose_score

src/ambianic/pipeline/avsource/picam.py

@@ -47,7 +47,11 @@ def _get_camera(self):
             return picamera_override.PiCamera()
 
     def run(self):
-        with self._get_camera() as camera:
+        cam = self._get_camera()
+        if cam is None:
+            # picam not available
+            return
+        with cam as camera:
 
             if self.has_failure():
                 return None

src/ambianic/server.py

@@ -2,14 +2,14 @@
 import logging
 import logging.handlers
 import os
-import pathlib
 import time
 from watchdog.observers import Observer
 
 from ambianic.pipeline import timeline
 from ambianic.pipeline.interpreter import PipelineServer
 from ambianic.util import ServiceExit
-from ambianic import logger, config, get_config_file, get_secrets_file, load_config
+from ambianic import logger, config, get_config_file, \
+    get_secrets_file, load_config
 from ambianic.webapp.flaskr import FlaskServer
 
 log = logging.getLogger(__name__)

tests/pipeline/ai/test_fall_detect.py

@@ -73,16 +73,14 @@ def test_fall_detection_thumbnail_present():
 
     def sample_callback(image=None, thumbnail=None, inference_result=None, **kwargs):
         nonlocal result
-        result = image is not None and thumbnail is not None and inference_result is not None
+        result = image is not None and thumbnail is not None and \
+            inference_result is not None
 
     fall_detector = FallDetector(**config)
-
     output = _OutPipeElement(sample_callback=sample_callback)
     fall_detector.connect_to_next_element(output)
-
     img_1 = _get_image(file_name='fall_img_1.png')
     fall_detector.receive_next_sample(image=img_1)
-
     assert result is True
 
 

tests/pipeline/ai/test_fall_detect_more.py

@@ -1,38 +1,19 @@
 """Test fall detection pipe element."""
 from ambianic.pipeline.ai.fall_detect import FallDetector
-from ambianic.pipeline.ai.object_detect import ObjectDetector
-from ambianic.pipeline import PipeElement
-import os
-import time
-from PIL import Image
+from ambianic import DEFAULT_DATA_DIR
+from ambianic import logger
+from test_fall_detect import _fall_detect_config, _get_image, _OutPipeElement
+from pathlib import Path
 
 
-def _fall_detect_config():
-
-    _dir = os.path.dirname(os.path.abspath(__file__))
-    _good_tflite_model = os.path.join(
-        _dir,
-        'posenet_mobilenet_v1_100_257x257_multi_kpt_stripped.tflite'
-        )
-    _good_edgetpu_model = os.path.join(
-        _dir,
-        'posenet_mobilenet_v1_075_721_1281_quant_decoder_edgetpu.tflite'
-        )
-    _good_labels = os.path.join(_dir, 'pose_labels.txt')
-    config = {
-        'model': {
-            'tflite': _good_tflite_model,
-            'edgetpu': _good_edgetpu_model,
-            },
-        'labels': _good_labels,
-        'top_k': 3,
-        'confidence_threshold': 0.235,
-    }
-    return config
+assert DEFAULT_DATA_DIR is not None
+_data_dir = Path(DEFAULT_DATA_DIR)
+_data_dir.mkdir(parents=True, exist_ok=True)
 
 
-def test_model_inputs():
+def test_model_inputs():    
     """Verify against known model inputs."""
+    global _data_dir
     config = _fall_detect_config()
     fall_detector = FallDetector(**config)
     tfe = fall_detector._tfengine
@@ -46,6 +27,7 @@ def test_model_inputs():
     colors = tfe.input_details[0]['shape'][3]
     assert colors == 3
 
+
 def test_config_confidence_threshold():
     """Verify against known confidence threshold. Make sure it propagates at all levels."""
     config = _fall_detect_config()
@@ -62,3 +44,45 @@ def test_config_confidence_threshold():
     assert fall_detector.confidence_threshold == config['confidence_threshold']
     assert pe.confidence_threshold == config['confidence_threshold']
     assert tfe.confidence_threshold == config['confidence_threshold']
+
+
+def test_debug_image_save():
+    """In DEBUG mode Fall detection should be saving pose and fall detection
+        images while making a detection decision."""
+
+    log_config = {
+        'level': 'DEBUG'
+    }
+    logger.configure(config=log_config)
+
+    """Expected to receive thumnail in result if image is provided and
+        poses are detected."""
+    config = _fall_detect_config()
+    result = None
+
+    def sample_callback(image=None, thumbnail=None, inference_result=None, **kwargs):
+        nonlocal result
+        result = image is not None and thumbnail is not None and \
+            inference_result is not None
+
+    fall_detector = FallDetector(**config)
+    output = _OutPipeElement(sample_callback=sample_callback)
+    fall_detector.connect_to_next_element(output)
+    img_1 = _get_image(file_name='fall_img_1.png')
+    fall_detector.receive_next_sample(image=img_1)
+    assert result is True
+    # now that we know there was a positive detection
+    # lets check if the the interim debug images were saved as expected
+    pose_img_files = list(_data_dir.glob('tmp-pose-detect-image*.jpg'))
+    assert len(pose_img_files) > 0
+    fall_img_files = list(_data_dir.glob('tmp-fall-detect-thumbnail*.jpg'))
+    assert len(fall_img_files) > 0
+    # cleanup after test
+    all_tmp_files = pose_img_files + fall_img_files
+    for f in all_tmp_files:
+        f.unlink
+    # return logger level to INFO to prevent side effects in other tests
+    log_config = {
+        'level': 'INFO'
+    }
+    logger.configure(config=log_config)

tests/pipeline/avsource/test_avsource_http.py

@@ -278,7 +278,7 @@ def test_exception_on_http_fetch_continuous():
         )
     t.start()
     avsource.stop()
-    avsource._fetch_img_exception_recovery_called.wait(timeout=10)
+    assert avsource._fetch_img_exception_recovery_called.wait(timeout=10)
     t.join(timeout=10)
     assert not t.is_alive()
     assert avsource._run_http_fetch_called

tests/test_logger.py

@@ -2,17 +2,13 @@
 import logging
 import logging.handlers
 import os
-from time import sleep
-import ambianic
-from ambianic.server import AmbianicServer
-from ambianic import server, config, logger, load_config
-import yaml
-import pytest
+from ambianic import config, logger
 
 log = logging.getLogger(__name__)
 
 _dir = os.path.dirname(os.path.abspath(__file__))
 
+
 def setup_module(module):
     """ setup any state specific to the execution of the given module."""
     config.reload()