diff --git a/src/behaviour_finger_counter.py b/src/behaviour_finger_counter.py new file mode 100644 index 0000000..f5273ea --- /dev/null +++ b/src/behaviour_finger_counter.py @@ -0,0 +1,212 @@ +""" + +Behaviour node. + +Detects the number of fingers being held up (1 to 5) +and prints the result to the terminal in real time. + +Uses MediaPipe HandLandmarker for accurate finger joint detection. + +REQUIRES: Python 3.11 or 3.12, mediapipe installed. + +""" + +import time +import cv2 +import mediapipe as mp +import threading + +import middleware as mw + + +FRAME_W = 640 +FRAME_H = 480 +LOOP_RATE = 10 + + +class BehaviourFingerCounter: + """ + Middleware behaviour that counts raised fingers in real time using MediaPipe. + + Runs MediaPipe HandLandmarker in live stream mode with a background thread + continuously reading camera frames. Counts fingers by comparing landmark + positions and prints count changes to terminal. + + > ## Attributes + + ``node : mw.Node`` : Middleware node used for shutdown and logging. + + ``result : mp.tasks.vision.HandLandmarkerResult or None`` : Most recent hand landmark result from the MediaPipe callback. + + ``lock_result : threading.Lock`` : Lock protecting concurrent access to result. + + ``landmarker : mp.tasks.vision.HandLandmarker`` : MediaPipe hand landmarker running in live stream mode. + + ``stream : cv2.VideoCapture`` : MJPEG camera stream used as the video source. + + ``latest_frame : numpy.ndarray or None`` : Most recent frame captured from the camera stream. + + ``lock_frame : threading.Lock`` : Lock protecting concurrent access to latest_frame. + + ``running : bool`` : Controls the camera reader thread loop. + + > ## Functions + """ + + def __init__(self): + """ + Initialize middleware node, MediaPipe hand landmarker, and camera stream. + """ + self.node = mw.Node("behaviour_finger_counter") + self.lock_frame = threading.Lock() + self.lock_result = threading.Lock() + self.stream = cv2.VideoCapture("http://localhost:8080/stream.mjpg") + self.latest_frame = None + self.result = None + self.running = True + t = threading.Thread(target=self.reader, daemon=True) + t.start() + time.sleep(2) + self.node.loginfo("Camera ready.") + + # output_image and timestamp_ms are required by the callback of MediaPipe. If removed: + # TypeError: update_result() missing 2 required positional arguments: 'output_image' and 'timestamp_ms' + + def update_result(result, output_image, timestamp_ms): + with self.lock_result: + self.result = result + + self.landmarker = mp.tasks.vision.HandLandmarker.create_from_options( + mp.tasks.vision.HandLandmarkerOptions( + base_options=mp.tasks.BaseOptions( + model_asset_path="/home/idmind/elmo-v2/src/hand_landmarker.task" + ), + running_mode=mp.tasks.vision.RunningMode.LIVE_STREAM, + num_hands=1, + min_hand_detection_confidence=0.3, + min_hand_presence_confidence=0.3, + min_tracking_confidence=0.3, + result_callback=update_result, + ) + ) + + def reader(self): + """ + Background thread that continuously reads frames from the camera stream. + + Behavior + -------- + - Reads frames from the MJPEG stream in a tight loop. + - Stores the latest successfully decoded frame in `latest_frame`. + - Runs until `running` is set to False. + """ + while self.running: + ret, frame = self.stream.read() + if ret: + with self.lock_frame: + self.latest_frame = frame + + def finger_counter(self, hand_landmarks): + """ + Count raised fingers from a set of hand landmarks. + + Behavior + -------- + - For the four fingers (index, middle, ring, pinky): a finger is raised + if its tip y-coordinate is above all three joints below it. + - For the thumb: uses x-coordinate comparison relative to palm orientation + to detect whether the tip is extended. + + Parameters + ---------- + hand_landmarks : list + List of 21 normalized hand landmark objects from MediaPipe. + + Returns + ------- + int + Number of raised fingers (0–5). + """ + count = 0 + + # four fingers — tip is higher (smaller y) than all joints below it + for tip_idx in [8, 12, 16, 20]: + tip_y = hand_landmarks[tip_idx].y + dip_y = hand_landmarks[tip_idx - 1].y + pip_y = hand_landmarks[tip_idx - 2].y + mcp_y = hand_landmarks[tip_idx - 3].y + if tip_y < min(dip_y, pip_y, mcp_y): + count += 1 + + # thumb — use x position + tip_x = hand_landmarks[4].x + dip_x = hand_landmarks[3].x + pip_x = hand_landmarks[2].x + mcp_x = hand_landmarks[1].x + palm_x = hand_landmarks[0].x + if mcp_x > palm_x: + if tip_x > max(dip_x, pip_x, mcp_x): + count += 1 + else: + if tip_x < min(dip_x, pip_x, mcp_x): + count += 1 + + return count + + def run(self): + """ + Main behaviour loop. + + Behavior + -------- + - Logs startup. + - Polls at `LOOP_RATE` for new camera frames. + - Sends each frame asynchronously to the MediaPipe landmarker. + - Reads the latest landmark result and counts raised fingers. + - Prints the finger count to the terminal only when it changes. + - Prints "No hand detected." when no hand is visible. + - Stops the camera reader, releases resources, and shuts down the node + in the finally block. + """ + self.node.loginfo("Behaviour started.") + last_count = None + + try: + while not self.node.is_shutdown(): + time.sleep(1.0 / LOOP_RATE) + + with self.lock_frame: + frame = self.latest_frame + if frame is None: + continue + + # send frame to mediapipe + mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=frame) + self.landmarker.detect_async(mp_image, int(time.time() * 1000)) + + # read latest result + with self.lock_result: + result = self.result + + if result is None or not result.hand_landmarks: + if last_count is not None: + self.node.loginfo("No hand detected.") + last_count = None + continue + + count = self.finger_counter(result.hand_landmarks[0]) + + if count != last_count: + self.node.loginfo(f"Fingers lifted: {count}") + last_count = count + + finally: + self.running = False + self.stream.release() + self.landmarker.close() + self.node.shutdown() + + +if __name__ == "__main__": + node = BehaviourFingerCounter() + node.run() diff --git a/src/hand_landmarker.task b/src/hand_landmarker.task new file mode 100644 index 0000000..0d53faf Binary files /dev/null and b/src/hand_landmarker.task differ