File size: 4,716 Bytes
dd87944
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
"""Capture écran / webcam — port Mark-XLVIII."""
from __future__ import annotations

import io
import json
import sys
from pathlib import Path
from typing import Any

from emo.desktop.actions._base import SkillResult
from emo.desktop.config import load_config

_IMG_MAX_W = 1280
_IMG_MAX_H = 720
_JPEG_Q = 82

try:
    import cv2
    import numpy as np

    _CV2 = True
except ImportError:
    _CV2 = False
    np = None  # type: ignore

try:
    import mss
    import mss.tools

    _MSS = True
except ImportError:
    _MSS = False

try:
    import PIL.Image

    _PIL = True
except ImportError:
    _PIL = False

_OUT = Path(__file__).resolve().parent.parent / "data" / "screenshots"


def _get_os() -> str:
    return str(load_config().get("os_system") or "windows").lower()


def _compress(img_bytes: bytes, source_format: str = "PNG") -> tuple[bytes, str]:
    if not _PIL:
        return img_bytes, f"image/{source_format.lower()}"
    try:
        img = PIL.Image.open(io.BytesIO(img_bytes)).convert("RGB")
        img.thumbnail((_IMG_MAX_W, _IMG_MAX_H), PIL.Image.BILINEAR)
        buf = io.BytesIO()
        img.save(buf, format="JPEG", quality=_JPEG_Q, optimize=False)
        return buf.getvalue(), "image/jpeg"
    except Exception:
        return img_bytes, f"image/{source_format.lower()}"


def capture_screen() -> tuple[bytes, str]:
    if not _MSS:
        try:
            import pyautogui

            _OUT.mkdir(parents=True, exist_ok=True)
            dest = _OUT / "screen_fallback.png"
            img = pyautogui.screenshot()
            img.save(dest)
            return _compress(dest.read_bytes(), "PNG")
        except ImportError as e:
            raise RuntimeError("Installez mss ou pyautogui pour les captures") from e

    with mss.mss() as sct:
        monitors = sct.monitors
        target = monitors[1] if len(monitors) > 1 else monitors[0]
        shot = sct.grab(target)
        png = mss.tools.to_png(shot.rgb, shot.size)
    return _compress(png, "PNG")


def _cv2_backend() -> int:
    if not _CV2:
        return 0
    os_name = _get_os()
    if os_name == "windows":
        return cv2.CAP_DSHOW
    if os_name == "mac":
        return cv2.CAP_AVFOUNDATION
    return cv2.CAP_ANY


def _probe_camera(index: int, backend: int, warmup: int = 5) -> bool:
    if not _CV2:
        return False
    cap = cv2.VideoCapture(index, backend)
    if not cap.isOpened():
        cap.release()
        return False
    for _ in range(warmup):
        cap.read()
    ret, frame = cap.read()
    cap.release()
    if not ret or frame is None:
        return False
    return bool(np.mean(frame) > 8)


def _detect_camera_index() -> int:
    backend = _cv2_backend()
    for idx in range(6):
        if _probe_camera(idx, backend):
            return idx
    return 0


def _get_camera_index() -> int:
    cfg = load_config()
    if "camera_index" in cfg:
        return int(cfg["camera_index"])
    return _detect_camera_index()


def capture_camera() -> tuple[bytes, str]:
    if not _CV2:
        raise RuntimeError("OpenCV (cv2) requis pour la webcam")

    index = _get_camera_index()
    backend = _cv2_backend()
    cap = cv2.VideoCapture(index, backend)
    if not cap.isOpened():
        raise RuntimeError(f"Camera index {index} could not be opened.")

    for _ in range(10):
        cap.read()
    ret, frame = cap.read()
    cap.release()

    if not ret or frame is None:
        raise RuntimeError("Camera returned no frame.")

    if _PIL:
        rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        img = PIL.Image.fromarray(rgb)
        img.thumbnail((_IMG_MAX_W, _IMG_MAX_H), PIL.Image.BILINEAR)
        buf = io.BytesIO()
        img.save(buf, format="JPEG", quality=_JPEG_Q)
        return buf.getvalue(), "image/jpeg"

    _, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, _JPEG_Q])
    return buf.tobytes(), "image/jpeg"


def run(args: dict) -> Any:
    angle = (args.get("angle") or "screen").lower().strip()
    text = (args.get("text") or args.get("prompt") or "").strip()
    try:
        if angle == "camera":
            img_b, mime = capture_camera()
            label = "camera"
        else:
            img_b, mime = capture_screen()
            label = "screen"
        _OUT.mkdir(parents=True, exist_ok=True)
        ext = "jpg" if mime == "image/jpeg" else "png"
        dest = _OUT / f"{label}_capture.{ext}"
        dest.write_bytes(img_b)
        return SkillResult.ok(
            path=str(dest),
            bytes=len(img_b),
            mime=mime,
            message=f"Capture {label}: {len(img_b):,} octets — {text[:60] or 'analyse en cours'}",
        )
    except Exception as e:
        return SkillResult.fail(str(e))