    generative_ai/sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SageMaker JumpStart - deploy VS model\n",
    "\n",
    "This notebook demonstrates how to use the SageMaker Python SDK to deploy a SageMaker JumpStart VS model and invoke the endpoint.\n",
    "\n",
    "The model used in this notebook is Meta's SAM 2.1 model. This model is used in segmentation tasks over images and video. This notebook will be demonstrating the video segementation use case using the SAM 2.1 model. We will be using both point and box prompts in various contexts to infer segmentation masks over a sample video. Single object and combination prompts of points and a box is supported as well as the ability to correct masks and re-infer the masks.\n",
    "\n",
    "NOTE: The masks generated are on a frame by frame basis using surrounding frame inferences to infer on each frame. This means that adding points after propgating the prompts across the video will only correct the alredy inferred masks. To remask a new inference sessions needs to be started."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b4e567a4",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "This notebook's CI test result for us-west-2 is as follows. CI test results in other regions can be found at the end of the notebook.\n",
    "\n",
    "![This us-west-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-west-2/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install opencv-python-headless"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install -U sagemaker"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "%conda install -y ffmpeg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.jumpstart.model import JumpStartModel"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Select your desired model ID. You can search for available models in the [Built-in Algorithms with pre-trained Model Table](https://sagemaker.readthedocs.io/en/stable/doc_utils/pretrainedmodels.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jumpStartAlterations": [
     "modelIdOnly"
    ],
    "tags": []
   },
   "outputs": [],
   "source": [
    "model_id = \"meta-vs-sam-2-1-hiera-small\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploy model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the model ID, define your model as a JumpStart model. You can deploy the model on other instance types by passing `instance_type` to `JumpStartModel`. See [Deploy publicly available foundation models with the JumpStartModel class](https://docs.aws.amazon.com/sagemaker/latest/dg/jumpstart-foundation-models-use-python-sdk.html#jumpstart-foundation-models-use-python-sdk-model-class) for more configuration options."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = JumpStartModel(model_id=model_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can now deploy your JumpStart model. The deployment might take few minutes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictor = model.deploy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SAM2 Endpoint Testing\n",
    "\n",
    "This notebook demonstrates testing of SAM2 endpoint functionality using modular test classes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import logging\n",
    "import sys\n",
    "import boto3\n",
    "import json\n",
    "from typing import List, Dict\n",
    "import numpy as np\n",
    "import cv2\n",
    "from pathlib import Path\n",
    "import base64\n",
    "import zlib\n",
    "from typing import Union, Optional\n",
    "import time\n",
    "from datetime import datetime\n",
    "from IPython.display import Video, display\n",
    "from IPython.core.display import HTML\n",
    "\n",
    "logging.basicConfig(\n",
    "    level=logging.INFO,\n",
    "    format=\"%(asctime)s [%(levelname)s] %(message)s\",\n",
    "    handlers=[logging.StreamHandler(sys.stdout)],\n",
    ")\n",
    "\n",
    "logger = logging.getLogger(__name__)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Stream Parser\n",
    "\n",
    "Due to functionality of SAM2.1 generating masks we need a way to make sure that we return the large payload without reaching sagemaker limits. Therefore we leverage Sagemaker Streaming Responses to stream back chunks of our payload. \n",
    "We expect Json Line items from the endpoint so here we have a StreamParser class to parse the chunks we add to it and parse out complete Json Line items."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class StreamParser:\n",
    "    \"\"\"\n",
    "    Parses streaming JSON responses with nested structure support.\n",
    "    Maintains a buffer (max 10MB) and accumulates valid JSON dictionaries.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, max_size: int = 10 * 1024 * 1024):\n",
    "        self.buffer = bytearray()\n",
    "        self.parsed_responses: List[Dict] = []\n",
    "        self.max_size = max_size\n",
    "        self.decoder = json.JSONDecoder()\n",
    "\n",
    "    def write(self, event: dict) -> None:\n",
    "        \"\"\"Processes streaming event and extracts JSON objects.\"\"\"\n",
    "        try:\n",
    "            if payload := event.get(\"PayloadPart\", {}).get(\"Bytes\"):\n",
    "                self.buffer.extend(payload)\n",
    "                self._process_buffer()\n",
    "        except Exception as e:\n",
    "            logger.error(f\"Processing error: {e}\")\n",
    "\n",
    "    def _process_buffer(self) -> None:\n",
    "        \"\"\"Extracts and parses complete JSON objects from buffer.\"\"\"\n",
    "        if len(self.buffer) > self.max_size:\n",
    "            logger.warning(\"Buffer exceeded max size, clearing\")\n",
    "            self.buffer.clear()\n",
    "            return\n",
    "\n",
    "        try:\n",
    "            index = 0\n",
    "            buffer_str = self.buffer.decode(\"utf-8\")\n",
    "            buffer_len = len(buffer_str)\n",
    "            complete_objects = 0\n",
    "            incomplete_objects = 0\n",
    "\n",
    "            while index < buffer_len:\n",
    "                try:\n",
    "                    # Look for start of JSON object\n",
    "                    while index < buffer_len and buffer_str[index] not in \"{[\":\n",
    "                        index += 1\n",
    "\n",
    "                    if index >= buffer_len:\n",
    "                        break\n",
    "\n",
    "                    # Try to parse JSON object\n",
    "                    try:\n",
    "                        result, end = self.decoder.raw_decode(buffer_str[index:])\n",
    "                        if isinstance(result, dict):\n",
    "                            self.parsed_responses.append(result)\n",
    "                            complete_objects += 1\n",
    "                        index += end\n",
    "                    except json.JSONDecodeError as e:\n",
    "                        # Check if we might have an incomplete object\n",
    "                        if buffer_str[index] == \"{\" and buffer_str[-1] != \"}\":\n",
    "                            incomplete_objects += 1\n",
    "                            break\n",
    "                        index += 1\n",
    "\n",
    "                except Exception as e:\n",
    "                    logger.error(f\"Error processing buffer at position {index}: {e}\")\n",
    "                    index += 1\n",
    "\n",
    "            # Remove processed data if we found complete objects\n",
    "            if complete_objects > 0:\n",
    "                processed_bytes = len(buffer_str[:index].encode(\"utf-8\"))\n",
    "                del self.buffer[:processed_bytes]\n",
    "\n",
    "        except UnicodeDecodeError as e:\n",
    "            logger.warning(\n",
    "                f\"Unicode decode error: {e}. Buffer might contain incomplete UTF-8 sequences.\"\n",
    "            )\n",
    "\n",
    "    def get_responses(self) -> List[Dict]:\n",
    "        \"\"\"Returns accumulated parsed JSON responses.\"\"\"\n",
    "        return self.parsed_responses"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Util Funtions\n",
    "\n",
    "Computer Vision tasks require that we transform videos into formats that are easily transmitable and visualizeable. The following functions are set up to help with those intermediate steps.\n",
    "\n",
    "Functions:\n",
    "* encode video (Mandatory)\n",
    "  * encode the binary data of an video into base64 for the endpoint to decode and infer upon. \n",
    "* decompress_mask (Mandatory)\n",
    "  * as mentioned before we compress and stream the data back and to complement that we have a function to decompress the mask back into its original boolean array. The expected size is 1 channel of the same dimensions of the input image. The method of compresion is compressing a numpy array using zlib then compressing tha using base64. Decompression reverses that by decompressing using base64 and then decompressing using zlib. The resulitng data is loaded into a numpy array from buffer.\n",
    "* save_visualization (Customizeable)\n",
    "  * we use opencv to apply the masks we get to each frame then stitch the new frames back together back into a video now with the masks.\n",
    "  * NOTE: this function, as is, requires FFmpeg and OpenCV."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def encode_video(video_path: str) -> str:\n",
    "    \"\"\"Encode video to base64.\"\"\"\n",
    "    try:\n",
    "        with open(video_path, \"rb\") as f:\n",
    "            return base64.b64encode(f.read()).decode(\"utf-8\")\n",
    "    except Exception as e:\n",
    "        logger.error(f\"Failed to encode video {video_path}: {str(e)}\")\n",
    "        raise\n",
    "\n",
    "\n",
    "def decompress_mask(mask_data: Union[Dict, np.ndarray]) -> Optional[np.ndarray]:\n",
    "    \"\"\"Decompress mask from zlib+base64 format.\"\"\"\n",
    "    if isinstance(mask_data, np.ndarray):\n",
    "        return mask_data.astype(bool)\n",
    "\n",
    "    if not isinstance(mask_data, dict):\n",
    "        logger.warning(f\"Invalid mask data type: {type(mask_data)}\")\n",
    "        return None\n",
    "\n",
    "    if mask_data.get(\"compression\") != \"zlib_base64\":\n",
    "        logger.warning(f\"Mask data is not in compressed format: {mask_data.keys()}\")\n",
    "        return None\n",
    "\n",
    "    counts, shape, dtype = (mask_data.get(key) for key in (\"counts\", \"shape\", \"dtype\"))\n",
    "    if not all((counts, shape, dtype)):\n",
    "        logger.warning(\n",
    "            f\"Missing required mask data fields: counts={bool(counts)}, shape={bool(shape)}, dtype={bool(dtype)}\"\n",
    "        )\n",
    "        return None\n",
    "\n",
    "    try:\n",
    "        decompressed = zlib.decompress(base64.b64decode(counts))\n",
    "        array = np.frombuffer(decompressed, dtype=np.dtype(dtype))\n",
    "        return array.reshape(shape) > 0.0\n",
    "    except Exception as e:\n",
    "        logger.error(f\"Failed to decompress mask: {e}\")\n",
    "        return None\n",
    "\n",
    "\n",
    "def convert_to_web_compatible_format(output_path: str):\n",
    "    try:\n",
    "        import subprocess\n",
    "\n",
    "        output_web_path = str(output_path).replace(\".mp4\", \"_web.mp4\")\n",
    "\n",
    "        # Use ffmpeg to convert to web-compatible format\n",
    "        cmd = [\n",
    "            \"ffmpeg\",\n",
    "            \"-i\",\n",
    "            str(output_path),\n",
    "            \"-vcodec\",\n",
    "            \"libx264\",\n",
    "            \"-acodec\",\n",
    "            \"aac\",\n",
    "            \"-y\",  # Overwrite output file if it exists\n",
    "            output_web_path,\n",
    "        ]\n",
    "\n",
    "        subprocess.run(cmd, check=True)\n",
    "\n",
    "        # Replace original file with converted file\n",
    "        import shutil\n",
    "\n",
    "        shutil.move(output_web_path, output_path)\n",
    "\n",
    "    except Exception as e:\n",
    "        logger.error(f\"Failed to convert video to web-compatible format: {e}\")\n",
    "        logger.warning(\"The video file may not play in web browsers\")\n",
    "\n",
    "\n",
    "def save_visualization(video_path, results, output_dir, debug_dir=None):\n",
    "    \"\"\"Save visualization of masks overlaid on video frames.\"\"\"\n",
    "    output_dir = Path(output_dir)\n",
    "    output_dir.mkdir(parents=True, exist_ok=True)\n",
    "\n",
    "    # Convert list results to dictionary if needed\n",
    "    if isinstance(results, list):\n",
    "        results_dict = {}\n",
    "        for frame_data in results:\n",
    "            if \"frame_idx\" in frame_data:\n",
    "                results_dict[str(frame_data[\"frame_idx\"])] = {\n",
    "                    \"masks\": frame_data.get(\"masks\", []),\n",
    "                    \"obj_ids\": frame_data.get(\"obj_ids\", []),\n",
    "                }\n",
    "        results = results_dict\n",
    "\n",
    "    # Generate colors for objects\n",
    "    all_obj_ids = set()\n",
    "    for frame_data in results.values():\n",
    "        if \"obj_ids\" in frame_data:\n",
    "            all_obj_ids.update(frame_data[\"obj_ids\"])\n",
    "    object_colors = {\n",
    "        obj_id: cv2.cvtColor(\n",
    "            np.uint8([[[i / len(all_obj_ids) * 180, 204, 255]]]), cv2.COLOR_HSV2RGB\n",
    "        )[0][0]\n",
    "        / 255.0\n",
    "        for i, obj_id in enumerate(sorted(all_obj_ids))\n",
    "    }\n",
    "\n",
    "    # Open video and get properties\n",
    "    cap = cv2.VideoCapture(str(video_path))\n",
    "    fps, width, height, total_frames = [\n",
    "        cap.get(prop)\n",
    "        for prop in [\n",
    "            cv2.CAP_PROP_FPS,\n",
    "            cv2.CAP_PROP_FRAME_WIDTH,\n",
    "            cv2.CAP_PROP_FRAME_HEIGHT,\n",
    "            cv2.CAP_PROP_FRAME_COUNT,\n",
    "        ]\n",
    "    ]\n",
    "    width, height, total_frames = map(int, [width, height, total_frames])\n",
    "\n",
    "    # Set up video writer\n",
    "    output_path = output_dir / \"visualization.mp4\"\n",
    "    fourcc = cv2.VideoWriter_fourcc(*\"mp4v\")\n",
    "    out = cv2.VideoWriter(str(output_path), fourcc, fps, (width, height))\n",
    "\n",
    "    try:\n",
    "        for frame_idx in range(total_frames):\n",
    "            ret, frame = cap.read()\n",
    "            if not ret:\n",
    "                break\n",
    "\n",
    "            frame_data = results.get(str(frame_idx), {\"masks\": [], \"obj_ids\": []})\n",
    "            masks_data = frame_data[\"masks\"]\n",
    "            obj_ids = frame_data[\"obj_ids\"]\n",
    "\n",
    "            if len(obj_ids) > 0:\n",
    "                processed_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n",
    "\n",
    "                # Decompress masks\n",
    "                if isinstance(masks_data, dict):\n",
    "                    decompressed_masks = decompress_mask(masks_data)\n",
    "                elif isinstance(masks_data, list):\n",
    "                    masks = [decompress_mask(mask_data) for mask_data in masks_data]\n",
    "                    decompressed_masks = (\n",
    "                        np.concatenate(masks, axis=0) if all(m is not None for m in masks) else None\n",
    "                    )\n",
    "\n",
    "                # Apply masks if decompression was successful\n",
    "                if decompressed_masks is not None:\n",
    "                    for i, obj_id in enumerate(obj_ids):\n",
    "                        mask = decompressed_masks[i]\n",
    "                        color = object_colors[obj_id]\n",
    "                        # Apply mask with alpha blending\n",
    "                        mask = np.squeeze(mask) > 0\n",
    "                        mask_image = np.zeros((*processed_frame.shape[:2], 4), dtype=np.float32)\n",
    "                        mask_image[mask] = [*color, 0.6]\n",
    "                        alpha = np.repeat(mask_image[..., 3:4], 3, axis=2)\n",
    "                        processed_frame = (\n",
    "                            (\n",
    "                                processed_frame.astype(np.float32) / 255.0 * (1 - alpha)\n",
    "                                + mask_image[..., :3] * alpha\n",
    "                            )\n",
    "                            * 255\n",
    "                        ).astype(np.uint8)\n",
    "\n",
    "                    processed_frame_bgr = cv2.cvtColor(processed_frame, cv2.COLOR_RGB2BGR)\n",
    "                    out.write(processed_frame_bgr)\n",
    "            else:\n",
    "                out.write(frame)\n",
    "    finally:\n",
    "        cap.release()\n",
    "        out.release()\n",
    "    convert_to_web_compatible_format(output_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sagemaker Endpoint interaction abstraction\n",
    "\n",
    "The following functions are completely optional when interacting with the Sagemaker endpoint. We abstracted repetitive code into the following functions to handle both streaming and non streaming requests. This endpoint has been built with Sticky(Stateful) Sessions in mind so operation of the endpoint is dependent on actions specified.\n",
    "\n",
    "NOTE: The operations to start and end session should be non streaming requests. When starting a new session, the header `X-Amzn-SageMaker-Session-Id` or the parameter in the `invoke_endpoint/invoke_endpoint_with_response_stream` function, in the sagemaker runtime client, needs to be `NEW_SESSION`. [Blog Post](https://aws.amazon.com/blogs/machine-learning/build-ultra-low-latency-multimodal-generative-ai-applications-using-sticky-session-routing-in-amazon/) discussing Sagemaker Sticky Sessions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def send_and_check_request(runtime_client, endpoint_name, request, session_id=None):\n",
    "    \"\"\"Send request to endpoint and check response.\"\"\"\n",
    "    request_type = request.get(\"type\", \"unknown\")\n",
    "    logger.info(f\"Sending {request_type} request to endpoint {endpoint_name}\")\n",
    "    try:\n",
    "        session_id, response = _handle_request(runtime_client, endpoint_name, request, session_id)\n",
    "        return response, session_id\n",
    "    except Exception as e:\n",
    "        _handle_error(e, request, request_type, session_id)\n",
    "        raise\n",
    "\n",
    "\n",
    "def _handle_request(runtime_client, endpoint_name, request, session_id):\n",
    "    request_type = request.get(\"type\", \"unknown\")\n",
    "\n",
    "    if request_type == \"start_session\":\n",
    "        session_id = \"NEW_SESSION\"\n",
    "    elif request_type == \"add_points\" and \"clear_old_points\" not in request:\n",
    "        request[\"clear_old_points\"] = False\n",
    "\n",
    "    if request_type in [\"start_session\", \"end_session\"]:\n",
    "        return _handle_non_streaming_request(runtime_client, endpoint_name, request, session_id)\n",
    "    else:\n",
    "        return _handle_streaming_request(runtime_client, endpoint_name, request, session_id)\n",
    "\n",
    "\n",
    "def _handle_streaming_request(runtime_client, endpoint_name, request, session_id):\n",
    "    response = runtime_client.invoke_endpoint_with_response_stream(\n",
    "        EndpointName=endpoint_name,\n",
    "        ContentType=\"application/json\",\n",
    "        Body=json.dumps(request),\n",
    "        SessionId=session_id,\n",
    "        Accept=\"application/jsonlines\",\n",
    "    )\n",
    "\n",
    "    parser = StreamParser()\n",
    "    for event in response[\"Body\"]:\n",
    "        parser.write(event)\n",
    "\n",
    "    parsed_responses = parser.get_responses()\n",
    "    logger.info(f\"Total responses received: {len(parsed_responses)}\")\n",
    "\n",
    "    headers = response[\"ResponseMetadata\"][\"HTTPHeaders\"]\n",
    "    session_id = _get_session_id(headers, request.get(\"type\"), session_id)\n",
    "\n",
    "    return session_id, parsed_responses\n",
    "\n",
    "\n",
    "def _handle_non_streaming_request(runtime_client, endpoint_name, request, session_id):\n",
    "    response = runtime_client.invoke_endpoint(\n",
    "        EndpointName=endpoint_name,\n",
    "        ContentType=\"application/json\",\n",
    "        Body=json.dumps(request),\n",
    "        SessionId=session_id,\n",
    "    )\n",
    "\n",
    "    response_body = json.loads(response[\"Body\"].read())\n",
    "    headers = response[\"ResponseMetadata\"][\"HTTPHeaders\"]\n",
    "    logger.info(f\"Response headers: {headers}\")\n",
    "\n",
    "    session_id = _get_session_id(headers, request.get(\"type\"), session_id)\n",
    "\n",
    "    return session_id, response_body\n",
    "\n",
    "\n",
    "def _get_session_id(headers, request_type, current_session_id):\n",
    "    if request_type == \"start_session\":\n",
    "        session_id = headers.get(\"x-amzn-sagemaker-new-session-id\", \"\")\n",
    "        if session_id and \"; Expires=\" in session_id:\n",
    "            session_id, expiry = session_id.split(\"; Expires=\")\n",
    "            logger.info(f\"New session created: {session_id} (expires: {expiry})\")\n",
    "    else:\n",
    "        session_id = headers.get(\"x-amzn-sagemaker-session-id\", current_session_id)\n",
    "        if headers.get(\"x-amzn-sagemaker-closed-session-id\"):\n",
    "            logger.info(f\"Session closed: {session_id}\")\n",
    "\n",
    "    if not session_id and request_type == \"start_session\":\n",
    "        raise RuntimeError(\"No session ID returned for new session request\")\n",
    "\n",
    "    return session_id\n",
    "\n",
    "\n",
    "def _handle_error(error, request, request_type, session_id):\n",
    "    logger.error(f\"Request failed: {str(error)}\")\n",
    "    logger.error(f\"Request details: type={request_type}, session_id={session_id}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### SAM2.1 Video Predictor\n",
    "\n",
    "### Session Management\n",
    "\n",
    "#### 1. Start Session (`start_session`)\n",
    "- **Purpose**: Initializes a new inference session\n",
    "- **Parameters**:\n",
    "  - `path`: Path to input image/video\n",
    "  - `input_type`: 'image' or 'video'\n",
    "  - `session_id`: Optional custom identifier\n",
    "- **Returns**:\n",
    "  - Session ID\n",
    "  - Success status\n",
    "  - Image/video dimensions\n",
    "\n",
    "#### 2. Close Session (`close_session`)\n",
    "- **Purpose**: Terminates an active session\n",
    "- **Parameter**:\n",
    "  - `session_id`: Session to close\n",
    "- **Returns**:\n",
    "  - Success status\n",
    "\n",
    "### Video Segmentation Controls\n",
    "\n",
    "#### 3. Add Points (`add_points`)\n",
    "- **Purpose**: Adds point prompts for video segmentation\n",
    "- **Parameters**:\n",
    "  - `frame_index`: Frame number\n",
    "  - `object_id`: Object tracking identifier\n",
    "  - `points`: Point coordinates\n",
    "  - `labels`: Point labels\n",
    "  - `clear_old_points`: Whether to clear existing points\n",
    "- **Returns**:\n",
    "  - Frame index\n",
    "  - Object IDs list\n",
    "  - Predicted masks\n",
    "\n",
    "#### 4. Add Box (`add_box`)\n",
    "- **Purpose**: Adds box prompt with optional points\n",
    "- **Parameters**:\n",
    "  - `frame_index`: Frame number\n",
    "  - `object_id`: Object tracking identifier\n",
    "  - `box`: Box coordinates [x1,y1,x2,y2]\n",
    "  - `points`: Optional point coordinates\n",
    "  - `labels`: Optional point labels\n",
    "- **Returns**:\n",
    "  - Frame index\n",
    "  - Object IDs list\n",
    "  - Predicted masks\n",
    "\n",
    "### State Management\n",
    "\n",
    "#### 5. Clear Points in Frame (`clear_points_in_frame`)\n",
    "- **Purpose**: Clears prompts for specific frame\n",
    "- **Parameters**:\n",
    "  - `frame_index`: Frame number\n",
    "  - `object_id`: Object tracking identifier\n",
    "- **Returns**:\n",
    "  - Success status\n",
    "\n",
    "#### 6. Clear Points in Video (`clear_points_in_video`)\n",
    "- **Purpose**: Clears all prompts in video\n",
    "- **Parameters**: None\n",
    "- **Returns**:\n",
    "  - Success status\n",
    "\n",
    "#### 7. Propagate in Video (`propagate_in_video`)\n",
    "- **Purpose**: Propagates masks through video frames\n",
    "- **Parameter**:\n",
    "  - `start_frame_index`: Starting frame number\n",
    "- **Returns**:\n",
    "  - Frame index\n",
    "  - Object IDs list\n",
    "  - Predicted masks\n",
    "\n",
    "\n",
    "**NOTE**: When adding points and boxes it is recommended to do this before propogation to get full masks. Any points added after a propogation is called will only edit the masks previously generated. The edits may not be substantial and might only alter the masks minimally. If focus of prompt needs to change it is recommended to end session and start again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class VideoPredictor:\n",
    "    def __init__(self, runtime_client, endpoint_name, video_path):\n",
    "        self.runtime_client = runtime_client\n",
    "        self.endpoint_name = endpoint_name\n",
    "        self.video_path = video_path\n",
    "\n",
    "    def save_prompt_viz(\n",
    "        self, frame_index: int, test_type: str, points=None, box=None, response=None\n",
    "    ):\n",
    "        \"\"\"Save visualization of prompts and masks on a video frame.\n",
    "\n",
    "        Args:\n",
    "            frame_index: Frame index to visualize\n",
    "            test_type: Type of test being run (e.g. \"single_object\", \"multiple_objects\")\n",
    "            points: List of point dictionaries with coordinates and labels\n",
    "            box: List of box coordinates [x1,y1,x2,y2]\n",
    "            response: Response from endpoint containing masks\n",
    "        \"\"\"\n",
    "        # Get the frame\n",
    "        cap = cv2.VideoCapture(self.video_path)\n",
    "        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_index)\n",
    "        ret, frame = cap.read()\n",
    "        cap.release()\n",
    "\n",
    "        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n",
    "\n",
    "        # Draw points if provided\n",
    "        if points:\n",
    "            for point in points:\n",
    "                x, y = map(int, point[\"coordinates\"])\n",
    "                color = (\n",
    "                    (0, 255, 0) if point[\"label\"] == 1 else (255, 0, 0)\n",
    "                )  # Green for positive, Red for negative\n",
    "                cv2.circle(frame_rgb, (x, y), 5, color, -1)  # Filled circle\n",
    "                cv2.circle(frame_rgb, (x, y), 7, color, 2)  # Border\n",
    "\n",
    "        # Draw box if provided\n",
    "        if box:\n",
    "            x1, y1, x2, y2 = map(int, box)\n",
    "            cv2.rectangle(frame_rgb, (x1, y1), (x2, y2), (255, 128, 0), 2)  # Orange box\n",
    "\n",
    "        with open(\"debug.json\", \"w+\") as f:\n",
    "            f.write(json.dumps(response))\n",
    "\n",
    "        # Draw masks if available in response\n",
    "        if response and \"masks\" in response:\n",
    "            masks = response[\"masks\"]\n",
    "            if isinstance(masks, list) and len(masks) > 0:\n",
    "                for mask_data in masks:\n",
    "                    mask = decompress_mask(mask_data)\n",
    "                    if mask is not None:\n",
    "                        # Apply semi-transparent mask\n",
    "                        mask_color = np.array([0, 0, 255])  # Blue for mask\n",
    "                        mask_overlay = np.zeros_like(frame_rgb)\n",
    "                        mask_overlay[mask] = mask_color\n",
    "                        frame_rgb = cv2.addWeighted(frame_rgb, 1, mask_overlay, 0.5, 0)\n",
    "\n",
    "        # Save in test-specific directory without timestamp\n",
    "        output_dir = os.path.join(\"outputs\", \"video_predictor\", f\"{test_type}\")\n",
    "        os.makedirs(output_dir, exist_ok=True)\n",
    "\n",
    "        # Save frame with descriptive name\n",
    "        frame_bgr = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR)\n",
    "        cv2.imwrite(\n",
    "            os.path.join(\n",
    "                output_dir, f\"prompt_frame_{frame_index:04d}_{datetime.now().timestamp()}.jpg\"\n",
    "            ),\n",
    "            frame_bgr,\n",
    "        )\n",
    "\n",
    "    def _start_session(self):\n",
    "        start_request = {\n",
    "            \"type\": \"start_session\",\n",
    "            \"input_type\": \"video\",\n",
    "            \"path\": encode_video(self.video_path),\n",
    "        }\n",
    "        _, session_id = send_and_check_request(\n",
    "            self.runtime_client, self.endpoint_name, start_request\n",
    "        )\n",
    "        return session_id\n",
    "\n",
    "    def _end_session(self, session_id):\n",
    "        close_request = {\"type\": \"close_session\", \"session_id\": session_id}\n",
    "        send_and_check_request(self.runtime_client, self.endpoint_name, close_request, session_id)\n",
    "\n",
    "    def _propagate_video(self, session_id, test_type: str, start_frame_index=0):\n",
    "        \"\"\"Propagate tracking in video and return results.\"\"\"\n",
    "        propagate_request = {\n",
    "            \"type\": \"propagate_in_video\",\n",
    "            \"session_id\": session_id,\n",
    "            \"start_frame_index\": start_frame_index,\n",
    "        }\n",
    "\n",
    "        # Time video propagation\n",
    "        prop_start_time = datetime.now()\n",
    "        response, _ = send_and_check_request(\n",
    "            self.runtime_client, self.endpoint_name, propagate_request, session_id\n",
    "        )\n",
    "\n",
    "        results, frame_count = response, len(response)\n",
    "        prop_duration = (datetime.now() - prop_start_time).total_seconds()\n",
    "        logger.info(\n",
    "            f\"Video propagation completed: {frame_count} frames in {prop_duration:.2f} seconds\"\n",
    "        )\n",
    "\n",
    "        # Time visualization saving\n",
    "        vis_start_time = datetime.now()\n",
    "        output_dir = os.path.join(\"outputs\", \"video_predictor\", f\"{test_type}\")\n",
    "        save_visualization(self.video_path, results, output_dir)\n",
    "        vis_duration = (datetime.now() - vis_start_time).total_seconds()\n",
    "        logger.info(f\"Visualization saving completed in {vis_duration:.2f} seconds\")\n",
    "\n",
    "        return results\n",
    "\n",
    "    def _add_points(\n",
    "        self, session_id, prompts, test_type: str, object_id=1, frame_index=0, clear_old_points=True\n",
    "    ):\n",
    "        \"\"\"Add points for video segmentation.\"\"\"\n",
    "        # Extract points and labels\n",
    "        points = []\n",
    "        labels = []\n",
    "        for prompt in prompts:\n",
    "            if prompt[\"type\"] == \"point\":\n",
    "                points.append(prompt[\"coordinates\"])\n",
    "                labels.append(prompt[\"label\"])\n",
    "\n",
    "        if points:\n",
    "            request = {\n",
    "                \"type\": \"add_points\",\n",
    "                \"session_id\": session_id,\n",
    "                \"frame_index\": frame_index,\n",
    "                \"object_id\": object_id,\n",
    "                \"points\": points,\n",
    "                \"labels\": labels,\n",
    "                \"clear_old_points\": clear_old_points,\n",
    "            }\n",
    "            response, _ = send_and_check_request(\n",
    "                self.runtime_client, self.endpoint_name, request, session_id\n",
    "            )\n",
    "            self.save_prompt_viz(frame_index, test_type, points=prompts, response=response)\n",
    "\n",
    "    def _add_box(self, session_id, box_prompt, points, test_type: str, object_id=1, frame_index=0):\n",
    "        \"\"\"Add box for video segmentation, optionally with points.\"\"\"\n",
    "        # Extract points and labels if provided\n",
    "        point_coords = []\n",
    "        point_labels = []\n",
    "        if points:\n",
    "            for point in points:\n",
    "                if point[\"type\"] == \"point\":\n",
    "                    point_coords.append(point[\"coordinates\"])\n",
    "                    point_labels.append(point[\"label\"])\n",
    "\n",
    "        request = {\n",
    "            \"type\": \"add_box\",\n",
    "            \"session_id\": session_id,\n",
    "            \"frame_index\": frame_index,\n",
    "            \"object_id\": object_id,\n",
    "            \"box\": box_prompt[\"coordinates\"],\n",
    "            \"points\": point_coords,\n",
    "            \"labels\": point_labels,\n",
    "        }\n",
    "        response, _ = send_and_check_request(\n",
    "            self.runtime_client, self.endpoint_name, request, session_id\n",
    "        )\n",
    "        self.save_prompt_viz(\n",
    "            frame_index, test_type, points=points, box=box_prompt[\"coordinates\"], response=response\n",
    "        )\n",
    "\n",
    "    def _add_refinement_points(\n",
    "        self, session_id, points, test_type: str, object_id=1, frame_index=0\n",
    "    ):\n",
    "        \"\"\"Add refinement points to tweak the segmentation after initial propagation.\n",
    "        These points help refine the object's segmentation without redefining what object\n",
    "        is being tracked. They can be added on any frame and will attempt to maintain consistency in the object definition.\n",
    "        \"\"\"\n",
    "        self._add_points(\n",
    "            session_id, points, test_type, object_id, frame_index, clear_old_points=False\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Video Predictor Tests\n",
    "\n",
    "Test video segmentation functionality with various tracking scenarios."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3 = boto3.client(\"s3\")\n",
    "region = boto3.Session().region_name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Download a sample image from jumpstart assets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_bucket = f\"jumpstart-cache-prod-{region}\"\n",
    "key_prefix = \"inference-notebook-assets\"\n",
    "\n",
    "\n",
    "def download_from_s3(key_filenames):\n",
    "    for key_filename in key_filenames:\n",
    "        s3.download_file(s3_bucket, f\"{key_prefix}/{key_filename}\", key_filename)\n",
    "\n",
    "\n",
    "basketball_layup_mp4 = \"basketball-layup.mp4\"\n",
    "\n",
    "# Download images and label-mapping file.\n",
    "download_from_s3(key_filenames=[basketball_layup_mp4])\n",
    "\n",
    "Video(filename=basketball_layup_mp4, width=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "os.makedirs(\"outputs/video_predictor\", exist_ok=True)\n",
    "runtime_client = boto3.client(\"sagemaker-runtime\")\n",
    "endpoint_name = predictor.endpoint_name\n",
    "video_predictor = VideoPredictor(runtime_client, endpoint_name, basketball_layup_mp4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\"\"\"Test tracking single object.\"\"\"\n",
    "\n",
    "logger.info(\"\\n=== Testing Single Object Tracking ===\")\n",
    "test_start_time = datetime.now()\n",
    "try:\n",
    "    # Start session\n",
    "    session_id = video_predictor._start_session()\n",
    "\n",
    "    # Add initial points\n",
    "    points = [\n",
    "        {\"type\": \"point\", \"coordinates\": [1478, 649], \"label\": 1},\n",
    "        {\"type\": \"point\", \"coordinates\": [1433, 689], \"label\": 0},\n",
    "    ]\n",
    "    video_predictor._add_points(session_id, points, \"single_object\")\n",
    "\n",
    "    # Propagate tracking and get results\n",
    "    results = video_predictor._propagate_video(session_id, \"single_object\")\n",
    "\n",
    "    # End session\n",
    "    video_predictor._end_session(session_id)\n",
    "    test_duration = (datetime.now() - test_start_time).total_seconds()\n",
    "    logger.info(\n",
    "        f\"Single object tracking test completed successfully in {test_duration:.2f} seconds\"\n",
    "    )\n",
    "except Exception as e:\n",
    "    logger.error(f\"Single object tracking test failed: {str(e)}\")\n",
    "    raise\n",
    "\n",
    "Video(filename=\"outputs/video_predictor/single_object/visualization.mp4\", width=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\"\"\"Test tracking multiple objects.\"\"\"\n",
    "\n",
    "logger.info(\"\\n=== Testing Multiple Object Tracking ===\")\n",
    "test_start_time = datetime.now()\n",
    "try:\n",
    "    # Start session\n",
    "    session_id = video_predictor._start_session()\n",
    "\n",
    "    # Add first object\n",
    "    points1 = [\n",
    "        {\"type\": \"point\", \"coordinates\": [1478, 649], \"label\": 1},\n",
    "        {\"type\": \"point\", \"coordinates\": [1433, 689], \"label\": 0},\n",
    "    ]\n",
    "    video_predictor._add_points(session_id, points1, \"multiple_objects\", object_id=1)\n",
    "\n",
    "    # Add second object\n",
    "    points2 = [{\"type\": \"point\", \"coordinates\": [1433, 689], \"label\": 1}]\n",
    "    video_predictor._add_points(session_id, points2, \"multiple_objects\", object_id=2)\n",
    "\n",
    "    # Propagate tracking and get results\n",
    "    results = video_predictor._propagate_video(session_id, \"multiple_objects\")\n",
    "\n",
    "    # End session\n",
    "    video_predictor._end_session(session_id)\n",
    "    test_duration = (datetime.now() - test_start_time).total_seconds()\n",
    "    logger.info(\n",
    "        f\"Multiple object tracking test completed successfully in {test_duration:.2f} seconds\"\n",
    "    )\n",
    "except Exception as e:\n",
    "    logger.error(f\"Multiple object tracking test failed: {str(e)}\")\n",
    "    raise\n",
    "\n",
    "Video(filename=\"outputs/video_predictor/multiple_objects/visualization.mp4\", width=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\"\"\"Test tracking with box prompt.\"\"\"\n",
    "\n",
    "logger.info(\"\\n=== Testing Box Tracking ===\")\n",
    "test_start_time = datetime.now()\n",
    "try:\n",
    "    # Start session\n",
    "    session_id = video_predictor._start_session()\n",
    "\n",
    "    # Add box\n",
    "    box_prompt = {\"type\": \"box\", \"coordinates\": [1392, 562, 1531, 872]}\n",
    "    points = []\n",
    "    video_predictor._add_box(session_id, box_prompt, points, \"box_prompt\")\n",
    "\n",
    "    # Propagate tracking and get results\n",
    "    results = video_predictor._propagate_video(session_id, \"box_prompt\")\n",
    "\n",
    "    # End session\n",
    "    video_predictor._end_session(session_id)\n",
    "    test_duration = (datetime.now() - test_start_time).total_seconds()\n",
    "    logger.info(f\"Box prompt tracking test completed successfully in {test_duration:.2f} seconds\")\n",
    "except Exception as e:\n",
    "    logger.error(f\"Box prompt tracking test failed: {str(e)}\")\n",
    "    raise\n",
    "\n",
    "Video(filename=\"outputs/video_predictor/box_prompt/visualization.mp4\", width=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\"\"\"Test tracking with multiple box prompt.\"\"\"\n",
    "\n",
    "logger.info(\"\\n=== Testing Multiple Box Tracking ===\")\n",
    "test_start_time = datetime.now()\n",
    "try:\n",
    "    # Start session\n",
    "    session_id = video_predictor._start_session()\n",
    "\n",
    "    # Add box\n",
    "    box_prompt = {\"type\": \"box\", \"coordinates\": [1392, 562, 1531, 872]}\n",
    "    points = []\n",
    "    video_predictor._add_box(session_id, box_prompt, points, \"multiple_box_prompt\", object_id=1)\n",
    "\n",
    "    # Add box\n",
    "    box_prompt = {\"type\": \"box\", \"coordinates\": [1195, 216, 1485, 440]}\n",
    "    points = []\n",
    "    video_predictor._add_box(session_id, box_prompt, points, \"multiple_box_prompt\", object_id=2)\n",
    "\n",
    "    # Propagate tracking and get results\n",
    "    results = video_predictor._propagate_video(session_id, \"multiple_box_prompt\")\n",
    "\n",
    "    # End session\n",
    "    video_predictor._end_session(session_id)\n",
    "    test_duration = (datetime.now() - test_start_time).total_seconds()\n",
    "    logger.info(\n",
    "        f\"Multiple Box prompt tracking test completed successfully in {test_duration:.2f} seconds\"\n",
    "    )\n",
    "except Exception as e:\n",
    "    logger.error(f\"Multiple Box prompt tracking test failed: {str(e)}\")\n",
    "    raise\n",
    "\n",
    "Video(filename=\"outputs/video_predictor/multiple_box_prompt/visualization.mp4\", width=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\"\"\"Test tracking with combined box and point prompts.\"\"\"\n",
    "\n",
    "logger.info(\"\\n=== Testing Combined Box and Point Tracking ===\")\n",
    "test_start_time = datetime.now()\n",
    "try:\n",
    "    # Start session\n",
    "    session_id = video_predictor._start_session()\n",
    "\n",
    "    # Add box with points\n",
    "    box_prompt = {\"type\": \"box\", \"coordinates\": [1392, 562, 1531, 872]}\n",
    "    points = [{\"type\": \"point\", \"coordinates\": [1433, 689], \"label\": 0}]\n",
    "    video_predictor._add_box(session_id, box_prompt, points, \"combined_prompts\")\n",
    "\n",
    "    # Propagate tracking and get results\n",
    "    results = video_predictor._propagate_video(session_id, \"combined_prompts\")\n",
    "\n",
    "    # End session\n",
    "    video_predictor._end_session(session_id)\n",
    "    test_duration = (datetime.now() - test_start_time).total_seconds()\n",
    "    logger.info(\n",
    "        f\"Combined prompts tracking test completed successfully in {test_duration:.2f} seconds\"\n",
    "    )\n",
    "except Exception as e:\n",
    "    logger.error(f\"Combined prompts tracking test failed: {str(e)}\")\n",
    "    raise\n",
    "\n",
    "Video(filename=\"outputs/video_predictor/combined_prompts/visualization.mp4\", width=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\"\"\"Test tracking with prompt refinement on different frames.\"\"\"\n",
    "\n",
    "logger.info(\"\\n=== Testing Prompt Refinement in Video ===\")\n",
    "test_start_time = datetime.now()\n",
    "try:\n",
    "    # Start session\n",
    "    session_id = video_predictor._start_session()\n",
    "\n",
    "    # Add initial points on frame 0\n",
    "    points = [\n",
    "        {\"type\": \"point\", \"coordinates\": [1478, 649], \"label\": 0},\n",
    "        {\"type\": \"point\", \"coordinates\": [1433, 689], \"label\": 1},\n",
    "    ]\n",
    "    video_predictor._add_points(session_id, points, \"prompt_refinement\")\n",
    "\n",
    "    # Propagate tracking and get results\n",
    "    results = video_predictor._propagate_video(session_id, \"prompt_refinement_initial\")\n",
    "\n",
    "    # Add refinement point on frame 114\n",
    "    refine_points = [\n",
    "        {\"type\": \"point\", \"coordinates\": [940, 155], \"label\": 0},\n",
    "    ]\n",
    "    video_predictor._add_points(\n",
    "        session_id, refine_points, \"prompt_refinement\", frame_index=114, clear_old_points=False\n",
    "    )\n",
    "\n",
    "    # Propagate tracking and get results\n",
    "    results = video_predictor._propagate_video(session_id, \"prompt_refinement_final\")\n",
    "\n",
    "    # End session\n",
    "    video_predictor._end_session(session_id)\n",
    "    test_duration = (datetime.now() - test_start_time).total_seconds()\n",
    "    logger.info(f\"Prompt refinement test completed successfully in {test_duration:.2f} seconds\")\n",
    "except Exception as e:\n",
    "    logger.error(f\"Prompt refinement test failed: {str(e)}\")\n",
    "    raise\n",
    "\n",
    "Video(filename=\"outputs/video_predictor/prompt_refinement_initial/visualization.mp4\", width=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictor.delete_predictor()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c4998b86",
   "metadata": {},
   "source": [
    "## Notebook CI Test Results\n",
    "\n",
    "This notebook was tested in multiple regions. The test results are as follows, except for us-west-2 which is shown at the top of the notebook.\n",
    "\n",
    "\n",
    "![This us-east-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-east-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This us-east-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-east-2/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This us-west-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-west-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This ca-central-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ca-central-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This sa-east-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/sa-east-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This eu-west-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This eu-west-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-2/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This eu-west-3 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-3/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This eu-central-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-central-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This eu-north-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-north-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This ap-southeast-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-southeast-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This ap-southeast-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-southeast-2/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This ap-northeast-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-northeast-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This ap-northeast-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-northeast-2/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n",
    "![This ap-south-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-south-1/generative_ai|sm-jumpstart_foundation_sam_2_1_video_segmentation.ipynb)\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}