Merge branch 'feature/memory-face-detection' into 'master'

Feature/memory face detection

See merge request !145

python/armarx_face_recognition/armarx_face_recognition/datatypes.py

@@ -5,20 +5,21 @@ import typing as ty
 from armarx_memory import client as mem
 from armarx_memory.aron.aron_dataclass import AronDataclass
 from armarx_memory.aron.conversion import to_aron
+from armarx_memory.segments.human.profile import Profile
 
 
 @dc.dataclass
-class Person(AronDataclass):
+class Person:
 
     face_encodings: ty.List[np.ndarray] = dc.field(default_factory=list)
 
     profile_id: ty.Optional[mem.MemoryID] = None
-    profile: ty.Dict = dc.field(default_factory=dict)
+    profile: ty.Optional[Profile] = None
 
     def name(self) -> str:
         try:
-            return self.profile["spokenNames"][0]
-        except KeyError:
+            return self.profile.names.spoken[0]
+        except (KeyError, AttributeError):
             if self.profile_id is not None:
                 return self.profile_id.entity_name
             else:

python/armarx_face_recognition/armarx_face_recognition/face_detection.py

@@ -10,14 +10,14 @@ import cv2
 import numpy as np
 import face_recognition
 
-from armarx import cmake_helper
+from armarx_core import cmake_helper
 from armarx import arviz as viz
 from armarx_memory import client as mem
 
 from armarx.pose_helper import convert_position_to_global
 from armarx import FramedPositionBase
 
-from visionx.image_processor import ImageProcessor
+from armarx_vision.image_processor import ImageProcessor
 from visionx import StereoCalibrationInterfacePrx
 
 from armarx_face_recognition.datatypes import Person, FaceRecognition
@@ -65,22 +65,21 @@ class FaceDetection(ImageProcessor):
             mns: mem.MemoryNameSystem,
             logger=None
     ):
-        human_reader = mns.wait_for_reader(mem.MemoryID("Human"))
-        result = human_reader.query_core_segment("Profile")
+        from armarx_memory.segments.human.profile import ProfileReader
 
-        def gather_persons(id: mem.MemoryID, profile) -> ty.Optional[Person]:
+        profile_reader = ProfileReader.from_mns(mns)
+        id_to_profile_dict = profile_reader.query_latest()
+
+        persons = []
+        for id, profile in id_to_profile_dict.items():
             if logger:
                 logger.info(f"Gathering face images of {id} ...")
 
             face_encodings = []
-            for face_image_path in profile["faceImagePaths"]:
-                package: str = face_image_path["package"]
-                rel_path: str = face_image_path["path"]
-
-                [data_path] = cmake_helper.get_data_path(package)
-                abs_path = os.path.join(data_path, package, rel_path)
+            for face_image_path in profile.face_image_paths:
+                abs_path = face_image_path.get_system_path()
                 if os.path.isfile(abs_path):
-                    name = profile["names"]["spoken"][0] or id.entity_name
+                    name = profile.names.spoken[0] or id.entity_name
                     if logger is not None:
                         logger.info("Loading person '%s'.", name)
                         logger.debug("Creating face encoding for person '%s'.", name)
@@ -90,11 +89,8 @@ class FaceDetection(ImageProcessor):
                     face_encodings.append(encoding)
 
             if face_encodings:
-                return Person(face_encodings=face_encodings, profile_id=id, profile=profile)
-            else:
-                return None
+                persons.append(Person(face_encodings=face_encodings, profile_id=id, profile=profile))
 
-        persons: ty.List[Person] = human_reader.for_each_instance_data(gather_persons, result, discard_none=True)
         return persons
 
     def update_calibration(self):
@@ -122,7 +118,7 @@ class FaceDetection(ImageProcessor):
         # self.logger.info("Found %s faces.", len(face_locations))
 
         detected_persons = self.detect_persons(face_encodings)
-        self.logger.info(f"Found {len(detected_persons)} persons "
+        self.logger.info(f"Detected {len(detected_persons)} persons "
                          f"({', '.join(p.name() for p in detected_persons)}).")
 
         face_recognitions = self.make_face_recognitions(images, face_locations, detected_persons)

scenarios/MemoryFaceRecognition/MemoryFaceRecognition.scx

 <?xml version="1.0" encoding="utf-8"?>
 <scenario name="MemoryFaceRecognition" creation="2022-09-08.13:48:46" globalConfigName="./config/global.cfg" package="VisionX" deploymentType="local" nodeName="NodeMain">
-	<application name="PythonApplicationManager" instance="" package="ArmarXCore" nodeName="" enabled="true" iceAutoRestart="false"/>
+	<application name="PythonApplicationManager" instance="memory_face_recognition" package="ArmarXCore" nodeName="" enabled="true" iceAutoRestart="false"/>
 </scenario>