(self, payload: PredictRequest)
| 175 | return item |
| 176 | |
| 177 | async def predict(self, payload: PredictRequest) -> PredictResponse: |
| 178 | async with self.inference_lock: |
| 179 | cache_item = await self._get_image_state(payload.image_path) |
| 180 | state = cache_item["image_state"] |
| 181 | canvas_w, canvas_h = cache_item["canvas_size"] |
| 182 | |
| 183 | score_threshold = payload.score_threshold or self.score_threshold |
| 184 | epsilon_factor = payload.epsilon_factor or self.epsilon_factor |
| 185 | min_area = payload.min_area or self.min_area |
| 186 | |
| 187 | all_results: List[Dict] = [] |
| 188 | |
| 189 | for prompt in payload.prompts: |
| 190 | self.processor.reset_all_prompts(state) |
| 191 | result_state = self.processor.set_text_prompt(prompt=prompt, state=state) |
| 192 | masks = result_state.get("masks", []) |
| 193 | boxes = result_state.get("boxes", []) |
| 194 | scores = result_state.get("scores", []) |
| 195 | |
| 196 | if masks is None or len(masks) == 0: |
| 197 | continue |
| 198 | |
| 199 | num_masks = masks.shape[0] if isinstance(masks, torch.Tensor) else len(masks) |
| 200 | for i in range(num_masks): |
| 201 | score_val = scores[i] |
| 202 | score_val = score_val.item() if hasattr(score_val, "item") else float(score_val) |
| 203 | if score_val < score_threshold: |
| 204 | continue |
| 205 | |
| 206 | box = boxes[i] |
| 207 | bbox = box.detach().cpu().numpy().tolist() if isinstance(box, torch.Tensor) else box |
| 208 | bbox = [int(v) for v in bbox] |
| 209 | x1, y1, x2, y2 = bbox |
| 210 | |
| 211 | mask = masks[i] |
| 212 | binary_mask = mask.detach().cpu().numpy() if isinstance(mask, torch.Tensor) else np.array(mask) |
| 213 | if binary_mask.ndim > 2: |
| 214 | binary_mask = binary_mask.squeeze() |
| 215 | binary_mask = (binary_mask > 0.5).astype(np.uint8) * 255 |
| 216 | |
| 217 | polygon = _extract_polygon(binary_mask, epsilon_factor) |
| 218 | if len(polygon) == 0 or cv2.contourArea(np.array(polygon)) < min_area: |
| 219 | continue |
| 220 | |
| 221 | mask_payload = None |
| 222 | mask_shape = None |
| 223 | if payload.return_masks: |
| 224 | mask_shape = [binary_mask.shape[0], binary_mask.shape[1]] |
| 225 | if payload.mask_format == "png": |
| 226 | mask_payload = _encode_mask_png(binary_mask) |
| 227 | else: |
| 228 | mask_payload = _encode_mask_rle(binary_mask) |
| 229 | |
| 230 | all_results.append( |
| 231 | self._build_detection( |
| 232 | prompt=prompt, |
| 233 | score=score_val, |
| 234 | bbox=bbox, |
no test coverage detected