diff --git a/edge/adapters/picamera_source.py b/edge/adapters/picamera_source.py
index 2fa7c8e..c9ae8c5 100644
--- a/edge/adapters/picamera_source.py
+++ b/edge/adapters/picamera_source.py
@@ -1,10 +1,29 @@
-"""CSI 카메라 소스 (실장비). 카메라당 별 프로세스로 picamera2 구동."""
+"""CSI 카메라 소스 (실장비 Pi5, picamera2). Implements SourcePort.
+
+picamera2 is a SYSTEM package on Raspberry Pi OS (apt: python3-picamera2 + libcamera),
+NOT a pip wheel — the venv sees it via --system-site-packages (see deploy/README).
+
+COLOR NOTE: configured for "RGB888", which picamera2 delivers as a 3-channel array in
+**BGR** byte order (documented picamera2 quirk). The downstream pose adapters
+(MoveNet/Hailo) expect BGR (they do cv2.cvtColor(BGR2RGB)), same as cv2.VideoCapture,
+so read() returns the array as-is. Validate the channel order on device during bringup
+(a BGR<->RGB swap yields plausible-but-wrong keypoints).
+"""
 
 
 class PiCameraSource:
-    def __init__(self, index: int = 0):
+    def __init__(self, index: int = 0, size=(640, 480), picam2=None):
         self.index = index
-        raise NotImplementedError("실장비 전용: picamera2 배선 필요")
+        if picam2 is None:  # pragma: no cover (device-only: needs picamera2 + CSI camera)
+            from picamera2 import Picamera2  # system apt package (python3-picamera2)
+            picam2 = Picamera2(camera_num=index)
+            picam2.configure(picam2.create_video_configuration(
+                main={"size": tuple(size), "format": "RGB888"}))
+            picam2.start()
+        self.picam2 = picam2
 
     def read(self):
-        raise NotImplementedError
+        frame = self.picam2.capture_array()
+        if frame is None:
+            return False, None
+        return True, frame
diff --git a/edge/daemon.py b/edge/daemon.py
index 226b027..256c176 100644
--- a/edge/daemon.py
+++ b/edge/daemon.py
@@ -40,7 +40,6 @@ def main() -> None:  # pragma: no cover (integration entrypoint, exercised on de
     from edge.alert import PostClient
     from edge.queue import OfflineQueue
     from edge.runtime import PipelineRuntime
-    from edge.adapters.uvc_source import UvcSource
 
     cfg = load_config(dict(os.environ))
     nonce = lambda: os.urandom(8).hex()
@@ -49,7 +48,14 @@ def main() -> None:  # pragma: no cover (integration entrypoint, exercised on de
     client = CloudClient(base, cfg.device_id, cfg.secret, nonce, now)
     post = PostClient(base + "/api/events", cfg.device_id, cfg.secret, nonce, now)
     queue = OfflineQueue(os.environ.get("QUEUE_DB", "queue.db"), post)
-    source = UvcSource(int(os.environ.get("UVC_DEVICE", "0")))
+    # CAMERA_BACKEND selects the capture source: "csi" (Pi CSI cam via picamera2,
+    # uses CSI_DEVICE) or the default "uvc" (USB webcam via OpenCV, uses UVC_DEVICE).
+    if os.environ.get("CAMERA_BACKEND", "uvc").lower() == "csi":
+        from edge.adapters.picamera_source import PiCameraSource
+        source = PiCameraSource(int(os.environ.get("CSI_DEVICE", "0")))
+    else:
+        from edge.adapters.uvc_source import UvcSource
+        source = UvcSource(int(os.environ.get("UVC_DEVICE", "0")))
     # POSE_BACKEND selects the accelerator path: "hailo" (Pi5 + Hailo-8, needs HAILO_HEF)
     # or the default CPU "movenet" (needs MOVENET_ONNX).
     if os.environ.get("POSE_BACKEND", "movenet").lower() == "hailo":
diff --git a/tests/test_picamera_source.py b/tests/test_picamera_source.py
new file mode 100644
index 0000000..24cad99
--- /dev/null
+++ b/tests/test_picamera_source.py
@@ -0,0 +1,36 @@
+"""TDD for PiCameraSource (CSI camera via picamera2) — no hardware/picamera2 needed,
+by injecting a fake Picamera2 object. Locks the SourcePort contract: read() -> (ok, frame)."""
+import numpy as np
+
+from edge.adapters.picamera_source import PiCameraSource
+
+
+class _FakePicam2:
+    def __init__(self, frames):
+        self.frames = list(frames)
+        self.i = 0
+        self.started = False
+
+    def capture_array(self):
+        frame = self.frames[self.i % len(self.frames)]
+        self.i += 1
+        return frame
+
+
+def test_read_returns_ok_and_frame():
+    frame = np.full((480, 640, 3), 128, np.uint8)
+    src = PiCameraSource(picam2=_FakePicam2([frame]))
+    ok, f = src.read()
+    assert ok is True
+    assert f.shape == (480, 640, 3)
+
+
+def test_read_none_frame_returns_false():
+    class _Empty:
+        def capture_array(self):
+            return None
+
+    src = PiCameraSource(picam2=_Empty())
+    ok, f = src.read()
+    assert ok is False
+    assert f is None