concat_buv_doc_oct_2023.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyPWNn5s5FtwbB065Ta3vhnh",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/wildlifeai/spyfish_analysis/blob/main/concat_buv_doc_oct_2023.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Requirements"
      ],
      "metadata": {
        "id": "i_qqwckw8DDE"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "DJEH2wZ9z1_v"
      },
      "outputs": [],
      "source": [
        "!pip install boto3\n",
        "# base imports\n",
        "import os\n",
        "import pandas as pd\n",
        "import getpass\n",
        "import gdown\n",
        "import zipfile\n",
        "import boto3\n",
        "import logging\n",
        "import sys\n",
        "import urllib\n",
        "import subprocess\n",
        "import ipywidgets as widgets\n",
        "from tqdm import tqdm\n",
        "from pathlib import Path\n",
        "import datetime\n",
        "\n",
        "\n",
        "# widget imports\n",
        "from IPython.display import display\n",
        "from ipywidgets import Layout\n",
        "import ipywidgets as widgets\n",
        "\n",
        "# Logging\n",
        "logging.basicConfig()\n",
        "logging.getLogger().setLevel(logging.INFO)\n",
        "\n",
        "def aws_credentials():\n",
        "    # Save your access key for the s3 bucket.\n",
        "    aws_access_key_id = getpass.getpass(\"Enter the key id for the aws server\")\n",
        "    aws_secret_access_key = getpass.getpass(\n",
        "        \"Enter the secret access key for the aws server\"\n",
        "    )\n",
        "\n",
        "    return aws_access_key_id, aws_secret_access_key\n",
        "\n",
        "\n",
        "def connect_s3(aws_access_key_id: str, aws_secret_access_key: str):\n",
        "    # Connect to the s3 bucket\n",
        "    client = boto3.client(\n",
        "        \"s3\",\n",
        "        aws_access_key_id=aws_access_key_id,\n",
        "        aws_secret_access_key=aws_secret_access_key,\n",
        "    )\n",
        "    return client\n",
        "\n",
        "\n",
        "def get_aws_client():\n",
        "    # Set aws account credentials\n",
        "    aws_access_key_id, aws_secret_access_key = os.getenv(\"SPY_KEY\"), os.getenv(\n",
        "        \"SPY_SECRET\"\n",
        "    )\n",
        "    if aws_access_key_id is None or aws_secret_access_key is None:\n",
        "        aws_access_key_id, aws_secret_access_key = aws_credentials()\n",
        "\n",
        "    # Connect to S3\n",
        "    client = connect_s3(aws_access_key_id, aws_secret_access_key)\n",
        "\n",
        "    return client\n",
        "\n",
        "\n",
        "def get_matching_s3_objects(\n",
        "    client: boto3.client, bucket: str, prefix: str = \"\", suffix: str = \"\"\n",
        "):\n",
        "    \"\"\"\n",
        "    ## Code modified from alexwlchan (https://alexwlchan.net/2019/07/listing-s3-keys/)\n",
        "    Generate objects in an S3 bucket.\n",
        "\n",
        "    :param client: S3 client.\n",
        "    :param bucket: Name of the S3 bucket.\n",
        "    :param prefix: Only fetch objects whose key starts with\n",
        "        this prefix (optional).\n",
        "    :param suffix: Only fetch objects whose keys end with\n",
        "        this suffix (optional).\n",
        "    \"\"\"\n",
        "\n",
        "    paginator = client.get_paginator(\"list_objects_v2\")\n",
        "\n",
        "    kwargs = {\"Bucket\": bucket}\n",
        "\n",
        "    # We can pass the prefix directly to the S3 API.  If the user has passed\n",
        "    # a tuple or list of prefixes, we go through them one by one.\n",
        "    if isinstance(prefix, str):\n",
        "        prefixes = (prefix,)\n",
        "    else:\n",
        "        prefixes = prefix\n",
        "\n",
        "    for key_prefix in prefixes:\n",
        "        kwargs[\"Prefix\"] = key_prefix\n",
        "\n",
        "        for page in paginator.paginate(**kwargs):\n",
        "            try:\n",
        "                contents = page[\"Contents\"]\n",
        "            except KeyError:\n",
        "                break\n",
        "\n",
        "            for obj in contents:\n",
        "                key = obj[\"Key\"]\n",
        "                if key.endswith(suffix):\n",
        "                    yield obj\n",
        "\n",
        "\n",
        "def get_matching_s3_keys(\n",
        "    client: boto3.client, bucket: str, prefix: str = \"\", suffix: str = \"\"\n",
        "):\n",
        "    \"\"\"\n",
        "    ## Code from alexwlchan (https://alexwlchan.net/2019/07/listing-s3-keys/)\n",
        "    Generate the keys in an S3 bucket.\n",
        "\n",
        "    :param client: S3 client.\n",
        "    :param bucket: Name of the S3 bucket.\n",
        "    :param prefix: Only fetch keys that start with this prefix (optional).\n",
        "    :param suffix: Only fetch keys that end with this suffix (optional).\n",
        "    return a list of the matching objects\n",
        "    \"\"\"\n",
        "\n",
        "    # Select the relevant bucket\n",
        "    s3_keys = [\n",
        "        obj[\"Key\"] for obj in get_matching_s3_objects(client, bucket, prefix, suffix)\n",
        "    ]\n",
        "\n",
        "    return s3_keys\n",
        "\n",
        "def download_object_from_s3(\n",
        "    client: boto3.client,\n",
        "    *,\n",
        "    bucket: str,\n",
        "    key: str,\n",
        "    version_id: str = None,\n",
        "    filename: str,\n",
        "):\n",
        "    \"\"\"\n",
        "    Download an object from S3 with a progress bar.\n",
        "\n",
        "    From https://alexwlchan.net/2021/04/s3-progress-bars/\n",
        "    \"\"\"\n",
        "\n",
        "    # First get the size, so we know what tqdm is counting up to.\n",
        "    # Theoretically the size could change between this HeadObject and starting\n",
        "    # to download the file, but this would only affect the progress bar.\n",
        "    kwargs = {\"Bucket\": bucket, \"Key\": key}\n",
        "\n",
        "    if version_id is not None:\n",
        "        kwargs[\"VersionId\"] = version_id\n",
        "\n",
        "    object_size = client.head_object(**kwargs)[\"ContentLength\"]\n",
        "\n",
        "    if version_id is not None:\n",
        "        ExtraArgs = {\"VersionId\": version_id}\n",
        "    else:\n",
        "        ExtraArgs = None\n",
        "\n",
        "    with tqdm(\n",
        "        total=object_size,\n",
        "        unit=\"B\",\n",
        "        unit_scale=True,\n",
        "        desc=filename,\n",
        "        position=0,\n",
        "        leave=True,\n",
        "    ) as pbar:\n",
        "        client.download_file(\n",
        "            Bucket=bucket,\n",
        "            Key=key,\n",
        "            ExtraArgs=ExtraArgs,\n",
        "            Filename=filename,\n",
        "            Callback=lambda bytes_transferred: pbar.update(bytes_transferred),\n",
        "        )\n",
        "\n",
        "def upload_file_to_s3(client: boto3.client, *, bucket: str, key: str, filename: str):\n",
        "    \"\"\"\n",
        "    > Upload a file to S3, and show a progress bar if the file is large enough\n",
        "    :param client: The boto3 client to use\n",
        "    :param bucket: The name of the bucket to upload to\n",
        "    :param key: The name of the file in S3\n",
        "    :param filename: The name of the file to upload\n",
        "    \"\"\"\n",
        "\n",
        "    # Get the size of the file to upload\n",
        "    file_size = os.stat(filename).st_size\n",
        "\n",
        "    # Prevent issues with small files (<1MB) and tqdm\n",
        "    if file_size > 1000000:\n",
        "        with tqdm(\n",
        "            total=file_size,\n",
        "            unit=\"B\",\n",
        "            unit_scale=True,\n",
        "            desc=filename,\n",
        "            position=0,\n",
        "            leave=True,\n",
        "        ) as pbar:\n",
        "            client.upload_file(\n",
        "                Filename=filename,\n",
        "                Bucket=bucket,\n",
        "                Key=key,\n",
        "                Callback=lambda bytes_transferred: pbar.update(bytes_transferred),\n",
        "            )\n",
        "    else:\n",
        "        client.upload_file(\n",
        "            Filename=filename,\n",
        "            Bucket=bucket,\n",
        "            Key=key,\n",
        "        )\n",
        "\n",
        "def delete_file_from_s3(client: boto3.client, *, bucket: str, key: str):\n",
        "    \"\"\"\n",
        "    > Delete a file from S3.\n",
        "\n",
        "    :param client: boto3.client - the client object that you created in the previous step\n",
        "    :type client: boto3.client\n",
        "    :param bucket: The name of the bucket that contains the object to delete\n",
        "    :type bucket: str\n",
        "    :param key: The name of the file\n",
        "    :type key: str\n",
        "    \"\"\"\n",
        "    client.delete_object(Bucket=bucket, Key=key)\n",
        "\n",
        "def get_movie_extensions():\n",
        "    # Specify the formats of the movies to select\n",
        "    return tuple([\"wmv\", \"mpg\", \"mov\", \"avi\", \"mp4\", \"MOV\", \"MP4\"])\n",
        "\n",
        "def check_movies_from_server(client):\n",
        "    \"\"\"\n",
        "    It takes in a dataframe with movies information and a dictionary with the database information, and\n",
        "    returns two dataframes: one with the movies that are missing from the server, and one with the\n",
        "    movies that are missing from the csv\n",
        "\n",
        "    :param db_info_dict: a dictionary with the following keys:\n",
        "    :param project: the project object\n",
        "    \"\"\"\n",
        "    # Download movies csv\n",
        "    download_object_from_s3(\n",
        "            client,\n",
        "            bucket=\"marine-buv\",\n",
        "            key=\"init_db_doc_buv/movies_buv_doc.csv\",\n",
        "            filename=\"movies_buv_doc.csv\"\n",
        "        )\n",
        "\n",
        "    # Load the csv with movies information\n",
        "    movies_df = pd.read_csv(\"movies_buv_doc.csv\")\n",
        "\n",
        "    # Get a dataframe of all movies from AWS\n",
        "    movies_s3_pd = get_matching_s3_keys(\n",
        "        client = client,\n",
        "        bucket = \"marine-buv\",\n",
        "        suffix=get_movie_extensions(),\n",
        "    )\n",
        "\n",
        "    # Calling DataFrame constructor on list\n",
        "    movies_s3_pd = pd.DataFrame(movies_s3_pd, columns =['Key'])\n",
        "    # Specify the key of the movies (path in S3 of the object)\n",
        "    movies_s3_pd[\"filename\"] = movies_s3_pd.Key.str.split(\"/\").str[-1]\n",
        "\n",
        "    # Create a column with the deployment folder of each movie\n",
        "    movies_s3_pd[\"deployment_folder\"] = (\n",
        "        movies_s3_pd.Key.str.split(\"/\").str[:2].str.join(\"/\")\n",
        "    )\n",
        "\n",
        "    # Missing info for files in the \"buv-zooniverse-uploads\"\n",
        "    missing_info = movies_df.merge(\n",
        "        movies_s3_pd, on=[\"filename\"], how=\"outer\", indicator=True\n",
        "    )\n",
        "\n",
        "    # Find out files missing from the Server\n",
        "    missing_from_server = missing_info[missing_info[\"_merge\"] == \"left_only\"]\n",
        "\n",
        "    logging.info(f\"There are {len(missing_from_server.index)} movies missing\")\n",
        "\n",
        "    # Find out files missing from the csv\n",
        "    missing_from_csv = missing_info[missing_info[\"_merge\"] == \"right_only\"].reset_index(\n",
        "        drop=True\n",
        "    )\n",
        "\n",
        "    logging.info(\n",
        "        f\"There are {len(missing_from_csv.index)} movies missing from movies.csv\"\n",
        "    )\n",
        "\n",
        "    return missing_from_server, missing_from_csv\n",
        "\n",
        "def select_deployment(missing_from_csv: pd.DataFrame):\n",
        "    \"\"\"\n",
        "    > This function takes a dataframe of missing files and returns a widget that allows the user to\n",
        "    select the deployment of interest\n",
        "\n",
        "    :param missing_from_csv: a dataframe of the files that are in the data folder but not in the csv file\n",
        "    :return: A widget object\n",
        "    \"\"\"\n",
        "    if missing_from_csv.shape[0] > 0:\n",
        "        # Widget to select the deployment of interest\n",
        "        deployment_widget = widgets.SelectMultiple(\n",
        "            options=sorted(missing_from_csv.deployment_folder.unique()),\n",
        "            description=\"New deployment:\",\n",
        "            disabled=False,\n",
        "            rows=10,\n",
        "            layout=Layout(width=\"80%\"),\n",
        "            style={\"description_width\": \"initial\"},\n",
        "        )\n",
        "        display(deployment_widget)\n",
        "\n",
        "        return deployment_widget\n",
        "\n",
        "def select_eventdate():\n",
        "    \"\"\"\n",
        "    > This function creates a date picker widget that allows the user to select a date.\n",
        "    The function is called `select_eventdate()` and it returns a date picker widget.\n",
        "    :return: The date widget\n",
        "    \"\"\"\n",
        "    # Select the date\n",
        "    date_widget = widgets.DatePicker(\n",
        "        description=\"Date of deployment:\",\n",
        "        value=datetime.date.today(),\n",
        "        disabled=False,\n",
        "        layout=Layout(width=\"50%\"),\n",
        "        style={\"description_width\": \"initial\"},\n",
        "    )\n",
        "    display(date_widget)\n",
        "\n",
        "    return date_widget\n",
        "\n",
        "def concatenate_videos(movie_list, output_file):\n",
        "    # Write relative paths to a temporary text file\n",
        "    temp_file_path = os.path.abspath(\"temp_file.txt\")\n",
        "    with open(temp_file_path, \"w\") as textfile:\n",
        "        for movie_i in movie_list:\n",
        "            textfile.write(f\"file '{movie_i}'\\n\")\n",
        "\n",
        "    textfile.close()\n",
        "\n",
        "    # Concatenate the videos\n",
        "    try:\n",
        "        result = subprocess.run(\n",
        "            [\n",
        "                \"ffmpeg\",\n",
        "                \"-f\",\n",
        "                \"concat\",\n",
        "                \"-safe\",\n",
        "                \"0\",\n",
        "                \"-i\",\n",
        "                temp_file_path,\n",
        "                \"-c\",\n",
        "                \"copy\",\n",
        "                output_file,\n",
        "            ],\n",
        "            check=True,\n",
        "            capture_output=True,\n",
        "            text=True\n",
        "        )\n",
        "\n",
        "    except subprocess.CalledProcessError as e:\n",
        "        print(f\"Error during concatenation: {e}\")\n",
        "        print(e.stderr)\n",
        "\n",
        "    # Remove the temporary text file\n",
        "    os.remove(temp_file_path)\n",
        "\n",
        "\n",
        "def update_new_deployments(\n",
        "    deployment_selected, event_date: widgets.Widget, delete_go_pro_s3=False, test_concatenation=False\n",
        "):\n",
        "    for deployment_i in deployment_selected.value:\n",
        "        logging.info(\n",
        "            f\"Starting to concatenate {deployment_i} out of {len(deployment_selected.value)} deployments selected\"\n",
        "        )\n",
        "\n",
        "        # Get a dataframe of movies from the deployment\n",
        "        movies_s3_pd = get_matching_s3_keys(\n",
        "            client = client,\n",
        "            bucket = \"marine-buv\",\n",
        "            prefix=deployment_i,\n",
        "            suffix=get_movie_extensions(),\n",
        "        )\n",
        "\n",
        "        # Calling DataFrame constructor on list\n",
        "        movies_s3_pd = pd.DataFrame(movies_s3_pd, columns =['Key'])\n",
        "\n",
        "        # Create a list of the list of movies inside the deployment selected\n",
        "        movie_files_server = movies_s3_pd.Key.unique().tolist()\n",
        "\n",
        "        if len(movie_files_server) < 2:\n",
        "            logging.info(\n",
        "              f\"Deployment {deployment_i} will not be concatenated because it only has {movies_s3_pd.Key.unique()}\"\n",
        "            )\n",
        "        else:\n",
        "            # Concatenate the files if multiple\n",
        "            logging.info(f\"The files {movie_files_server} will be concatenated\")\n",
        "\n",
        "            # Start text file and list to keep track of the videos to concatenate\n",
        "            movie_list = []\n",
        "\n",
        "            for movie_i in sorted(movie_files_server):\n",
        "                # Specify the temporary output of the go pro file\n",
        "                movie_i_output = movie_i.split(\"/\")[-1]\n",
        "\n",
        "                # Download the files from the S3 bucket\n",
        "                if not os.path.exists(movie_i_output):\n",
        "                    download_object_from_s3(\n",
        "                        client = client,\n",
        "                        bucket = \"marine-buv\",\n",
        "                        key=movie_i,\n",
        "                        filename=movie_i_output,\n",
        "                    )\n",
        "\n",
        "                # Keep track of the videos to concatenate\n",
        "                movie_list.append(movie_i_output)\n",
        "\n",
        "            # Save eventdate as str\n",
        "            EventDate_str = event_date.value.strftime(\"%d_%m_%Y\")\n",
        "\n",
        "            # Specify the name of the concatenated video\n",
        "            filename = deployment_i.split(\"/\")[-1] + \"_\" + EventDate_str + \".mp4\"\n",
        "\n",
        "            # Concatenate the files\n",
        "            if not os.path.exists(filename):\n",
        "                logging.info(f\"Concatenating {filename}\")\n",
        "\n",
        "                # Concatenate the videos\n",
        "                concatenate_videos(movie_list, filename)\n",
        "\n",
        "            if test_concatenation:\n",
        "                logging.info(f\"{filename} concatenated but not uploaded to the S3 bucket\")\n",
        "\n",
        "            else:\n",
        "                # Upload the concatenated video to the S3\n",
        "                upload_file_to_s3(\n",
        "                    client = client,\n",
        "                    bucket = \"marine-buv\",\n",
        "                    key=deployment_i + \"/\" + filename,\n",
        "                    filename=filename,\n",
        "                )\n",
        "\n",
        "                logging.info(f\"{filename} successfully uploaded to {deployment_i}\")\n",
        "\n",
        "                # Delete the raw videos downloaded from the S3 bucket\n",
        "                for f in movie_list:\n",
        "                    os.remove(f)\n",
        "\n",
        "                # Delete the concat video\n",
        "                os.remove(filename)\n",
        "\n",
        "                if delete_go_pro_s3:\n",
        "                    # Delete the movies from the S3 bucket\n",
        "                    for movie_i in sorted(movie_files_server):\n",
        "                        delete_file_from_s3(\n",
        "                            client=client,\n",
        "                            bucket=\"marine-buv\",\n",
        "                            key=movie_i,\n",
        "                        )"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Connect to s3"
      ],
      "metadata": {
        "id": "7O7I2szT73gG"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Your acess key for the s3 bucket.\n",
        "aws_access_key_id = getpass.getpass('Enter the key id for the aws server')\n",
        "aws_secret_access_key = getpass.getpass('Enter the secret access key for the aws server')"
      ],
      "metadata": {
        "id": "0pRSMF3v1bpd"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Connect to the s3 bucket\n",
        "client = boto3.client('s3',\n",
        "                      aws_access_key_id = aws_access_key_id,\n",
        "                      aws_secret_access_key = aws_secret_access_key)"
      ],
      "metadata": {
        "id": "wFytTIQL0_rO"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Get info from go pro movies"
      ],
      "metadata": {
        "id": "wbJb13_D1AhN"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "missing_from_server, missing_from_csv = check_movies_from_server(\n",
        "    client\n",
        ")"
      ],
      "metadata": {
        "id": "PIqe28PQ072M"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Select deployments recorded on the same date"
      ],
      "metadata": {
        "id": "CnWiQgri1DBJ"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "deployment_selected = select_deployment(missing_from_csv)"
      ],
      "metadata": {
        "id": "6h3FZv3W1GmY"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Select date of recording"
      ],
      "metadata": {
        "id": "HRpXGNXL1KZL"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "event_date = select_eventdate()"
      ],
      "metadata": {
        "id": "jSRogso51JJs"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Concatenate the movies"
      ],
      "metadata": {
        "id": "XlAOj7kl1ONU"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "update_new_deployments(deployment_selected, event_date)"
      ],
      "metadata": {
        "id": "cKwknS5g1NbB"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}