Add Analyze service

src/application/dtos/LLM.py

+from pydantic import BaseModel, Field
+from typing import Optional, List
+from enum import Enum
+
+class LLMProvider(str, Enum):
+    HUGGINGFACE = "huggingface"
+    OPENAI = "openai"
+    LOCAL = "local"
+
+class LLMRequestDTO(BaseModel):
+    """Input for LLM analysis requests"""
+    image_bytes: bytes = Field(..., description="Binary image data (PNG/JPEG)")
+    question: str = Field(..., max_length=500, description="Question about the chart")
+    provider: LLMProvider = Field(LLMProvider.HUGGINGFACE, description="LLM service provider")
+    max_tokens: int = Field(300, gt=0, le=2000, description="Max response length")
+    # temperature: float = Field(0.3, ge=0.1, le=1.0, description="Creativity control")
+
+    class Config:
+        json_encoders = {
+            bytes: lambda v: v.decode('latin1')  # For base64 conversion
+        }
+
+class LLMResponseDTO(BaseModel):
+    """Structured LLM output"""
+    answer: str = Field(..., description="Generated analysis")
+    model_used: str = Field(..., description="Model identifier")
+    tokens_used: Optional[int] = Field(None, description="Token count")
+    processing_time: Optional[float] = Field(None, description="Inference time in seconds")
+    confidence: Optional[float] = Field(None, ge=0, le=1, description="Confidence score")
+    raw_response: Optional[dict] = Field(None, description="Full API response")
+
+    def to_api_response(self):
+        return {
+            "analysis": self.answer,
+            "metadata": {
+                "model": self.model_used,
+                "tokens": self.tokens_used,
+                "processing_time": self.processing_time
+            }
+        }
\ No newline at end of file

src/application/dtos/error.py

+from pydantic import BaseModel
+from typing import Optional
+
+class LLMErrorDTO(BaseModel):
+    """Standardized error response"""
+    error_type: str
+    message: str
+    provider: str
+    suggestion: Optional[str] = None
+
+    @classmethod
+    def from_exception(cls, exc: Exception):
+        return cls(
+            error_type=exc.__class__.__name__,
+            message=str(exc),
+            provider="chartgemma",
+            suggestion="Try reducing image size or question length"
+        )
\ No newline at end of file

src/application/ports/analyze_service.py

+# application/ports/analysis_service_port.py
+from abc import ABC, abstractmethod
+from application.dtos.analysis import AnalysisRequestDTO, AnalysisResponseDTO
+
+class AnalysisServicePort(ABC):
+    """Port/Interface for the analysis service"""
+    
+    @abstractmethod
+    async def analyze(self, request: AnalysisRequestDTO) -> AnalysisResponseDTO:
+        """Analyze a chart image with a given question"""
+        raise NotImplementedError
\ No newline at end of file

src/application/ports/llm_service_port.py

+# domain/ports/llm_service_port.py
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from dtos.LLM import LLMResponseDTO, LLMRequestDTO
+
+class LLMServicePort(ABC):
+    """Port/Interface for LLM services"""
+    
+    @abstractmethod
+    async def analyze_chart(self, request: LLMRequestDTO) -> LLMResponseDTO:
+        """
+        Analyze a chart image with a given question
+        
+        Args:
+            image_data: Binary image data
+            question: Natural language question about the chart
+            kwargs: Additional model-specific parameters
+            
+        Returns:
+            LLMResponse: Structured response from the LLM
+        """
+        raise NotImplementedError
\ No newline at end of file

src/application/services/analyze_service.py

+# application/services/analysis_service.py
+from application.ports.analyze_service import AnalysisServicePort
+from application.dtos.analysis import AnalysisRequestDTO, AnalysisResponseDTO
+# from domain.ports.image_processor import IImageProcessor  # Domain interface
+from application.ports.llm_service_port import LLMServicePort  # Domain interface
+import uuid
+
+class AnalyzeService(AnalysisServicePort):
+    """Concrete implementation of the analysis service"""
+    
+    def __init__(self, 
+                 llm_service: LLMServicePort
+                 #image_processor: IImageProcessor
+                    ):
+        self.llm_service = llm_service
+        # self.image_processor = image_processor
+    
+    async def analyze(self, request: AnalysisRequestDTO) -> AnalysisResponseDTO:
+        """Process the analysis request"""
+        
+        # 1. Validate and preprocess image
+        # processed_image = await self.image_processor.process(request.image_bytes)
+        
+        # 2. Prepare LLM prompt
+        prompt = self._build_prompt(
+            image_data=request.image_bytes,
+            question=request.question
+        )
+        
+        # 3. Call LLM service
+        llm_response = await self.llm_service.analyze(prompt)
+        
+        # 4. Return formatted response
+        return AnalysisResponseDTO(
+            answer=llm_response.answer,
+            analysis_id=str(uuid.uuid4())  # Generate unique ID
+        )
+    
+    def _build_prompt(self, image_data: bytes, question: str) -> str:
+        """Construct the prompt for the LLM"""
+        return (
+            f"You are a data analysis expert. Analyze this chart and "
+            f"answer the following question: {question}\n"
+            f"Chart data: {image_data}"
+        )
\ No newline at end of file

src/application/services/llm_service.py

+# application/services/llm_service.py
+from application.dtos.LLM import LLMRequestDTO, LLMResponseDTO
+from application.dtos.error import LLMErrorDTO
+
+class LLMService:
+    async def analyze(self, request: LLMRequestDTO) -> LLMResponseDTO:
+        try:
+            # ... implementation ...
+            return LLMResponseDTO(
+                answer="The chart shows...",
+                model_used="chartgemma",
+                tokens_used=150,
+                processing_time=2.3
+            )
+        except Exception as e:
+            raise LLMErrorDTO.from_exception(e)
\ No newline at end of file

src/dependencies.py

@@ -8,6 +8,9 @@ from src.domain.ports.repositories.user_repository import UserRepositoryPort
 from src.domain.ports.repositories.charts_repository import ChartsRepositoryPort
 from src.infrastructure.adapters.sqlserver.sql_user_repository import SqlUserRepository
 from src.infrastructure.adapters.sqlserver.sql_charts_repository import SqlChartsRepository
+from application.services.analyze_service import AnalyzeService
+# from infrastructure.services.llm.openai_service import OpenAIService  # Concrete LLM impl
+# from infrastructure.services.image.pillow_service import PillowImageService  # Concrete impl
 from src.config import settings
 
 engine = create_async_engine(settings.DATABASE_URL, echo=True)
@@ -36,3 +39,11 @@ def get_auth_service(
 
 def get_upload_use_case(charts_repo: ChartsRepositoryPort = Depends(get_charts_repository)) -> UploadChartUseCase:
     return UploadChartUseCase(charts_repo)
+
+
+
+def get_analysis_service() -> AnalyzeService:
+    """Factory for the analysis service"""
+    llm_service = OpenAIService(api_key="your-api-key")
+    image_service = PillowImageService()
+    return AnalyzeService(llm_service, image_service)
\ No newline at end of file

src/infrastructure/api/analyze_chart.py

+# infrastructure/api/adapters/analysis_adapter.py
+from fastapi import UploadFile
+from application.services.analyze_service import AnalyzeService
+from application.dtos.analysis import AnalysisRequestDTO, AnalysisResponseDTO
+
+class FastAPIAnalysisAdapter:
+    """Adapter to convert FastAPI-specific types to our DTOs"""
+    
+    @staticmethod
+    async def to_request_dto(
+        image: UploadFile, 
+        question: str
+    ) -> AnalysisRequestDTO:
+        """Convert FastAPI UploadFile to our DTO"""
+        return AnalysisRequestDTO(
+            image_bytes=await image.read(),
+            question=question
+        )
+    
+    @staticmethod
+    def to_api_response(response_dto: AnalysisResponseDTO) -> dict:
+        """Convert our DTO to API response format"""
+        return {
+            "analysis_id": response_dto.analysis_id,
+            "answer": response_dto.answer
+        }
\ No newline at end of file

src/infrastructure/api/fastapi/routes/charts.py

 from fastapi import APIRouter, UploadFile, File, Depends
-from src.application.use_cases.upload_chart import UploadChartUseCase
+from src.application.services.analyze_service import AnalyzeService
 from src.dependencies import get_upload_use_case
 
 router = APIRouter()
 
-@router.post("/upload")
-async def upload_chart(
+@router.post("/analyze")
+async def analyze_chart(
     file: UploadFile = File(...),
-    use_case: UploadChartUseCase = Depends(get_upload_use_case)
+    service: AnalyzeService  = Depends(get_upload_use_case)
 ) -> dict:
     image_bytes = await file.read()
-    image_id = use_case.execute(image_bytes, "user123")
+    image_id = service.execute(image_bytes, "user123")
     return {"image_id": image_id}
\ No newline at end of file

test.py

+# 1. Install required package
+# pip install huggingface-hub
+
+# 2. Import and initialize
+from huggingface_hub import InferenceClient
+import base64
+import requests
+
+# 1. Convert image to base64
+with open("/home/zeinabrm/Pictures/Screenshots/Screenshot from 2025-02-17 22-46-46.png", "rb") as image_file:
+    base64_image = base64.b64encode(image_file.read()).decode('utf-8')
+
+# 2. Prepare payload
+headers = {
+    "Authorization": f"Bearer REPLACED",
+    "Content-Type": "application2/json"
+}
+
+payload = {
+    "inputs": f"<image>{base64_image}</image>\nQuestion: What does this chart show?\nAnswer:",
+    "parameters": {"max_new_tokens": 200}
+}
+
+# 3. Manually call the API
+response = requests.post(
+    "https://api-inference.huggingface.co/yuan-tian/chartgpt-llama3",
+    headers=headers,
+    json=payload
+)
+print(f"Status Code: {response.status_code}")
+print(f"Raw Response: {response.text}")
+print(response.json())
\ No newline at end of file