| 1 | # Copyright (C) 2025 AIDC-AI |
| 2 | # |
| 3 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | # you may not use this file except in compliance with the License. |
| 5 | # You may obtain a copy of the License at |
| 6 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 7 | # Unless required by applicable law or agreed to in writing, software |
| 8 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 9 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 10 | # See the License for the specific language governing permissions and |
| 11 | # limitations under the License. |
| 12 | |
| 13 | """ |
| 14 | Image Analysis Service - ComfyUI Workflow-based implementation |
| 15 | |
| 16 | Uses Florence-2 or other vision models to analyze images and generate descriptions. |
| 17 | """ |
| 18 | |
| 19 | from typing import Optional, Literal |
| 20 | from pathlib import Path |
| 21 | |
| 22 | from comfykit import ComfyKit |
| 23 | from loguru import logger |
| 24 | |
| 25 | from pixelle_video.services.comfy_base_service import ComfyBaseService |
| 26 | |
| 27 | |
| 28 | class ImageAnalysisService(ComfyBaseService): |
| 29 | """ |
| 30 | Image analysis service - Workflow-based |
| 31 | |
| 32 | Uses ComfyKit to execute image analysis workflows (e.g., Florence-2, BLIP, etc.). |
| 33 | Returns detailed textual descriptions of images. |
| 34 | |
| 35 | Convention: workflows follow {source}/analyse_image.json pattern |
| 36 | - runninghub/analyse_image.json (default, cloud-based) |
| 37 | - selfhost/analyse_image.json (local ComfyUI) |
| 38 | |
| 39 | Usage: |
| 40 | # Use default (runninghub cloud) |
| 41 | description = await pixelle_video.image_analysis("path/to/image.jpg") |
| 42 | |
| 43 | # Use local ComfyUI |
| 44 | description = await pixelle_video.image_analysis( |
| 45 | "path/to/image.jpg", |
| 46 | source="selfhost" |
| 47 | ) |
| 48 | |
| 49 | # List available workflows |
| 50 | workflows = pixelle_video.image_analysis.list_workflows() |
| 51 | """ |
| 52 | |
| 53 | WORKFLOW_PREFIX = "analyse_" |
| 54 | WORKFLOWS_DIR = "workflows" |
| 55 | |
| 56 | def __init__(self, config: dict, core=None): |
| 57 | """ |
| 58 | Initialize image analysis service |
| 59 | |
| 60 | Args: |
| 61 | config: Full application config dict |
| 62 | core: PixelleVideoCore instance (for accessing shared ComfyKit) |
| 63 | """ |
| 64 | super().__init__(config, service_name="image_analysis", core=core) |
| 65 | |
| 66 | async def __call__( |
| 67 | self, |
| 68 | image_path: str, |
| 69 | # Workflow source selection |
| 70 | source: Literal['runninghub', 'selfhost'] = 'runninghub', |
| 71 | workflow: Optional[str] = None, |
| 72 | # ComfyUI connection (optional overrides) |
| 73 | comfyui_url: Optional[str] = None, |
| 74 | runninghub_api_key: Optional[str] = None, |
| 75 | # Additional workflow parameters |
| 76 | **params |
| 77 | ) -> str: |
| 78 | """ |
| 79 | Analyze an image using workflow |
| 80 | |
| 81 | Args: |
| 82 | image_path: Path to the image file (local or URL) |
| 83 | source: Workflow source - 'runninghub' (cloud, default) or 'selfhost' (local ComfyUI) |
| 84 | workflow: Workflow filename (optional, overrides source-based resolution) |
| 85 | comfyui_url: ComfyUI URL (optional, overrides config) |
| 86 | runninghub_api_key: RunningHub API key (optional, overrides config) |
| 87 | **params: Additional workflow parameters |
| 88 | |
| 89 | Returns: |
| 90 | str: Text description of the image |
| 91 | |
| 92 | Examples: |
| 93 | # Simplest: use default (runninghub cloud) |
| 94 | description = await pixelle_video.image_analysis("temp/06.JPG") |
| 95 | |
| 96 | # Use local ComfyUI |
| 97 | description = await pixelle_video.image_analysis( |
| 98 | "temp/06.JPG", |
| 99 | source="selfhost" |
| 100 | ) |
| 101 | |
| 102 | # Use specific workflow (bypass source-based resolution) |
| 103 | description = await pixelle_video.image_analysis( |
| 104 | "temp/06.JPG", |
| 105 | workflow="selfhost/custom_analysis.json" |
| 106 | ) |
| 107 | """ |
| 108 | from pixelle_video.utils.workflow_util import resolve_workflow_path |
| 109 | |
| 110 | # 1. Validate image path |
| 111 | image_path_obj = Path(image_path) |
| 112 | if not image_path_obj.exists(): |
| 113 | raise FileNotFoundError(f"Image file not found: {image_path}") |
| 114 | |
| 115 | # 2. Resolve workflow path using convention |
| 116 | if workflow is None: |
| 117 | # Use standardized naming: {source}/analyse_image.json |
| 118 | workflow = resolve_workflow_path("analyse_image", source) |
| 119 | logger.info(f"Using {source} workflow: {workflow}") |
| 120 | |
| 121 | # 2. Resolve workflow (returns structured info) |
| 122 | workflow_info = self._resolve_workflow(workflow=workflow) |
| 123 | |
| 124 | # 3. Build workflow parameters |
| 125 | workflow_params = { |
| 126 | "image": str(image_path) # Pass image path to workflow |
| 127 | } |
| 128 | |
| 129 | # Add any additional parameters |
| 130 | workflow_params.update(params) |
| 131 | |
| 132 | logger.debug(f"Workflow parameters: {workflow_params}") |
| 133 | |
| 134 | # 4. Execute workflow using shared ComfyKit instance from core |
| 135 | try: |
| 136 | # Get shared ComfyKit instance (lazy initialization + config hot-reload) |
| 137 | kit = await self.core._get_or_create_comfykit() |
| 138 | |
| 139 | # Determine what to pass to ComfyKit based on source |
| 140 | if workflow_info["source"] == "runninghub" and "workflow_id" in workflow_info: |
| 141 | # RunningHub: pass workflow_id |
| 142 | workflow_input = workflow_info["workflow_id"] |
| 143 | logger.info(f"Executing RunningHub workflow: {workflow_input}") |
| 144 | else: |
| 145 | # Selfhost: pass file path |
| 146 | workflow_input = workflow_info["path"] |
| 147 | logger.info(f"Executing selfhost workflow: {workflow_input}") |
| 148 | |
| 149 | result = await kit.execute(workflow_input, workflow_params) |
| 150 | |
| 151 | # 5. Extract description from result |
| 152 | if result.status != "completed": |
| 153 | error_msg = result.msg or "Unknown error" |
| 154 | logger.error(f"Image analysis failed: {error_msg}") |
| 155 | raise Exception(f"Image analysis failed: {error_msg}") |
| 156 | |
| 157 | # Extract text description from result (format varies by source) |
| 158 | description = None |
| 159 | |
| 160 | # Try format 1: Selfhost outputs (direct text in outputs) |
| 161 | # Format: {'6': {'text': ['description text']}} |
| 162 | if result.outputs: |
| 163 | for node_id, node_output in result.outputs.items(): |
| 164 | if 'text' in node_output: |
| 165 | text_list = node_output['text'] |
| 166 | if text_list and len(text_list) > 0: |
| 167 | description = text_list[0] |
| 168 | break |
| 169 | |
| 170 | # Try format 2: RunningHub raw_data (text file URL) |
| 171 | # Format: {'raw_data': [{'fileUrl': 'https://...txt', 'fileType': 'txt', ...}]} |
| 172 | if not description and result.outputs and 'raw_data' in result.outputs: |
| 173 | raw_data = result.outputs['raw_data'] |
| 174 | if raw_data and len(raw_data) > 0: |
| 175 | # Find text file entry |
| 176 | for item in raw_data: |
| 177 | if item.get('fileType') == 'txt' and 'fileUrl' in item: |
| 178 | # Download text content from URL |
| 179 | import aiohttp |
| 180 | async with aiohttp.ClientSession() as session: |
| 181 | async with session.get(item['fileUrl']) as resp: |
| 182 | if resp.status == 200: |
| 183 | description = await resp.text() |
| 184 | description = description.strip() |
| 185 | break |
| 186 | |
| 187 | if not description: |
| 188 | logger.error(f"No text found in outputs: {result.outputs}") |
| 189 | raise Exception("No description generated") |
| 190 | |
| 191 | logger.info(f"✅ Image analyzed: {description[:100]}...") |
| 192 | |
| 193 | return description |
| 194 | |
| 195 | except Exception as e: |
| 196 | logger.error(f"Image analysis error: {e}") |
| 197 | raise |
| 198 |