Detect Impersonations with Deepfaic Edge

Published: 17 June 2026

Deepfaic Edge lets you run impersonation checks entirely inside your own environment. In this tutorial, we will build a Python command-line script that stages a media file into the edge runtime uploads directory, creates an edge job, polls the partial job endpoint for progress, fetches the full job record when processing completes, and prints the final scores together with the total pipeline time.

What We're Building

We are going to create a Python script that can:

1. Confirm the edge API is reachable with your local credential
2. Stage a media file into the default edge uploads directory
3. Create a job with POST /v1/jobs
4. Poll GET /v1/jobs/partial/{job_id} every 2 seconds
5. Fetch the full job with GET /v1/jobs/{job_id}
6. Print the raw scores, thresholds, chunk scores, frame scores, and total pipeline time

Prerequisites

Before you start, make sure you have:

• A Deepfaic Edge runtime already installed
• Access to the edge runtime root that contains CONFIG.yaml, api_cred.txt, and docker-compose.yml
• Python 3.8 or higher
• The requests library installed
• A media file you want to analyze

Install the Python dependency:

pip install requests

Step 1: Confirm Edge Is Installed and Running

From the edge runtime root, run the installer check command:

./deepfaic-installer check

This validates the managed runtime files and tells you if anything is missing or stale.

If you want a quick API-level confirmation as well, you can call the health endpoint:

curl http://localhost:9123/v1/health

Step 2: Ensure api_cred.txt Is Configured

The script below reads the edge API key from <runtime-root>/api_cred.txt. Make sure that file contains a valid df_ key:

grep '^df_' api_cred.txt

If the file is empty or missing, ask your Deepfaic administrator for an edge API key and write it into api_cred.txt.

The Complete Script

Save this script as deepfaic_edge.py in the edge runtime root:

"""
Deepfaic Edge tutorial script.

This script:
1. Loads the edge API key from api_cred.txt
2. Verifies the edge API is reachable
3. Copies a media file into the default edge uploads directory
4. Creates a job through the edge API
5. Polls the partial job endpoint until completion
6. Fetches the full job and prints raw scores

Usage:
    python deepfaic_edge.py /path/to/media.mp4
    python deepfaic_edge.py /path/to/media.mp4 --base-url http://localhost:9123
    python deepfaic_edge.py /path/to/media.mp4 --runtime-root /opt/deepfaic
"""

from __future__ import annotations

import argparse
import shutil
import sys
import time
from pathlib import Path
from typing import Any

import requests


DEFAULT_BASE_URL = "http://localhost:9123"
CONTAINER_UPLOADS_ROOT = Path("/var/lib/deepfaic/uploads")


class DeepfaicEdgeClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update(
            {
                "Authorization": f"Bearer {api_key}",
                "Accept": "application/json",
            }
        )

    def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
        response = self.session.request(
            method,
            f"{self.base_url}{path}",
            timeout=60,
            **kwargs,
        )
        response.raise_for_status()

        payload = response.json()
        if not payload.get("status", False):
            raise RuntimeError(
                f"Edge API returned an error for {path}: {payload.get('message', 'unknown error')}"
            )
        return payload

    def health(self) -> dict[str, Any]:
        payload = self._request("GET", "/v1/health")
        return payload["data"]

    def create_job(self, input_path: str, title: str) -> int:
        payload = self._request(
            "POST",
            "/v1/jobs",
            json={
                "input_path": input_path,
                "title": title,
            },
        )
        return int(payload["data"]["job_id"])

    def get_partial_job(self, job_id: int) -> dict[str, Any]:
        payload = self._request("GET", f"/v1/jobs/partial/{job_id}")
        records = payload.get("data", [])
        if not records:
            raise RuntimeError(f"Partial job {job_id} was not found")
        return records[0]

    def get_full_job(self, job_id: int) -> dict[str, Any]:
        payload = self._request("GET", f"/v1/jobs/{job_id}")
        records = payload.get("data", [])
        if not records:
            raise RuntimeError(f"Full job {job_id} was not found")
        return records[0]


def load_api_key(runtime_root: Path) -> str:
    credential_path = runtime_root / "api_cred.txt"
    if not credential_path.exists():
        raise RuntimeError(
            f"Missing api_cred.txt at {credential_path}. Run this script from the edge runtime root or pass --runtime-root."
        )

    api_key = credential_path.read_text(encoding="utf-8").strip()
    if not api_key.startswith("df_"):
        raise RuntimeError(
            f"{credential_path} does not contain a valid edge API key starting with 'df_'."
        )
    return api_key


def stage_media_file(source_path: Path, runtime_root: Path) -> tuple[Path, str]:
    if not source_path.exists():
        raise RuntimeError(f"Media file was not found: {source_path}")
    if not source_path.is_file():
        raise RuntimeError(f"Expected a file, received: {source_path}")

    uploads_dir = runtime_root / "data" / "uploads"
    uploads_dir.mkdir(parents=True, exist_ok=True)

    staged_name = f"{int(time.time())}-{source_path.name}"
    staged_host_path = uploads_dir / staged_name
    shutil.copy2(source_path, staged_host_path)

    container_path = str(CONTAINER_UPLOADS_ROOT / staged_name)
    return staged_host_path, container_path


def format_optional(value: Any) -> str:
    if value is None:
        return "n/a"
    return str(value)


def format_float(value: Any) -> str:
    if value is None:
        return "n/a"
    try:
        return f"{float(value):.4f}"
    except (TypeError, ValueError):
        return str(value)


def print_lane(name: str, lane: dict[str, Any]) -> None:
    print(f"{name} score: {format_float(lane.get('score'))}")
    print(f"{name} threshold: {format_float(lane.get('threshold'))}")
    print(f"{name} coverage: {format_float(lane.get('coverage'))}")
    print(f"{name} status: {format_optional(lane.get('status'))}")
    print(f"{name} error: {format_optional(lane.get('error'))}")
    print(f"{name} chunk_scores: {lane.get('chunk_scores', [])}")

    if "frame_scores" in lane:
        print(f"{name} frame_scores: {lane.get('frame_scores', [])}")
    if "face_box" in lane:
        print(f"{name} face_box: {lane.get('face_box')}")

    print()


def wait_for_completion(
    client: DeepfaicEdgeClient,
    job_id: int,
    poll_interval_seconds: float,
    started_at: float,
) -> dict[str, Any]:
    while True:
        partial = client.get_partial_job(job_id)
        elapsed = time.perf_counter() - started_at

        status = partial.get("status", "unknown")
        progress = partial.get("progress")
        progress_text = (
            f"{float(progress):.2f}%"
            if isinstance(progress, (int, float))
            else "n/a"
        )

        print(
            f"[{elapsed:7.2f}s] job_id={job_id} status={status} progress={progress_text}"
        )

        if "video_score" in partial:
            print(
                f"  partial video_score={format_float(partial.get('video_score'))} "
                f"threshold={format_float(partial.get('video_threshold'))}"
            )
        if "audio_score" in partial:
            print(
                f"  partial audio_score={format_float(partial.get('audio_score'))} "
                f"threshold={format_float(partial.get('audio_threshold'))}"
            )
        if "image_score" in partial:
            print(
                f"  partial image_score={format_float(partial.get('image_score'))} "
                f"threshold={format_float(partial.get('image_threshold'))}"
            )

        if status == "complete":
            return partial
        if status in {"error", "failed"}:
            raise RuntimeError(f"Job {job_id} failed with status '{status}'")

        time.sleep(poll_interval_seconds)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Run a Deepfaic Edge impersonation check for a media file."
    )
    parser.add_argument("media_file", help="Path to the local media file to analyze")
    parser.add_argument(
        "--base-url",
        default=DEFAULT_BASE_URL,
        help=f"Edge API base URL. Defaults to {DEFAULT_BASE_URL}",
    )
    parser.add_argument(
        "--runtime-root",
        default=".",
        help="Path to the edge runtime root containing CONFIG.yaml and api_cred.txt",
    )
    parser.add_argument(
        "--poll-interval",
        type=float,
        default=2.0,
        help="Polling interval in seconds for the partial job endpoint",
    )
    args = parser.parse_args()

    runtime_root = Path(args.runtime_root).resolve()
    media_path = Path(args.media_file).resolve()

    try:
        api_key = load_api_key(runtime_root)
        client = DeepfaicEdgeClient(args.base_url, api_key)

        health = client.health()
        print("Deepfaic Edge service:", health.get("service"))
        print("Organization:", health.get("org"))
        print("Allowed models:", ", ".join(health.get("allowed_models", [])))
        print()

        staged_host_path, container_input_path = stage_media_file(media_path, runtime_root)
        print("Staged media file:", staged_host_path)
        print("Container input_path:", container_input_path)
        print()

        job_title = f"Edge impersonation check - {media_path.name}"

        started_at = time.perf_counter()
        job_id = client.create_job(container_input_path, job_title)
        print(f"Created edge job {job_id}")
        print()

        wait_for_completion(
            client=client,
            job_id=job_id,
            poll_interval_seconds=args.poll_interval,
            started_at=started_at,
        )

        full_job = client.get_full_job(job_id)
        total_seconds = time.perf_counter() - started_at

        print()
        print("Final job details")
        print("-----------------")
        print("job_id:", full_job.get("id"))
        print("title:", full_job.get("title"))
        print("file_type:", full_job.get("file_type"))
        print("status:", full_job.get("status"))
        print("models:", ", ".join(full_job.get("models", [])))
        print("video_time:", format_optional(full_job.get("video_time")))
        print("audio_time:", format_optional(full_job.get("audio_time")))
        print("progress:", format_optional(full_job.get("progress")))
        print(f"pipeline_time_seconds: {total_seconds:.2f}")
        print()

        if "video" in full_job:
            print_lane("video", full_job["video"])
        if "audio" in full_job:
            print_lane("audio", full_job["audio"])
        if "image" in full_job:
            print_lane("image", full_job["image"])

    except KeyboardInterrupt:
        print("Interrupted by user")
        sys.exit(1)
    except requests.HTTPError as error:
        response_text = ""
        if error.response is not None:
            response_text = error.response.text
        print(f"HTTP error: {error}\n{response_text}")
        sys.exit(1)
    except Exception as error:
        print(f"Error: {error}")
        sys.exit(1)


if __name__ == "__main__":
    main()

How to Use the Script

Step 1: Save the Script

Save the code as deepfaic_edge.py in your edge runtime root.

Step 2: Run the Script

Analyze a file using the default edge API URL:

python deepfaic_edge.py ./sample/video.mp4

Point the script at a different published edge URL when needed:

python deepfaic_edge.py ./sample/video.mp4 --base-url http://localhost:9123

If you are not running the script from the runtime root, pass it explicitly:

python deepfaic_edge.py ./sample/video.mp4 --runtime-root /opt/deepfaic

Step 3: Watch the Job Progress

The script polls the partial edge endpoint every 2 seconds and prints the current status, progress, and any scores that are already available:

Deepfaic Edge service: deepfaic-edge
Organization: acme
Allowed models: facia, diffusion, dfaudio

Staged media file: /opt/deepfaic/data/uploads/1781827200-video.mp4
Container input_path: /var/lib/deepfaic/uploads/1781827200-video.mp4

Created edge job 42

[   2.01s] job_id=42 status=processing progress=35.00%
  partial video_score=0.2147 threshold=0.4000
[   4.02s] job_id=42 status=processing progress=78.00%
  partial video_score=0.2651 threshold=0.4000
  partial audio_score=0.0419 threshold=0.4000
[   8.11s] job_id=42 status=complete progress=100.00%
  partial video_score=0.2874 threshold=0.4000
  partial audio_score=0.0523 threshold=0.4000

Understanding the Final Output

When the job completes, the script fetches the full edge job payload and prints the key fields:

• Job metadata: id, title, file_type, status, models
• Lane timings: video_time, audio_time
• Total pipeline time measured by the script
• Lane scores and thresholds
• Lane coverage
• Lane chunk_scores
• Lane frame_scores when returned by the full endpoint
• face_box when returned for a video lane that used facia

Conclusion

You now have a practical Deepfaic Edge command-line script that follows the edge runtime flow. From here, you can adapt the script to batch processing, dashboards, or internal review pipelines.

---

Have questions or need help? Reach out to Deepfaic team.