|
1 | | -# -*- coding: utf-8 -*- |
2 | 1 | from __future__ import annotations |
3 | 2 |
|
4 | 3 | import os |
| 4 | +import traceback |
5 | 5 | from typing import Any, List, Tuple, Optional |
6 | 6 | from urllib.parse import urlparse, unquote |
7 | 7 |
|
8 | 8 | import cv2 |
| 9 | +import numpy as np |
| 10 | +import requests |
9 | 11 |
|
10 | 12 | from lf_toolkit.evaluation import Result, Params |
11 | 13 |
|
12 | 14 |
|
13 | | -def _mk_result(is_correct: bool, items: List[Tuple[str, str]]) -> Result: |
14 | | - # feedback can be str or list in toolkit; tests accept both. |
15 | | - feedback = "<br>".join([f"{k}: {v}" for k, v in items]) |
| 15 | +def _pget(params: Params, key: str, default: Any) -> Any: |
16 | 16 | try: |
17 | | - return Result(is_correct=is_correct, feedback=feedback, feedback_items=items) |
18 | | - except TypeError: |
19 | | - return Result(is_correct=is_correct, feedback_items=items) |
| 17 | + return params.get(key, default) # type: ignore |
| 18 | + except Exception: |
| 19 | + try: |
| 20 | + return params[key] # type: ignore |
| 21 | + except Exception: |
| 22 | + return default |
20 | 23 |
|
21 | 24 |
|
22 | | -def _file_url_to_local_path(url: str) -> str: |
23 | | - """ |
24 | | - Convert file:// URL to local path. |
25 | | - Linux example: file:///tmp/a.jpg -> /tmp/a.jpg |
26 | | - Windows example: file:///C:/x/y.jpg -> C:/x/y.jpg |
27 | | - """ |
| 25 | +def _items_to_html(items: List[Tuple[Any, Any]]) -> str: |
| 26 | + lines: List[str] = [] |
| 27 | + for k, v in items: |
| 28 | + k = "" if k is None else str(k).strip() |
| 29 | + v = "" if v is None else str(v).strip() |
| 30 | + lines.append(f"{k}: {v}" if k else v) |
| 31 | + return "<br>".join(lines) |
| 32 | + |
| 33 | + |
| 34 | +def _escape_html(s: str) -> str: |
| 35 | + return s.replace("&", "&").replace("<", "<").replace(">", ">") |
| 36 | + |
| 37 | + |
| 38 | +def file_url_to_local_path(url: str) -> str: |
28 | 39 | parsed = urlparse(url) |
29 | 40 | path = unquote(parsed.path) |
30 | | - |
31 | | - # Windows fix: "/C:/..." -> "C:/..." |
| 41 | + # Windows: "/C:/..." -> "C:/..." |
32 | 42 | if os.name == "nt" and len(path) >= 3 and path[0] == "/" and path[2] == ":": |
33 | 43 | path = path[1:] |
34 | 44 | return path |
35 | 45 |
|
36 | 46 |
|
37 | | -def _load_bgr_image_from_url(url: str) -> Tuple[Optional[Any], Optional[str]]: |
38 | | - """ |
39 | | - Minimal loader for tests: |
40 | | - - supports file:// for local tests |
41 | | - - supports http/https (not used in tests but harmless) |
42 | | - Returns (img_bgr, error_message) |
43 | | - """ |
| 47 | +def _load_bgr_image_from_url(url: str, timeout: int = 15) -> Tuple[Optional[np.ndarray], Optional[str]]: |
44 | 48 | try: |
45 | 49 | if not isinstance(url, str) or not url: |
46 | | - return None, "Empty URL" |
| 50 | + return None, "Empty URL." |
47 | 51 |
|
48 | 52 | if url.startswith("file://"): |
49 | | - local_path = _file_url_to_local_path(url) |
| 53 | + local_path = file_url_to_local_path(url) |
50 | 54 | img = cv2.imread(local_path, cv2.IMREAD_COLOR) |
51 | 55 | if img is None: |
52 | 56 | return None, f"cv2.imread failed for local_path='{local_path}'" |
53 | 57 | return img, None |
54 | 58 |
|
55 | | - # For this minimal version, we won't actually fetch remote. |
56 | | - # If you want, you can add requests later. |
57 | 59 | if url.startswith("http://") or url.startswith("https://"): |
58 | | - return None, "Remote URL not supported in minimal test version" |
| 60 | + resp = requests.get(url, timeout=timeout) |
| 61 | + resp.raise_for_status() |
| 62 | + data = np.frombuffer(resp.content, dtype=np.uint8) |
| 63 | + img = cv2.imdecode(data, cv2.IMREAD_COLOR) |
| 64 | + if img is None: |
| 65 | + return None, "cv2.imdecode returned None (not a valid image)." |
| 66 | + return img, None |
59 | 67 |
|
60 | 68 | return None, f"Unsupported URL scheme: {url}" |
61 | | - |
62 | 69 | except Exception as e: |
63 | 70 | return None, str(e) |
64 | 71 |
|
65 | 72 |
|
66 | 73 | def evaluation_function(response: Any, answer: Any, params: Params) -> Result: |
67 | | - debug = False |
| 74 | + """ |
| 75 | + SMOKE TEST VERSION (NO YOLO) |
| 76 | + - Designed to satisfy evaluation_test.py expectations: |
| 77 | + * empty response -> is_correct=False |
| 78 | + * bad url -> feedback contains LOAD_FAIL |
| 79 | + """ |
| 80 | + items: List[Tuple[str, str]] = [] |
| 81 | + |
68 | 82 | try: |
69 | | - debug = bool(getattr(params, "debug", False) or (isinstance(params, dict) and params.get("debug"))) |
70 | | - except Exception: |
71 | | - debug = False |
72 | | - |
73 | | - # ---- Test requirement 1: missing image must be is_correct=False ---- |
74 | | - if not isinstance(response, list) or len(response) == 0: |
75 | | - items = [ |
76 | | - ("MISSING_IMAGE", "Please upload at least one image."), |
77 | | - ] |
78 | | - return _mk_result(is_correct=False, items=items) |
79 | | - |
80 | | - # Must be list[dict] with "url" |
81 | | - first = response[0] |
82 | | - url = first.get("url") if isinstance(first, dict) else None |
83 | | - if not isinstance(url, str) or not url: |
84 | | - items = [ |
85 | | - ("LOAD_FAIL", "Image item has no valid 'url' field."), |
86 | | - ] |
87 | | - return _mk_result(is_correct=False, items=items) |
88 | | - |
89 | | - # ---- Load image (this is what the tests exercise) ---- |
90 | | - img_bgr, err = _load_bgr_image_from_url(url) |
91 | | - |
92 | | - # ---- Test requirement 2: bad url must include LOAD_FAIL in feedback ---- |
93 | | - if img_bgr is None: |
94 | | - items = [ |
95 | | - ("LOAD_FAIL", f"Failed to load image from URL. ({err})"), |
96 | | - ] |
97 | | - if debug: |
98 | | - items.append(("InputURL", url)) |
99 | | - return _mk_result(is_correct=False, items=items) |
100 | | - |
101 | | - # ---- Success path for local file url test ---- |
102 | | - h, w = img_bgr.shape[:2] |
103 | | - items = [ |
104 | | - ("OK", "smoke test OK (no YOLO)"), |
105 | | - ("ImageShape", f"{w}x{h}"), |
106 | | - ] |
107 | | - if debug: |
108 | | - items.append(("InputURL", url)) |
109 | | - return _mk_result(is_correct=True, items=items) |
| 83 | + fast_return: bool = bool(_pget(params, "fast_return", True)) |
| 84 | + echo: bool = bool(_pget(params, "echo", True)) |
| 85 | + try_fetch: bool = bool(_pget(params, "try_fetch", False)) |
| 86 | + debug: bool = bool(_pget(params, "debug", True)) |
| 87 | + |
| 88 | + items.append(("SMOKE", "Hello / evaluation_function reached ✅")) |
| 89 | + items.append(("fast_return", str(fast_return))) |
| 90 | + items.append(("echo", str(echo))) |
| 91 | + items.append(("try_fetch", str(try_fetch))) |
| 92 | + items.append(("debug", str(debug))) |
| 93 | + |
| 94 | + # 1) Validate input |
| 95 | + if not isinstance(response, list) or len(response) == 0: |
| 96 | + items.append(("BAD_INPUT", "No images uploaded.")) |
| 97 | + html = _items_to_html(items) |
| 98 | + try: |
| 99 | + return Result(is_correct=False, feedback=html, feedback_items=items) |
| 100 | + except TypeError: |
| 101 | + return Result(is_correct=False, feedback_items=items) |
| 102 | + |
| 103 | + # 2) Extract first URL (tests use response[0]["url"]) |
| 104 | + first = response[0] if isinstance(response, list) and len(response) > 0 else None |
| 105 | + url = first.get("url") if isinstance(first, dict) else None |
| 106 | + |
| 107 | + if echo: |
| 108 | + items.append(("response_type", type(response).__name__)) |
| 109 | + items.append(("response_len", str(len(response)))) |
| 110 | + items.append(("first_item_type", type(first).__name__)) |
| 111 | + if isinstance(first, dict): |
| 112 | + items.append(("first_item_keys", ", ".join(sorted([str(k) for k in first.keys()])))) |
| 113 | + items.append(("first_url", str(url))) |
| 114 | + |
| 115 | + # 3) IMPORTANT: |
| 116 | + # To satisfy unit tests, if a URL exists we ALWAYS attempt a lightweight load check |
| 117 | + # (so bad file:// paths produce LOAD_FAIL). |
| 118 | + if not url: |
| 119 | + items.append(("LOAD_FAIL", "LOAD_FAIL: first image has no url field")) |
| 120 | + html = _items_to_html(items) |
| 121 | + try: |
| 122 | + return Result(is_correct=False, feedback=html, feedback_items=items) |
| 123 | + except TypeError: |
| 124 | + return Result(is_correct=False, feedback_items=items) |
| 125 | + |
| 126 | + # If user explicitly wants "fast_return" behavior on platform, |
| 127 | + # they can set try_fetch=False AND also set skip_load_check=True. |
| 128 | + # Default is to run the load-check for CI correctness. |
| 129 | + skip_load_check: bool = bool(_pget(params, "skip_load_check", False)) |
| 130 | + |
| 131 | + if (not skip_load_check) or try_fetch: |
| 132 | + img, err = _load_bgr_image_from_url(str(url)) |
| 133 | + if img is None: |
| 134 | + items.append(("LOAD_FAIL", f"LOAD_FAIL: Failed to load image. ({err})")) |
| 135 | + items.append(("url", str(url))) |
| 136 | + html = _items_to_html(items) |
| 137 | + try: |
| 138 | + return Result(is_correct=False, feedback=html, feedback_items=items) |
| 139 | + except TypeError: |
| 140 | + return Result(is_correct=False, feedback_items=items) |
| 141 | + |
| 142 | + h, w = img.shape[:2] |
| 143 | + items.append(("image_loaded", "OK")) |
| 144 | + items.append(("shape", f"{w}x{h}")) |
| 145 | + |
| 146 | + # 4) Now optionally exit early (platform smoke) |
| 147 | + if fast_return and not try_fetch: |
| 148 | + items.append(("note", "fast_return=True (no YOLO). Load-check already done.")) |
| 149 | + html = _items_to_html(items) |
| 150 | + try: |
| 151 | + return Result(is_correct=False, feedback=html, feedback_items=items) |
| 152 | + except TypeError: |
| 153 | + return Result(is_correct=False, feedback_items=items) |
| 154 | + |
| 155 | + # 5) Final message |
| 156 | + items.append(("note", "This build does NOT run YOLO. If you see this, platform execution is fine.")) |
| 157 | + html = _items_to_html(items) |
| 158 | + try: |
| 159 | + return Result(is_correct=False, feedback=html, feedback_items=items) |
| 160 | + except TypeError: |
| 161 | + return Result(is_correct=False, feedback_items=items) |
| 162 | + |
| 163 | + except Exception as e: |
| 164 | + tb = _escape_html(traceback.format_exc()) |
| 165 | + items.append(("UNHANDLED", f"{type(e).__name__}: {e}")) |
| 166 | + items.append(("TRACEBACK", tb.replace("\n", "<br>"))) |
| 167 | + html = _items_to_html(items) |
| 168 | + try: |
| 169 | + return Result(is_correct=False, feedback=html, feedback_items=items) |
| 170 | + except TypeError: |
| 171 | + return Result(is_correct=False, feedback_items=items) |
0 commit comments