diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml new file mode 100644 index 0000000..a4825ea --- /dev/null +++ b/.github/workflows/deploy.yml @@ -0,0 +1,96 @@ +name: Deploy API to Cloud Run + +on: + push: + branches: + - main + pull_request: + branches: + - main + workflow_dispatch: + +env: + PROJECT_ID: jing-264314 + REGION: europe-west1 + API_SERVICE_NAME: video-logo-api + ARTIFACT_REGISTRY: europe-west1-docker.pkg.dev + +jobs: + deploy-api: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Authenticate to Google Cloud + uses: google-github-actions/auth@v2 + with: + credentials_json: ${{ secrets.GCP_SA_KEY }} + + - name: Set up Cloud SDK + uses: google-github-actions/setup-gcloud@v2 + + - name: Configure Docker for Artifact Registry + run: gcloud auth configure-docker ${{ env.ARTIFACT_REGISTRY }} + + - name: Build Docker image + working-directory: ./api + run: | + docker build -t ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} . + docker tag ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ + ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:latest + + - name: Push to Artifact Registry + run: | + docker push ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} + docker push ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:latest + + - name: Deploy to Cloud Run + run: | + gcloud run deploy ${{ env.API_SERVICE_NAME }} \ + --image ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ + --region ${{ env.REGION }} \ + --allow-unauthenticated \ + --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ + --set-secrets=/app/credentials/gcp-key.json=neurons-cloud-function-service-account-key:latest \ + --set-env-vars GOOGLE_APPLICATION_CREDENTIALS=/app/credentials/gcp-key.json \ + --max-instances 1 + + - name: Show deployment URL + run: | + echo "Service URL: $(gcloud run services describe ${{ env.API_SERVICE_NAME }} \ + --region ${{ env.REGION }} \ + --format 'value(status.url)')" + + deploy-cloud-function-worker: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Authenticate to Google Cloud + uses: google-github-actions/auth@v2 + with: + credentials_json: ${{ secrets.GCP_SA_KEY }} + + - name: Set up Cloud SDK + uses: google-github-actions/setup-gcloud@v2 + + - name: Deploy Cloud Function + working-directory: ./worker + run: | + gcloud functions deploy video-logo-worker \ + --gen2 \ + --runtime python312 \ + --entry-point process_video \ + --source . \ + --memory 16GB \ + --cpu 8 \ + --max-instances 4 \ + --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ + --region europe-west1 \ + --timeout 60m \ + --trigger-http \ + --allow-unauthenticated diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ee37881 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +api/__pycache__/* +worker/__pycache__ +senior_ai_engineer_assignment/* +results/* +test/* +.env +.envrc +adc.json +.DS_Store +logo_detection.db +*.pyc \ No newline at end of file diff --git a/README.md b/README.md index 5d0a180..fe3f7b8 100644 --- a/README.md +++ b/README.md @@ -1 +1,18 @@ -# video-logo-tracker \ No newline at end of file +# video-logo-tracker (code challenge) + +This is a code challenge for a video logo tracker. + +It consists of two parts: + +1. API +2. Worker + +## API + +The API is a FastAPI application deployed on GCP cloud run that allows you to upload a video and a logo, and then process the video to detect the logo. + +## Worker + +The Worker is a Cloud Function deployed on GCP that processes the video to detect the logo using the SIFT algorithm. + +Visit the APP \ No newline at end of file diff --git a/api/Dockerfile b/api/Dockerfile new file mode 100644 index 0000000..7006ab4 --- /dev/null +++ b/api/Dockerfile @@ -0,0 +1,21 @@ +FROM python:3.13-slim +EXPOSE 8080 + +WORKDIR /app + +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +ENV PYTHONUNBUFFERED=1 \ + # Keeps Python from generating .pyc files in the container + PYTHONDONTWRITEBYTECODE=1 \ + # Turns off buffering for easier container logging + UV_SYSTEM_PYTHON=1 + +COPY requirements.txt . + +RUN uv pip install --system --no-cache -r requirements.txt + +COPY . . + + +CMD ["uvicorn", "main:api", "--host", "0.0.0.0", "--port", "8080"] \ No newline at end of file diff --git a/api/README.md b/api/README.md new file mode 100644 index 0000000..e10e383 --- /dev/null +++ b/api/README.md @@ -0,0 +1,26 @@ +# Video Logo Tracker API + +FastAPI application for video logo processing with GCP integration. + +## Run API Locally with Docker + +1. **Download GCP credentials** and save as `adc.json` in `/api` directory. + +```bash +export GOOGLE_APPLICATION_CREDENTIALS=/path/to/adc.json +``` + +2. **Build and run**: + +```bash +cd api +docker build -t api . +docker run -v $(pwd)/adc.json:/app/credentials/gcp-key.json:ro \ + -e GOOGLE_APPLICATION_CREDENTIALS=/app/credentials/gcp-key.json \ + -p 8000:8080 api +``` + +3. **Access API**: + +-Production Docs: https://video-logo-api-606507129150.europe-west1.run.app/docs# +-Local Docs: http://localhost:8000/docs# diff --git a/api/errors.py b/api/errors.py new file mode 100644 index 0000000..386ebc7 --- /dev/null +++ b/api/errors.py @@ -0,0 +1,49 @@ +from fastapi import HTTPException + + +class InvalidVideoFileExtensionException(HTTPException): + def __init__(self, file_name: str): + super().__init__( + status_code=400, + detail=f"Invalid video file extension, must be .mp4 got {file_name}", + ) + + +class InvalidImageFileExtensionException(HTTPException): + def __init__(self, file_name: str): + super().__init__( + status_code=400, + detail=f"Invalid image file extension, must be .jpg got {file_name}", + ) + + +class MissingVideoFileException(HTTPException): + def __init__(self): + super().__init__( + status_code=400, + detail="Missing video file", + ) + + +class MissingLogoFileException(HTTPException): + def __init__(self): + super().__init__( + status_code=400, + detail="Missing logo file", + ) + + +class JobNotFoundException(HTTPException): + def __init__(self, job_id: str): + super().__init__( + status_code=404, + detail=f"Job not found with id {job_id}", + ) + + +class InvalidGCSURLException(HTTPException): + def __init__(self, url: str): + super().__init__( + status_code=400, + detail=f"Invalid GCS URL: {url}", + ) diff --git a/api/input.py b/api/input.py new file mode 100644 index 0000000..b6248a8 --- /dev/null +++ b/api/input.py @@ -0,0 +1,29 @@ +from pydantic import BaseModel + +from typing import Optional + +from models import JobStatus +from services.db import JobDB + + +class UploadInput(BaseModel): + video_file_name: str + logo_file_name: str + + +class JobPatchInput(BaseModel): + output_gcs_url: Optional[str] = None + status: JobStatus + error: Optional[str] = None + + def to_job_db(self, job: JobDB) -> JobDB: + job.status = self.status + if self.output_gcs_url: + job.output_gcs_url = self.output_gcs_url + if self.error: + job.error = self.error + + return job + + def is_gcs_url(self, url: str) -> bool: + return url.startswith("gs://") diff --git a/api/main.py b/api/main.py new file mode 100644 index 0000000..be865e8 --- /dev/null +++ b/api/main.py @@ -0,0 +1,158 @@ +from contextlib import asynccontextmanager +from fastapi import FastAPI, BackgroundTasks +from fastapi.middleware.cors import CORSMiddleware +import httpx + +from datetime import datetime, timezone +from logging import getLogger, INFO +from typing import List +import uuid + +from errors import ( + InvalidVideoFileExtensionException, + InvalidImageFileExtensionException, + JobNotFoundException, + MissingVideoFileException, + MissingLogoFileException, + InvalidGCSURLException, +) +from models import JobStatus, ProcessVideoPayload +from input import UploadInput, JobPatchInput +from output import JobOutput, CreateJobOutput + +from services.db import get, save, delete, list, init_db, drop_db, JobDB +from services.storage import delete_by_url, get_gcs_url, blob_from_file_name, FileType + +LOGGER = getLogger(__name__) +LOGGER.setLevel(INFO) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + await init_db() + LOGGER.info("database initialized") + yield + await drop_db() + LOGGER.info("database dropped") + + +api = FastAPI(lifespan=lifespan) +api.add_middleware( + CORSMiddleware, + allow_origins=[ + "https://8a33b7dc-2d6d-43a9-af02-5ff81682d7f1.lovableproject.com", + "https://id-preview--8a33b7dc-2d6d-43a9-af02-5ff81682d7f1.lovable.app", + ], + allow_methods=["*"], + allow_headers=["*"], +) + + +@api.get("/health") +async def health(): + return {"status": "ok"} + + +@api.post("/jobs", response_model=CreateJobOutput) +async def create_job(input: UploadInput) -> CreateJobOutput: + validate_upload_input(input) + id = str(uuid.uuid4()) + job = JobDB( + id=id, + video_gcs_url=get_gcs_url( + blob_from_file_name(input.video_file_name, FileType.VIDEO, id) + ), + logo_gcs_url=get_gcs_url( + blob_from_file_name(input.logo_file_name, FileType.IMAGE, id) + ), + status=JobStatus.UPLOADING, + created_at=datetime.now(tz=timezone.utc), + ) + await save(job) + return CreateJobOutput.from_files( + input.video_file_name, + input.logo_file_name, + id, + ) + + +@api.get("/jobs/{id}", response_model=JobOutput) +async def get_job(id: str) -> JobOutput: + job = await get(id) + if job is None: + raise JobNotFoundException(id) + return JobOutput.from_job_db(job) + + +@api.post("/jobs/{id}/run", response_model=JobOutput) +async def run_job(id: str, background_tasks: BackgroundTasks) -> JobOutput: + + job = await get(id) + if job is None: + raise JobNotFoundException(id) + + payload = ProcessVideoPayload( + id=id, + video_gcs_path=job.video_gcs_url, + logo_gcs_path=job.logo_gcs_url, + ) + background_tasks.add_task(trigger_cloud_function, payload.model_dump()) + job.status = JobStatus.RUNNING + job.error = None + job.output_gcs_url = None + await save(job) + return JobOutput.from_job_db(job) + + +@api.get("/jobs", response_model=List[JobOutput]) +async def list_jobs() -> List[JobOutput]: + return JobOutput.from_jobs_db(await list()) + + +@api.patch("/jobs/{id}", response_model=JobOutput) +async def job_done(id: str, input: JobPatchInput) -> JobOutput: + if input.output_gcs_url and not input.is_gcs_url(input.output_gcs_url): + raise InvalidGCSURLException(input.output_gcs_url) + job = await get(id) + if job is None: + raise JobNotFoundException(id) + job = input.to_job_db(job) + await save(job) + return JobOutput.from_job_db(job) + + +@api.delete("/jobs/{id}", status_code=204) +async def delete_job(id: str): + job = await get(id) + if job: + await delete(id) + delete_by_url(job.video_gcs_url) + delete_by_url(job.logo_gcs_url) + if job.output_gcs_url: + delete_by_url(job.output_gcs_url) + return + + +def validate_upload_input(input: UploadInput): + video_file_name = input.video_file_name + logo_file_name = input.logo_file_name + if video_file_name is None or video_file_name == "": + raise MissingVideoFileException() + if logo_file_name is None or logo_file_name == "": + raise MissingLogoFileException() + if not video_file_name.lower().endswith((".mp4")): + raise InvalidVideoFileExtensionException(video_file_name) + if not logo_file_name.lower().endswith((".jpg")): + raise InvalidImageFileExtensionException(logo_file_name) + return + + +async def trigger_cloud_function(data: dict): + async with httpx.AsyncClient(timeout=0.1) as client: + try: + await client.post( + "https://europe-west1-jing-264314.cloudfunctions.net/video-logo-worker", + json=data, + ) + except Exception as e: + LOGGER.error(f"error triggering cloud function: {e}") diff --git a/api/models.py b/api/models.py new file mode 100644 index 0000000..0b9b895 --- /dev/null +++ b/api/models.py @@ -0,0 +1,39 @@ +from pydantic import BaseModel, computed_field + +from datetime import datetime +from enum import Enum +from pathlib import Path + + +class FileType(str, Enum): + VIDEO = "video" + IMAGE = "image" + + +class SignedURLMethod(str, Enum): + GET = "GET" + PUT = "PUT" + + +class SignedURL(BaseModel): + url: str + signed_url: str + expiry: datetime + method: SignedURLMethod + + @computed_field + @property + def name(self) -> str: + return Path(self.url).name + + +class JobStatus(str, Enum): + UPLOADING = "uploading" + RUNNING = "running" + SUCCESS = "success" + FAILED = "failed" + +class ProcessVideoPayload(BaseModel): + id: str + video_gcs_path: str + logo_gcs_path: str diff --git a/api/output.py b/api/output.py new file mode 100644 index 0000000..371c58e --- /dev/null +++ b/api/output.py @@ -0,0 +1,73 @@ +from pydantic import BaseModel + +from datetime import datetime +from logging import getLogger +from typing import Optional, List + +from models import JobStatus, SignedURL, SignedURLMethod, FileType +from services.db import JobDB +from services.storage import blob_from_file_name, get_signed_url, blob_from_gcs_url + +LOGGER = getLogger(__name__) + + +class CreateJobOutput(BaseModel): + id: str + video: SignedURL + logo: SignedURL + + def from_files( + video_file_name: str, logo_file_name: str, id: str + ) -> "CreateJobOutput": + return CreateJobOutput( + id=id, + video=get_signed_url( + blob=blob_from_file_name(video_file_name, FileType.VIDEO, id), + method=SignedURLMethod.PUT, + content_type="video/mp4", + ), + logo=get_signed_url( + blob=blob_from_file_name(logo_file_name, FileType.IMAGE, id), + method=SignedURLMethod.PUT, + content_type="image/jpeg", + ), + ) + + +class JobOutput(BaseModel): + id: str + video: SignedURL + logo: SignedURL + output: Optional[SignedURL] = None + status: JobStatus + error: Optional[str] = None + created_at: datetime + updated_at: datetime + + def from_job_db(job_db: JobDB) -> "JobOutput": + return JobOutput( + id=job_db.id, + video=get_signed_url( + blob=blob_from_gcs_url(job_db.video_gcs_url), + method=SignedURLMethod.GET, + ), + logo=get_signed_url( + blob=blob_from_gcs_url(job_db.logo_gcs_url), + method=SignedURLMethod.GET, + ), + output=( + get_signed_url( + blob=blob_from_gcs_url(job_db.output_gcs_url), + method=SignedURLMethod.GET, + ) + if job_db.output_gcs_url + else None + ), + status=job_db.status, + error=job_db.error, + created_at=job_db.created_at, + updated_at=job_db.updated_at, + ) + + def from_jobs_db(jobs_db: List[JobDB]) -> List["JobOutput"]: + return [JobOutput.from_job_db(job_db) for job_db in jobs_db] diff --git a/api/requirements.txt b/api/requirements.txt new file mode 100644 index 0000000..fe51a88 --- /dev/null +++ b/api/requirements.txt @@ -0,0 +1,7 @@ +aiosqlite==0.21.0 +fastapi==0.118.1 +google-cloud-storage==3.4.0 +pydantic==2.12.0 +urllib3==2.5.0 +uvicorn==0.37.0 +httpx==0.28.1 \ No newline at end of file diff --git a/api/services/db.py b/api/services/db.py new file mode 100644 index 0000000..5253ee4 --- /dev/null +++ b/api/services/db.py @@ -0,0 +1,161 @@ +import aiosqlite +from pydantic import BaseModel + +from datetime import datetime, timezone + +from models import JobStatus +from typing import List, Optional + + +DB_PATH = "logo_detection.db" + + +class JobDB(BaseModel): + id: str + video_gcs_url: str + logo_gcs_url: str + output_gcs_url: Optional[str] = None + status: JobStatus + error: Optional[str] = None + created_at: Optional[datetime] = None + updated_at: Optional[datetime] = None + + +async def init_db(): + statuses = ", ".join([f"'{status.value}'" for status in JobStatus]) + query = f""" + CREATE TABLE IF NOT EXISTS logo_detection ( + id TEXT PRIMARY KEY, + video_gcs_url TEXT NOT NULL, + logo_gcs_url TEXT NOT NULL, + output_gcs_url TEXT, + status TEXT CHECK(status IN ({statuses})), + error TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ + async with aiosqlite.connect(DB_PATH) as db: + await db.execute(query) + await db.commit() + + +async def drop_db(): + async with aiosqlite.connect(DB_PATH) as db: + await db.execute("DROP TABLE IF EXISTS logo_detection") + await db.commit() + + +async def get(id: str) -> JobDB: + async with aiosqlite.connect(DB_PATH) as db: + async with db.execute( + """ + SELECT + id, + video_gcs_url, + logo_gcs_url, + output_gcs_url, + status, + error, + created_at, + updated_at + FROM + logo_detection + WHERE + id = :id + """, + {"id": id}, + ) as cursor: + row = await cursor.fetchone() + if row is None: + return None + return JobDB( + id=row[0], + video_gcs_url=row[1], + logo_gcs_url=row[2], + output_gcs_url=row[3], + status=row[4], + error=row[5], + created_at=row[6], + updated_at=row[7], + ) + + +async def save(job: JobDB): + job.updated_at = datetime.now(tz=timezone.utc) + async with aiosqlite.connect(DB_PATH) as db: + await db.execute( + """INSERT INTO + logo_detection (id, + video_gcs_url, + logo_gcs_url, + output_gcs_url, + status, + error, + created_at, + updated_at) + VALUES + (:id, :video_gcs_url, :logo_gcs_url, :output_gcs_url, :status, :error, :created_at, :updated_at) + ON + CONFLICT(id) DO + UPDATE + SET + video_gcs_url=:video_gcs_url, + logo_gcs_url=:logo_gcs_url, + output_gcs_url=:output_gcs_url, + status=:status, + error=:error, + updated_at=:updated_at + WHERE id=:id + """, + { + "id": job.id, + "video_gcs_url": job.video_gcs_url, + "logo_gcs_url": job.logo_gcs_url, + "output_gcs_url": job.output_gcs_url, + "status": job.status, + "error": job.error, + "created_at": job.created_at, + "updated_at": job.updated_at, + }, + ) + await db.commit() + + +async def delete(id: str): + async with aiosqlite.connect(DB_PATH) as db: + await db.execute("DELETE FROM logo_detection WHERE id = :id", {"id": id}) + await db.commit() + + +async def list() -> List[JobDB]: + async with aiosqlite.connect(DB_PATH) as db: + async with db.execute( + """ + SELECT + id, + video_gcs_url, + logo_gcs_url, + output_gcs_url, + status, + error, + created_at, + updated_at + FROM + logo_detection + """ + ) as cursor: + rows = await cursor.fetchall() + return [ + JobDB( + id=row[0], + video_gcs_url=row[1], + logo_gcs_url=row[2], + output_gcs_url=row[3], + status=row[4], + error=row[5], + created_at=row[6], + updated_at=row[6], + ) + for row in rows + ] diff --git a/api/services/storage.py b/api/services/storage.py new file mode 100644 index 0000000..c1a4df1 --- /dev/null +++ b/api/services/storage.py @@ -0,0 +1,61 @@ +from google.cloud import storage +from google.api_core.exceptions import NotFound + +from datetime import datetime, timedelta, timezone +from typing import Optional + +from models import ( + FileType, + SignedURL, + SignedURLMethod, +) + + +storage_client = storage.Client(project="jing-264314") +BUCKET_NAME = "code_challenge_logo_detection" +EXPIRY_MINUTES = 60 + + +def get_signed_url( + blob: storage.Blob, method: SignedURLMethod, content_type: Optional[str] = None +) -> SignedURL: + expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=EXPIRY_MINUTES) + return SignedURL( + url=get_gcs_url(blob), + signed_url=blob.generate_signed_url( + expiration=expiry, + method=method, + content_type=content_type, + response_disposition="attachment", + ), + expiry=expiry, + method=method, + ) + + +def get_gcs_url(blob: storage.Blob) -> str: + return f"gs://{blob.bucket.name}/{blob.name}" + + +def blob_from_gcs_url(gcs_url: str) -> storage.Blob: + if gcs_url.startswith("gs://"): + parts = gcs_url.replace("gs://", "").split("/", 1) + bucket_name, blob_path = parts[0], parts[1] + else: + raise ValueError(f"Invalid GCS URL: {gcs_url}") + return storage_client.bucket(bucket_name).blob(blob_path) + + +def blob_from_file_name(file_name: str, file_type: FileType, id: str) -> storage.Blob: + return storage_client.bucket(BUCKET_NAME).blob( + f"uploads/{id}/{file_type.value}/{file_name}" + ) + + +def delete_by_url(url: str): + try: + blob = blob_from_gcs_url(url) + blob.delete() + except NotFound: + return + return diff --git a/worker/README.md b/worker/README.md new file mode 100644 index 0000000..3eae757 --- /dev/null +++ b/worker/README.md @@ -0,0 +1,20 @@ +# video-logo-tracker worker + +Cloud Function for video logo processing. + +## Run Worker Locally with Docker + +1. **Download GCP credentials** and save as `adc.json` in `/worker` directory. + +```bash +export GOOGLE_APPLICATION_CREDENTIALS=/path/to/adc.json +``` + +2. **Build and run**: + +```bash +cd worker +uv venv +uv pip install -r requirements.txt +uv run python main.py +``` diff --git a/worker/log.py b/worker/log.py new file mode 100644 index 0000000..f675f3e --- /dev/null +++ b/worker/log.py @@ -0,0 +1,6 @@ +import logging + +logging.basicConfig( + level=logging.INFO, +) +LOGGER = logging.getLogger(__name__) diff --git a/worker/main.py b/worker/main.py new file mode 100644 index 0000000..76ee919 --- /dev/null +++ b/worker/main.py @@ -0,0 +1,76 @@ +import functions_framework +from flask import Request +from google.cloud import storage +from pydantic import BaseModel, ValidationError +import requests + +from typing import Dict +from pathlib import Path + +from sift import detect_logo_with_sift +from models import JobStatus, JobResult, ProcessVideoRequest +from storage import download_from_gcs, upload_to_gcs +from log import LOGGER + +BUCKET_NAME = "code_challenge_logo_detection" +storage_client = storage.Client(project="jing") + + +@functions_framework.http +def process_video(request: Request): + + data: Dict = request.get_json(force=True) + try: + request = ProcessVideoRequest(**data) + except ValidationError as e: + return {"error": "invalid request", "details": e.errors()}, 400 + + result = JobResult( + id=request.id, + video_gcs_path=request.video_gcs_path, + logo_gcs_path=request.logo_gcs_path, + status=JobStatus.RUNNING, + ) + + local_output_path = get_local_path(request.video_gcs_path, request.id) + + try: + detect_logo_with_sift( + logo_path=download_from_gcs(request.logo_gcs_path, request.id), + video_path=download_from_gcs(request.video_gcs_path, request.id), + output_path=local_output_path, + min_matches=15, + ratio_threshold=0.75, + ) + except Exception as e: + error = str(e) + LOGGER.error(f"error processing video: {error}") + result.status = JobStatus.FAILED + result.error = error + update_job_status(result) + return result.model_dump(), 200 + + result.output_gcs_path = upload_to_gcs(local_output_path, request.id) + result.status = JobStatus.SUCCESS + update_job_status(result) + + return result.model_dump(), 200 + + +def get_local_path(video_gcs_path: str, id: str) -> str: + file_name = Path(video_gcs_path).name + return f"/tmp/{file_name}" + + +def update_job_status(result: JobResult): + resp = requests.patch( + f"https://video-logo-api-606507129150.europe-west1.run.app/jobs/{result.id}", + json={ + "output_gcs_url": result.output_gcs_path, + "status": result.status, + "error": result.error, + }, + ) + if resp.status_code != 200: + raise Exception(f"failed to update job status: {resp.status_code} {resp.text}") + return diff --git a/worker/models.py b/worker/models.py new file mode 100644 index 0000000..9a78c78 --- /dev/null +++ b/worker/models.py @@ -0,0 +1,31 @@ +from pydantic import BaseModel +from typing import Optional + +from enum import Enum + + +class ProcessVideoRequest(BaseModel): + id: str + video_gcs_path: str + logo_gcs_path: str + + +class JobStatus(str, Enum): + RUNNING = "running" + SUCCESS = "success" + FAILED = "failed" + + +class UpdateJobStatusRequest(BaseModel): + output_gcs_url: Optional[str] = None + status: JobStatus + error: Optional[str] = None + + +class JobResult(BaseModel): + id: str + video_gcs_path: str + logo_gcs_path: str + output_gcs_path: Optional[str] = None + status: Optional[JobStatus] = None + error: Optional[str] = None diff --git a/worker/requirements.txt b/worker/requirements.txt new file mode 100644 index 0000000..b16ccb7 --- /dev/null +++ b/worker/requirements.txt @@ -0,0 +1,5 @@ +functions-framework==3.4.0 +google-cloud-storage==2.9.0 +pydantic==2.11.7 +requests==2.28.1 +opencv-python==4.12.0.88 \ No newline at end of file diff --git a/worker/sift.py b/worker/sift.py new file mode 100644 index 0000000..a3557ba --- /dev/null +++ b/worker/sift.py @@ -0,0 +1,156 @@ +import cv2 +import numpy as np +from typing import List +from log import LOGGER + +FLANN_INDEX_KDTREE = 1 + + +def detect_logo_with_sift( + logo_path: str, + video_path: str, + output_path: str, + min_matches: int = 15, + ratio_threshold: float = 0.75, +): + detector = cv2.SIFT_create() + + logo_gray, key_point, descriptor = load_logo(detector, logo_path) + cap = load_video(video_path) + fps, width, height, total_frames = get_video_info(cap) + + index_params = dict(algorithm=FLANN_INDEX_KDTREE, trees=5) + search_params = dict(checks=50) + matcher = cv2.FlannBasedMatcher(index_params, search_params) + + fourcc = cv2.VideoWriter_fourcc(*"mp4v") + out = cv2.VideoWriter(output_path, fourcc, fps, (width, height)) + + frame_count = 0 + frames_with_logo = 0 + total_matches = 0 + + while True: + ret, frame = cap.read() + if not ret: + break + + frame_count += 1 + frame_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + frame_kp, frame_desc = detector.detectAndCompute(frame_gray, None) + + if frame_desc is not None and len(frame_kp) >= 4: + matches = matcher.knnMatch(descriptor, frame_desc, k=2) + + good_matches = [] + for match_pair in matches: + if len(match_pair) == 2: + m, n = match_pair + if m.distance < ratio_threshold * n.distance: + good_matches.append(m) + + if len(good_matches) >= min_matches: + frames_with_logo += 1 + total_matches += len(good_matches) + + frame = draw_logo_outline( + frame=frame, + logo_gray=logo_gray, + key_logo=key_point, + key_frame=frame_kp, + good_matches=good_matches, + ) + + out.write(frame) + + if frame_count % 30 == 0: + progress = (frame_count / total_frames) * 100 + LOGGER.info(f"Progress: {progress:.1f}% ({frame_count}/{total_frames})") + + cap.release() + out.release() + + +def load_logo( + detector: cv2.SIFT, + logo_path: str, +) -> tuple[np.ndarray, List[cv2.KeyPoint], List[cv2.KeyPoint]]: + logo = cv2.imread(logo_path) + if logo is None: + raise ValueError(f"Could not load logo: {logo_path}") + + logo_gray = cv2.cvtColor(logo, cv2.COLOR_BGR2GRAY) + key_point, descriptor = detector.detectAndCompute(logo_gray, None) + + if descriptor is None or len(key_point) < 4: + raise ValueError("Not enough features in logo. Try a more detailed logo image.") + + return logo_gray, key_point, descriptor + + +def load_video(video_path: str) -> cv2.VideoCapture: + cap = cv2.VideoCapture(video_path) + if not cap.isOpened(): + raise ValueError(f"Could not open video: {video_path}") + return cap + + +def draw_logo_outline( + frame: np.ndarray, + logo_gray: np.ndarray, + key_logo: np.ndarray, + key_frame: np.ndarray, + good_matches: List[cv2.DMatch], +) -> np.ndarray: + src_pts = np.float32([key_logo[m.queryIdx].pt for m in good_matches]).reshape( + -1, 1, 2 + ) + dst_pts = np.float32([key_frame[m.trainIdx].pt for m in good_matches]).reshape( + -1, 1, 2 + ) + + M, mask = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, 5.0) + if M is None: + return frame + + h, w = logo_gray.shape + logo_corners = np.float32([[0, 0], [w, 0], [w, h], [0, h]]).reshape(-1, 1, 2) + frame_corners = cv2.perspectiveTransform(logo_corners, M) + + frame = cv2.polylines(frame, [np.int32(frame_corners)], True, (0, 255, 0), 3) + + x, y = int(frame_corners[0][0][0]), int(frame_corners[0][0][1]) + label = f"Logo: {len(good_matches)} matches" + cv2.putText( + frame, + label, + (x, max(y - 10, 20)), + cv2.FONT_HERSHEY_SIMPLEX, + 0.7, + (0, 255, 0), + 2, + ) + return frame + + +def get_video_info(cap: cv2.VideoCapture) -> tuple[int, int, int, int]: + fps = int(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + + LOGGER.info(f"\nVideo: {width}x{height}, {fps} FPS, {total_frames} frames") + return fps, width, height, total_frames + + +# if __name__ == "__main__": + +# logo_path = "/Users/jing/code_challenge/video-logo-tracker/senior_ai_engineer_assignment/neurons_logo.jpg" +# video_path = "/Users/jing/code_challenge/video-logo-tracker/senior_ai_engineer_assignment/Video_1.mp4" +# output_path = "/Users/jing/code_challenge/video-logo-tracker/output_video_1_fearure_matching_SIFT.mp4" + +# detect_logo_with_sift( +# logo_path=logo_path, +# video_path=video_path, +# output_path=output_path, +# ) diff --git a/worker/storage.py b/worker/storage.py new file mode 100644 index 0000000..dc5ce86 --- /dev/null +++ b/worker/storage.py @@ -0,0 +1,38 @@ +from google.cloud import storage + +from pathlib import Path +from log import LOGGER + + +storage_client = storage.Client(project="jing-264314") + + +def download_from_gcs(gcs_url, id: str): + local_dir = Path(f"/tmp/{id}/input") + local_dir.mkdir(parents=True, exist_ok=True) + local_path = local_dir / Path(gcs_url).name + + # Parse gs:// URL properly + if gcs_url.startswith("gs://"): + parts = gcs_url.replace("gs://", "").split("/", 1) + bucket_name, blob_path = parts[0], parts[1] + else: + raise ValueError(f"Invalid GCS URl: {gcs_url}") + + bucket = storage_client.bucket(bucket_name) + blob = bucket.blob(blob_path) + LOGGER.info(f"downloading {gcs_url}") + blob.download_to_filename(local_path) + LOGGER.info(f"downloaded to {local_path}") + return local_path + + +def upload_to_gcs(local_path: str, id: str) -> str: + file_name = Path(local_path).name + blob = storage_client.bucket("code_challenge_logo_detection").blob( + f"results/{id}/{file_name}" + ) + blob.upload_from_filename(local_path) + gcs_url = f"gs://{blob.bucket.name}/{blob.name}" + LOGGER.info(f"result uploaded to {gcs_url}") + return gcs_url