03-Data Statistics Gen.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "ab165df6-56d5-4ba3-983b-3f540717829e",
   "metadata": {},
   "source": [
    "# 03 - Data Statistics Generation\n",
    "\n",
    "This notebook creates a component that calculates the image data statistics, which will be used for input data validation at later stage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "38852a2c-1efe-4a10-8208-3dc9dc37f32c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "import logging\n",
    "import kfp\n",
    "from google.cloud import bigquery, storage\n",
    "from google.cloud import aiplatform as vertex_ai\n",
    "from google_cloud_pipeline_components.experimental.custom_job import utils\n",
    "from kfp.v2 import compiler, dsl\n",
    "from kfp.v2.dsl import component\n",
    "from typing import NamedTuple\n",
    "from kfp.v2.dsl import (Artifact, Dataset, Input, InputPath, Model, Output, Metrics,\n",
    "                        OutputPath, component)\n",
    "\n",
    "from google_cloud_pipeline_components.experimental.custom_job import utils"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "southern-raising",
   "metadata": {},
   "outputs": [],
   "source": [
    "logging.basicConfig(level=logging.INFO)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "nearby-anderson",
   "metadata": {},
   "source": [
    "## Load Params and Resource Config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "abandoned-equilibrium",
   "metadata": {},
   "outputs": [],
   "source": [
    "from config.gcp_resource import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "continued-scanner",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Updated property [core/project].\n"
     ]
    }
   ],
   "source": [
    "if PROJECT_ID == \"\" or PROJECT_ID is None or PROJECT_ID == \"[your-project-id]\":\n",
    "    # Get your GCP project id from gcloud\n",
    "    shell_output = !gcloud config list --format 'value(core.project)' 2>/dev/null\n",
    "    PROJECT_ID = shell_output[0]\n",
    "    \n",
    "if SERVICE_ACCOUNT == \"\" or SERVICE_ACCOUNT is None or SERVICE_ACCOUNT == \"[your-service-account]\":\n",
    "    # Get your GCP project id from gcloud\n",
    "    shell_output = !gcloud config list --format 'value(core.account)' 2>/dev/null\n",
    "    SERVICE_ACCOUNT = shell_output[0]\n",
    "    \n",
    "if GCS_BUCKET == \"\" or GCS_BUCKET is None or GCS_BUCKET == \"[your-bucket-name]\":\n",
    "    # Get your bucket name to GCP projet id\n",
    "    GCS_BUCKET = PROJECT_ID\n",
    "    # Try to create the bucket if it doesn'exists\n",
    "    ! gsutil mb -l $REGION gs://$BUCKET\n",
    "    print(\"\")\n",
    "    \n",
    "!gcloud config set project {PROJECT_ID}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "handmade-utility",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Train machine type n1-standard-4\n",
      "Deploy machine type n1-standard-4\n",
      "Deployment: us-docker.pkg.dev/vertex-ai/prediction/tf2-cpu.2-8:latest\n",
      "PIPELINE_ROOT: gs://mle_airbus_dataset/airbusmlepipeline/pipeline_root\n",
      "MODULE_ROOT: gs://mle_airbus_dataset/airbusmlepipeline/pipeline_module\n",
      "DATA_ROOT: gs://mle_airbus_dataset/airbusmlepipeline/data\n",
      "SERVING_MODEL_DIR: gs://mle_airbus_dataset/airbusmlepipeline/serving_model\n"
     ]
    }
   ],
   "source": [
    "print(\"Train machine type\", TRAIN_COMPUTE)\n",
    "print(\"Deploy machine type\", DEPLOY_COMPUTE)\n",
    "print(\"Deployment:\", DEPLOY_IMAGE)\n",
    "print('PIPELINE_ROOT: {}'.format(PIPELINE_ROOT))\n",
    "print('MODULE_ROOT: {}'.format(MODULE_ROOT))\n",
    "print('DATA_ROOT: {}'.format(DATA_ROOT))\n",
    "print('SERVING_MODEL_DIR: {}'.format(SERVING_MODEL_DIR))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "broad-momentum",
   "metadata": {},
   "source": [
    "## Data Statistics Schema Generation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "initial-tunnel",
   "metadata": {},
   "outputs": [],
   "source": [
    "@component(packages_to_install=[\"google-cloud-storage\",\"opencv-python-headless\",\"pandas\",\"pyarrow\",\"fsspec\",\"gcsfs\"],\n",
    "           base_image='python:3.9',\n",
    "           output_component_file=\"./build/gen_train_hist_component.yaml\")\n",
    "def gen_train_hist_component(\n",
    "    project_dict: dict\n",
    "    ) -> NamedTuple(\n",
    "    \"Outputs\",\n",
    "    [\n",
    "        (\"train_hist_fpath\", str),  # Return path to histogram of training data.\n",
    "        (\"train_threshold_fpath\", str),  # Return path to threshold value.\n",
    "    ],\n",
    "    ):\n",
    "    \n",
    "    import cv2\n",
    "    import urllib\n",
    "    import numpy as np\n",
    "    import pandas as pd\n",
    "    from google.cloud import storage\n",
    "    \n",
    "        \n",
    "    PROJECT_ID = project_dict['PROJECT_ID']\n",
    "    GCS_BUCKET = project_dict['GCS_BUCKET']\n",
    "    GCS_TRAIN_IMAGES=f\"gs://{GCS_BUCKET}/train_v2/\"\n",
    "    \n",
    "    # read the parquet files\n",
    "    train_data = pd.read_parquet(f\"gs://{GCS_BUCKET}/train.parquet\")\n",
    "    test_data = pd.read_parquet(f\"gs://{GCS_BUCKET}/test.parquet\")\n",
    "    \n",
    "    # load training images to calculate histogram\n",
    "    gcs_url = f\"https://storage.googleapis.com/{GCS_TRAIN_IMAGES.replace('gs://','')}\"\n",
    "\n",
    "    train_images = []\n",
    "    test_images = []\n",
    "\n",
    "    for image_id in train_data['ImageId']:\n",
    "        resp = urllib.request.urlopen(f'{gcs_url}{image_id}')\n",
    "        image = np.asarray(bytearray(resp.read()), dtype=\"uint8\")\n",
    "        image = cv2.imdecode(image, cv2.IMREAD_COLOR)\n",
    "        train_images.append(image)\n",
    "        \n",
    "    for image_id in test_data['ImageId']:\n",
    "        resp = urllib.request.urlopen(f'{gcs_url}{image_id}')\n",
    "        image = np.asarray(bytearray(resp.read()), dtype=\"uint8\")\n",
    "        image = cv2.imdecode(image, cv2.IMREAD_COLOR)\n",
    "        test_images.append(image)\n",
    "    \n",
    "    channels = [0, 1, 2]\n",
    "    hist_size = [256] * 3\n",
    "    hist_ranges = [0, 256] * 3\n",
    "\n",
    "    # compute the image histograms for training data\n",
    "    train_hist = cv2.calcHist(train_images,\n",
    "                              channels,\n",
    "                              None,\n",
    "                              hist_size,\n",
    "                              hist_ranges,\n",
    "                              accumulate = True)\n",
    "    cv2.normalize(train_hist, train_hist, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX)\n",
    "        \n",
    "    # compute the image histograms for training data\n",
    "    test_hist = cv2.calcHist(test_images,\n",
    "                             channels,\n",
    "                             None,\n",
    "                             hist_size,\n",
    "                             hist_ranges,\n",
    "                             accumulate = True)\n",
    "    cv2.normalize(test_hist, test_hist, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX)\n",
    "    \n",
    "    # use correlation method for comparison\n",
    "    threshold = cv2.compareHist(train_hist, test_hist, 0)\n",
    "    \n",
    "    # reshaping the array from 3D matrice to 2D matrice.\n",
    "    arrReshaped = train_hist.reshape(train_hist.shape[0], -1)\n",
    "    # saving reshaped array to file.\n",
    "    np.savetxt('train_hist.csv', arrReshaped)\n",
    "    \n",
    "    with open('train_threshold.txt','w') as f:\n",
    "        f.write(f'{threshold}')\n",
    "        \n",
    "    # move the files to GCS\n",
    "    bucket = storage.Client().bucket(GCS_BUCKET)\n",
    "    \n",
    "    blob = bucket.blob('train_hist.csv')\n",
    "    blob.upload_from_filename('train_hist.csv')\n",
    "    blob = bucket.blob('train_threshold.txt')\n",
    "    blob.upload_from_filename('train_threshold.txt')\n",
    "    \n",
    "    return f\"gs://{GCS_BUCKET}/train_hist.csv\", f\"gs://{GCS_BUCKET}/train_threshold.txt\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9bdb8fcf-e13f-4f4b-b740-54a927043057",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "environment": {
   "kernel": "python3",
   "name": "tf2-gpu.2-8.m93",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/tf2-gpu.2-8:m93"
  },
  "kernelspec": {
   "display_name": "Python 3",
   "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.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}