Merge pull request #24 from ks6088ts-labs/feature/issue-7_image-analysis

add Azure AI Vision solution

Author: ks6088ts

azure_ai_vision.env.sample

@@ -0,0 +1,2 @@
+AZURE_AI_VISION_API_KEY="<your-api-key>"
+AZURE_AI_VISION_ENDPOINT="<your-endpoint>"

backend/fastapi.py

@@ -1,6 +1,7 @@
 from fastapi import FastAPI
 from fastapi.openapi.utils import get_openapi
 
+from backend.routers import azure_ai_vision as azure_ai_vision_router
 from backend.routers import azure_openai as azure_openai_router
 from backend.routers import azure_storage as azure_storage_router
 from backend.routers import document_intelligence as document_intelligence_router
@@ -12,6 +13,7 @@
 app.include_router(azure_openai_router.router)
 app.include_router(document_intelligence_router.router)
 app.include_router(azure_storage_router.router)
+app.include_router(azure_ai_vision_router.router)
 
 
 def custom_openapi():

backend/internals/azure_ai_vision.py

@@ -0,0 +1,66 @@
+from logging import getLogger
+
+from azure.ai.vision.imageanalysis import ImageAnalysisClient
+from azure.ai.vision.imageanalysis.models import VisualFeatures
+from azure.core.credentials import AzureKeyCredential
+
+from backend.settings.azure_ai_vision import Settings
+
+logger = getLogger(__name__)
+
+
+class AzureAiVisionClient:
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+
+    def get_image_analysis_client(self) -> ImageAnalysisClient:
+        return ImageAnalysisClient(
+            endpoint=self.settings.azure_ai_vision_endpoint,
+            credential=AzureKeyCredential(self.settings.azure_ai_vision_api_key),
+        )
+
+    def analyze_image(
+        self,
+        image: bytes,
+    ) -> dict:
+        image_analysis_client = self.get_image_analysis_client()
+        result = image_analysis_client.analyze(
+            image_data=image,
+            visual_features=[
+                VisualFeatures.CAPTION,
+                VisualFeatures.READ,
+            ],
+        )
+        logger.info("Analyzed image")
+        return result.as_dict()
+
+    def vectorize_image(
+        self,
+        image: bytes,
+    ) -> dict:
+        # FIXME: replace with Azure SDK when available
+        from urllib.parse import urljoin
+
+        import requests
+
+        url = urljoin(
+            self.settings.azure_ai_vision_endpoint,
+            "/computervision/retrieval:vectorizeImage",
+        )
+        params = {
+            "overload": "stream",
+            "api-version": "2023-02-01-preview",
+            "modelVersion": "latest",
+        }
+        headers = {
+            "Content-Type": "application/octet-stream",
+            "Ocp-Apim-Subscription-Key": self.settings.azure_ai_vision_api_key,
+        }
+        response = requests.post(
+            url=url,
+            params=params,
+            headers=headers,
+            data=image,
+        )
+        response.raise_for_status()
+        return response.json()

backend/routers/azure_ai_vision.py

@@ -0,0 +1,55 @@
+from logging import getLogger
+
+from fastapi import APIRouter, UploadFile
+
+from backend.internals import azure_ai_vision
+from backend.schemas import azure_ai_vision as azure_ai_vision_schemas
+from backend.settings.azure_ai_vision import Settings
+
+logger = getLogger(__name__)
+client = azure_ai_vision.AzureAiVisionClient(
+    settings=Settings(),
+)
+
+router = APIRouter(
+    prefix="/azure_ai_vision",
+    tags=["azure_ai_vision"],
+    responses={404: {"description": "Not found"}},
+)
+
+
+@router.post(
+    "/image/analyze/",
+    response_model=azure_ai_vision_schemas.ImageAnalysisResponse,
+    status_code=200,
+)
+async def analyze_image(file: UploadFile):
+    try:
+        content = await file.read()
+        result = client.analyze_image(
+            image=content,
+        )
+    except Exception as e:
+        logger.error(f"Failed to analyze image: {e}")
+        raise
+    return azure_ai_vision_schemas.ImageAnalysisResponse(
+        result=result,
+    )
+
+
+@router.post(
+    "/image/vectorize/",
+    status_code=200,
+)
+async def vectorize_image(file: UploadFile):
+    try:
+        content = await file.read()
+        result = client.vectorize_image(
+            image=content,
+        )
+    except Exception as e:
+        logger.error(f"Failed to vectorize image: {e}")
+        raise
+    return azure_ai_vision_schemas.VectorizeImageResponse(
+        result=result,
+    )

backend/schemas/azure_ai_vision.py

@@ -0,0 +1,13 @@
+from logging import getLogger
+
+from pydantic import BaseModel
+
+logger = getLogger(__name__)
+
+
+class ImageAnalysisResponse(BaseModel):
+    result: dict
+
+
+class VectorizeImageResponse(BaseModel):
+    result: dict

backend/settings/azure_ai_vision.py

@@ -0,0 +1,11 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    azure_ai_vision_endpoint: str = "https://<name>.cognitiveservices.azure.com/"
+    azure_ai_vision_api_key: str = "<api-key>"
+
+    model_config = SettingsConfigDict(
+        env_file="azure_ai_vision.env",
+        env_file_encoding="utf-8",
+    )

docs/README.md

@@ -24,15 +24,15 @@
 
 ## Azure
 
-### Azure OpenAI Service
-
-- [Azure-Samples/openai](https://github.com/Azure-Samples/openai)
-
 ### Azure Functions
 
 - [Quickstart: Create a Python function in Azure from the command line](https://learn.microsoft.com/en-us/azure/azure-functions/create-first-function-cli-python?tabs=linux%2Cbash%2Cazure-cli%2Cbrowser)
 - [Using FastAPI Framework with Azure Functions](https://learn.microsoft.com/en-us/samples/azure-samples/fastapi-on-azure-functions/fastapi-on-azure-functions/)
 
+### Azure OpenAI Service
+
+- [Azure-Samples/openai](https://github.com/Azure-Samples/openai)
+
 ### Azure AI Speech Service
 
 - [How to recognize speech > Recognize speech from a microphone](https://learn.microsoft.com/en-us/azure/ai-services/speech-service/how-to-recognize-speech?pivots=programming-language-python#recognize-speech-from-a-microphone)
@@ -42,6 +42,13 @@
 - [How to recognize speech > Use continuous recognition](https://learn.microsoft.com/en-us/azure/ai-services/speech-service/how-to-recognize-speech?pivots=programming-language-python#use-continuous-recognition)
 - [Task 02 - Perform speech requests with Streamlit (40 minutes)](https://microsoft.github.io/TechExcel-Implementing-automation-practices-using-Azure-OpenAI/docs/04_implement_audio_transcription/0402.html)
 
+### Azure AI Vision
+
+- [Quickstart: Image Analysis 4.0](https://learn.microsoft.com/en-us/azure/ai-services/computer-vision/quickstarts-sdk/image-analysis-client-library-40?tabs=visual-studio%2Cwindows&pivots=programming-language-python)
+- [Do image retrieval using multimodal embeddings (version 4.0)](https://learn.microsoft.com/en-us/azure/ai-services/computer-vision/how-to/image-retrieval?tabs=python)
+- [Multimodal embeddings (version 4.0)](https://learn.microsoft.com/en-us/azure/ai-services/computer-vision/concept-image-retrieval)
+- [Image Retrieval - Vectorize Image From Image Stream](https://learn.microsoft.com/en-us/rest/api/computervision/image-retrieval/vectorize-image-from-image-stream?view=rest-computervision-2023-02-01-preview&tabs=HTTP)
+
 ### Azure AI Document Intelligence
 
 - [Azure AI Document Intelligence client library for Python](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/documentintelligence/azure-ai-documentintelligence/README.md)

frontend/entrypoint.py

@@ -1,6 +1,6 @@
 import logging
 
-from frontend.solutions import azure_storage, document_intelligence, sandbox, transcription
+from frontend.solutions import azure_ai_vision, azure_storage, document_intelligence, sandbox, transcription
 from frontend.solutions.types import SolutionType
 
 logger = logging.getLogger(__name__)
@@ -31,3 +31,8 @@ def start(
             backend_url=backend_url,
             log_level=log_level,
         )
+    if solution_type == SolutionType.AZURE_AI_VISION:
+        return azure_ai_vision.start(
+            backend_url=backend_url,
+            log_level=log_level,
+        )

frontend/solutions/azure_ai_vision.py

@@ -0,0 +1,45 @@
+import asyncio
+import logging
+from io import BytesIO
+from urllib.parse import urljoin
+
+import streamlit as st
+
+from frontend.solutions.utilities import http_post_file
+
+logger = logging.getLogger(__name__)
+
+
+def start(
+    backend_url: str,
+    log_level: int,
+):
+    logger.setLevel(log_level)
+    logger.debug(f"set log level to {log_level}")
+
+    st.header("Azure AI Vision")
+
+    file_uploader = st.file_uploader(
+        label="Choose a file",
+        key="file_uploader",
+    )
+
+    analyze_button = st.button(
+        label="Analyze",
+        key="analyze_button",
+    )
+
+    if file_uploader is not None:
+        st.image(file_uploader, caption="Uploaded image")
+        if analyze_button:
+            with st.spinner("Analyzing..."):
+                bytes_data = file_uploader.getvalue()
+                response = asyncio.run(
+                    http_post_file(
+                        url=urljoin(base=backend_url, url="/azure_ai_vision/image/analyze/"),
+                        data_bytes_io=BytesIO(bytes_data),
+                    )
+                )
+                st.write(response)
+    else:
+        st.warning("Please upload a file first")

frontend/solutions/types.py

@@ -6,3 +6,4 @@ class SolutionType(Enum):
     TRANSCRIPTION = "TRANSCRIPTION"
     DOCUMENT_INTELLIGENCE = "DOCUMENT_INTELLIGENCE"
     AZURE_STORAGE = "AZURE_STORAGE"
+    AZURE_AI_VISION = "AZURE_AI_VISION"