battery_life_analyser.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e1ada9cb-2cb8-4c0c-841e-dbacb2b218d3",
   "metadata": {},
   "source": [
    "## Library Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "2a810962-53d3-4fec-8b38-fa6306bda77e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "OPEN_CV VER:\t 3.2.0\n",
      "NUMPY VER:\t 1.16.2\n",
      "TESSERECT VER:\t 0.3.9\n"
     ]
    }
   ],
   "source": [
    "%matplotlib inline\n",
    "%gui asyncio\n",
    "\n",
    "\n",
    "# PI Camera Libs\n",
    "import picamera\n",
    "from picamera.array import PiRGBArray\n",
    "from picamera import PiCamera\n",
    "\n",
    "# Servo Control libs\n",
    "from adafruit_servokit import ServoKit\n",
    "\n",
    "# Opnen CV libs (For OCR)\n",
    "from PIL import Image\n",
    "import cv2\n",
    "import numpy\n",
    "import matplotlib.pyplot as plt\n",
    "import pytesseract\n",
    "\n",
    "# Google cloud visison related\n",
    "from google.oauth2 import service_account\n",
    "from google.cloud import vision\n",
    "\n",
    "# import the necessary packages\n",
    "# from imutils.perspective import four_point_transform\n",
    "# from imutils import contours\n",
    "import imutils\n",
    "import threading\n",
    "import asyncio\n",
    "import re\n",
    "import io\n",
    "from time import sleep\n",
    "import datetime as dt\n",
    "import timeit\n",
    "\n",
    "# Confirm opencv import were done correctly\n",
    "print (\"OPEN_CV VER:\\t\", cv2.__version__)\n",
    "print (\"NUMPY VER:\\t\", numpy.__version__)\n",
    "print (\"TESSERECT VER:\\t\", pytesseract.__version__)\n",
    "\n",
    "# some handy functions to use along widgets\n",
    "from IPython.display import clear_output, Image, display, Markdown, HTML\n",
    "# widget packages\n",
    "import ipywidgets as widgets\n",
    "# https://medium.com/@jdchipox/how-to-interact-with-jupyter-33a98686f24e"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "cfc58435-a1a6-47e0-91e1-9e8a6f0c6ab8",
   "metadata": {},
   "outputs": [],
   "source": [
    "no_img = False\n",
    "batt_out = False\n",
    "no_img_count = 0\n",
    "stop_looping = False\n",
    "successful_api_call = False\n",
    "\n",
    "loop_delay = 120\n",
    "adjusted_loop_delay = 0"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "231be823-997f-441a-ae89-521e1db66300",
   "metadata": {},
   "source": [
    "## Camera Object setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "e9975fc4-6402-4584-ad36-573113f70c85",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Camera wasn't open ...\n",
      "Opening Camera ...\n",
      "Camera Opened!\n",
      "Setting camera properties for image taking ...\n",
      "\n",
      "Camera properties:\n",
      "--------------------\n",
      "RES: 320x240\n",
      "FR: 30\n",
      "ISO: 200\n",
      "SS: 33164\n",
      "EXP_MODE: off\n",
      "AWB_GAIN: (Fraction(367, 256), Fraction(331, 256))\n",
      "--------------------\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    camera.close()\n",
    "    print(\"Camera was open!\")\n",
    "    print(\"So just closed it\")\n",
    "except:\n",
    "    print(\"Camera wasn't open ...\")\n",
    "    \n",
    "    \n",
    "    \n",
    "def setup_cam_properties():\n",
    "    print(\"Setting camera properties for image taking ...\")\n",
    "    camera.resolution = (320, 240)\n",
    "    camera.framerate = 30\n",
    "    # Set ISO to the desired value\n",
    "    camera.iso = 200 \n",
    "    # Valid values are between 0 (auto) and 1600. \n",
    "    # The actual value used when iso is explicitly set will be one \n",
    "    # of the following values(whichever is closest): 100, 200, 320, 400, 500, 640, 800\n",
    "    # Wait for the automatic gain control to settle\n",
    "    sleep(2)\n",
    "    # Now fix the values\n",
    "    camera.shutter_speed = camera.exposure_speed\n",
    "    camera.exposure_mode = 'off'\n",
    "    g = camera.awb_gains\n",
    "    camera.awb_mode = 'off'\n",
    "    camera.awb_gains = g\n",
    "    \n",
    "    print(\"\\nCamera properties:\")\n",
    "    print(\"--------------------\")\n",
    "    print(\"RES:\", camera.resolution)\n",
    "    print(\"FR:\", camera.framerate)\n",
    "    print(\"ISO:\", camera.iso)\n",
    "    print(\"SS:\", camera.shutter_speed)\n",
    "    print(\"EXP_MODE:\", camera.exposure_mode)\n",
    "    print(\"AWB_GAIN:\", camera.awb_gains)\n",
    "    print(\"--------------------\")\n",
    "\n",
    "    \n",
    "    \n",
    "try:\n",
    "    print(\"Opening Camera ...\")\n",
    "    camera = PiCamera()\n",
    "except:\n",
    "    print(\"Err Opening Camera ...\")\n",
    "    print(\"Closing first ...\")\n",
    "    camera.close()\n",
    "    sleep(0.5)\n",
    "    camera = PiCamera()\n",
    "finally:\n",
    "    print(\"Camera Opened!\")\n",
    "    setup_cam_properties()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bedf9cf5-6e66-4754-be9b-b845761ed191",
   "metadata": {},
   "source": [
    "# Capture and Pre Processing method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "92c4f434-620d-42b1-a0d5-fb280cd191e7",
   "metadata": {},
   "outputs": [],
   "source": [
    "show_image_plots = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "e3b36774-ad94-4255-bd61-ed5d863e7797",
   "metadata": {},
   "outputs": [],
   "source": [
    "def capture_image():\n",
    "    global blurred\n",
    "    \n",
    "    # Capture and show raw camera content.\n",
    "    # Also will be later used for openCV\n",
    "    print(\"Capturing in RAW for OCR ...\")\n",
    "    rawCapture = PiRGBArray(camera, size=(320, 240))\n",
    "    camera.capture(rawCapture, format=\"rgb\")\n",
    "    img = rawCapture.array\n",
    "    \n",
    "    print(\"Converting to RGB ...\")\n",
    "    rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)\n",
    "    print(\"Converting to GRAY ...\")\n",
    "    gray_img = cv2.cvtColor(rgb, cv2.COLOR_BGR2GRAY)\n",
    "    # blurred = cv2.GaussianBlur(cropped_gray_img, (5, 5), 0)\n",
    "    # blurred = cv2.GaussianBlur(gray_img, (5, 5), 0)\n",
    "    print(\"Blurring to denoise ...\")\n",
    "    blurred = cv2.bilateralFilter(gray_img, 11, 17, 17) #Blur to reduce noise\n",
    "    cropped_gray_img = blurred[20:180, 70:200] # [y:h, x:w]\n",
    "    print(\"Writing file to disk for now ...\")\n",
    "    cv2.imwrite(\"processed_img.png\", blurred)\n",
    "    \n",
    "    if show_image_plots:\n",
    "        plt.imshow(img)\n",
    "        plt.show()\n",
    "        plt.imshow(rgb)\n",
    "        plt.show()\n",
    "        plt.imshow(gray_img, cmap=\"gray\") # without thje cmap=\"gray\" matplotlib doesn;t render gray.\n",
    "        plt.show()\n",
    "        plt.imshow(blurred, cmap=\"gray\") # without thje cmap=\"gray\" matplotlib doesn;t render gray.\n",
    "        plt.show()\n",
    "        \n",
    "    # show small thumbnail image\n",
    "    scale_percent = 50 # percent of original size\n",
    "    width = int(cropped_gray_img.shape[1] * scale_percent / 100)\n",
    "    height = int(cropped_gray_img.shape[0] * scale_percent / 100)\n",
    "    dim = (width, height)\n",
    "    smaller_img = cv2.resize(cropped_gray_img, dim, interpolation = cv2.INTER_AREA)\n",
    "    def cm_to_inch(value):\n",
    "        return value/2.54\n",
    "    plt.figure(figsize=(cm_to_inch(6/2), cm_to_inch(7/2)))\n",
    "    plt.imshow(smaller_img, cmap=\"gray\")\n",
    "    plt.show()\n",
    "   \n",
    "# capture_image()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ac86be17-5b12-4a8e-88dd-fd7de5159327",
   "metadata": {},
   "source": [
    "# Google Cloud vision API based method to do OCR\n",
    "\n",
    "[Guide here](https://pyimagesearch.com/2022/03/31/text-detection-and-ocr-with-google-cloud-vision-api/)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "d0fda18a-6042-4725-b9c6-e55a232f2c31",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create the client interface to access the Google Cloud Vision API\n",
    "credentials = service_account.Credentials.from_service_account_file(filename=\"ocrtest-348215-181db3e230e2.json\", scopes=[\"https://www.googleapis.com/auth/cloud-platform\"])\n",
    "client = vision.ImageAnnotatorClient(credentials=credentials)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "6fe8a7e2-7f96-4f6a-8a65-c711dd43cab3",
   "metadata": {},
   "outputs": [],
   "source": [
    "def getGoogleCloudOCRRes():\n",
    "    global no_img\n",
    "    global batt_out\n",
    "    global no_img_count\n",
    "    global successful_api_call\n",
    "    \n",
    "    print(\"Loading file from disk to transfer to Google Vision API ...\")\n",
    "    with io.open(\"processed_img.png\", \"rb\") as f:\n",
    "        byteImage = f.read()\n",
    "\n",
    "    # create an image object from the binary file and then make a request\n",
    "    # to the Google Cloud Vision API to OCR the input image\n",
    "    print(\"Making request to Google Cloud Vision API ...\")\n",
    "    successful_api_call = False\n",
    "    try:\n",
    "        image = vision.Image(content=byteImage)\n",
    "        response = client.text_detection(image=image)\n",
    "        successful_api_call = True\n",
    "        print(\"API response status:\", successful_api_call)\n",
    "    except Exception as e:\n",
    "        successful_api_call = False\n",
    "        print(\"Something went wrong while calling the service ...\")\n",
    "        print(\"--------------------------------------------------\")\n",
    "        print(e)\n",
    "        return None\n",
    "\n",
    "    # check to see if there was an error when making a request to the API\n",
    "    if response.error.message:\n",
    "        print(\"RESPONSE ERROR ...\")\n",
    "        print(\"------------------\")\n",
    "        raise Exception(\n",
    "            \"{}\\nFor more info on errors, check:\\n\"\n",
    "            \"https://cloud.google.com/apis/design/errors\".format(\n",
    "                response.error.message))\n",
    "        return None\n",
    "    else:\n",
    "        print(\"Checking len of response ...\")\n",
    "        print(\"Length of Resonse:\", len(response.text_annotations))\n",
    "        if len(response.text_annotations) is 0:\n",
    "            no_img = True\n",
    "            no_img_count = no_img_count +1\n",
    "        else:\n",
    "            no_img = False\n",
    "            batt_out = False\n",
    "            no_img_count = 0\n",
    "\n",
    "            print(\"Current Date time:\", dt.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))\n",
    "            # loop over the Google Cloud Vision API OCR result response\n",
    "            print(\"Found text/digits:\")\n",
    "            sleep(1)\n",
    "            ocr_time = []\n",
    "            for text in response.text_annotations[1::]:\n",
    "                ocr = text.description\n",
    "                ocr_time.append(ocr)\n",
    "            if len(response.text_annotations) > 1:\n",
    "                ocr_time.insert(1, ':')   \n",
    "            print(''.join(ocr_time))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e3606a23-294f-4815-9a0b-c154bcc4cd67",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Click function definition with Servo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "7e31fbe9-65c8-4845-9c75-a894f40739a3",
   "metadata": {},
   "outputs": [],
   "source": [
    "servo_kit = ServoKit(channels=16, address=0x40, frequency=60) \n",
    "servo_kit.servo[0].actuation_range = 180\n",
    "\n",
    "min_pos = 9\n",
    "max_pos = 30\n",
    "swing_delay = 0.5\n",
    "\n",
    "start = 0 # for time keeping, to measure elapsed time per run, till OCR is finished\n",
    "\n",
    "def click_with_servo(_):\n",
    "    global start\n",
    "    \n",
    "    servo_kit.servo[0].angle = max_pos\n",
    "    sleep(swing_delay)\n",
    "    servo_kit.servo[0].angle = min_pos\n",
    "    # Starts a timer to keep elapsed time for the process that will begin,\n",
    "    # once the servo has pressed the button\n",
    "    start = timeit.default_timer()\n",
    "    # \n",
    "    sleep(swing_delay)\n",
    "    servo_kit.servo[0].angle = max_pos    \n",
    "    sleep(0.25)\n",
    "    servo_kit.servo[0].angle = None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "42cdff37-ea6a-4f2c-81a3-856a8ce646c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "def validate_no_img_reason():\n",
    "    global no_img\n",
    "    global batt_out\n",
    "    global no_img_count\n",
    "    global stop_looping\n",
    "    \n",
    "    \n",
    "    if no_img and no_img_count == 1:\n",
    "        # no image found, will run again\n",
    "        print(\"No image found, will try 2 more run cycles\")\n",
    "    elif no_img and no_img_count == 2:\n",
    "        # no image found, will run again\n",
    "        print(\"No image found, will try 1 more run cycle\")\n",
    "    elif no_img and no_img_count == 3:\n",
    "        # battery dead (or something seriously wrong with google ocr)\n",
    "        batt_out = True\n",
    "        stop_looping = True\n",
    "        \n",
    "        print(\"No image found\")\n",
    "        sleep(1)\n",
    "        print(\"Battery is probably dead\")\n",
    "        sleep(1)\n",
    "        print(\"Stop Clicking the watch!\")\n",
    "        # dt.datetime.now().strftime('%Y-%m-%d %H:%M:%S')\n",
    "    elif no_img == False:\n",
    "        # OCR found, battery still have juice  \n",
    "        print(\"Battery still has some juice\")\n",
    "    \n",
    "        \n",
    "        \n",
    "\n",
    "def run_all_methods(_):\n",
    "    global start\n",
    "    global loop_delay\n",
    "    global adjusted_loop_delay\n",
    "    global successful_api_call\n",
    "    \n",
    "    click_with_servo(_)\n",
    "    sleep(0.5)\n",
    "    capture_image()\n",
    "    getGoogleCloudOCRRes()\n",
    "    \n",
    "    # Closes off the timer & calculate how much we need to adjust the delay between loops\n",
    "    end = timeit.default_timer()\n",
    "    elapsed_time = round(end-start, 2)\n",
    "    adjusted_loop_delay = round(loop_delay-elapsed_time, 2)\n",
    "    \n",
    "    print(\"\\nElapsed time in the process so far, since last button press: \", elapsed_time)\n",
    "    print(\"For a loop_delay of\", loop_delay, \"secs, next process should begin in\", adjusted_loop_delay, \"secs\")\n",
    "    \n",
    "    global no_img\n",
    "    print('\\nimage?', not no_img)\n",
    "    \n",
    "    validate_no_img_reason()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ab657c6b-20e8-4c9c-b46e-a1d67e283c38",
   "metadata": {},
   "source": [
    "# Servo click button"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "83c10c62-f6ad-4262-94c0-4f434537993d",
   "metadata": {},
   "outputs": [],
   "source": [
    "click_btn = widgets.Button(\n",
    "    description='click the watch',\n",
    "    disabled=False,\n",
    "    button_style='success', # 'success', 'info', 'warning', 'danger' or ''\n",
    "    tooltip='clicks the watch button with servo',\n",
    "    icon='hand-o-left'\n",
    ")\n",
    "\n",
    "\n",
    "click_btn.on_click(run_all_methods)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "540644d9-cf08-4426-b748-07204442a4c3",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Pure Ugly loop method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "1a3e9ef4-1213-4121-91f7-1749cfbac46f",
   "metadata": {},
   "outputs": [],
   "source": [
    "count = 0\n",
    "\n",
    "def begin_test_cycle(_):\n",
    "    while True:\n",
    "        global count\n",
    "        global loop_delay\n",
    "        global adjusted_loop_delay\n",
    "        global stop_looping\n",
    "\n",
    "        # Vars imported for restting, when battery is out\n",
    "        global no_img\n",
    "        global batt_out\n",
    "        global no_img_count\n",
    "        # -----------------------------------------------\n",
    "\n",
    "        if stop_looping:\n",
    "            no_img = False\n",
    "            batt_out = False\n",
    "            no_img_count = 0\n",
    "            stop_looping = False\n",
    "            break\n",
    "\n",
    "        print(\"\\nCLICK:\", count)\n",
    "        print(\"-------------\")\n",
    "\n",
    "        run_all_methods(_)\n",
    "        count = count + 1\n",
    "        sleep(adjusted_loop_delay)\n",
    "        \n",
    "\n",
    "# def stop_test_cycle(_):\n",
    "#     global stop_looping\n",
    "#     stop_looping = True\n",
    "#     print (\"Stopping next run ...\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af67fe02-c59f-4212-b0da-ab6e650cabf9",
   "metadata": {},
   "source": [
    "# Loop Start/Stop Buttons"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "ee599334-28f7-4af5-8d04-27722167a242",
   "metadata": {},
   "outputs": [],
   "source": [
    "loop_begin_btn = widgets.Button(\n",
    "    description='Begin Test Cycle',\n",
    "    disabled=False,\n",
    "    button_style='warning', # 'success', 'info', 'warning', 'danger' or ''\n",
    "    tooltip='loops the test cycle',\n",
    "    icon='refresh'\n",
    ")\n",
    "\n",
    "loop_begin_btn.on_click(begin_test_cycle)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b922205d-0ead-4a07-b2ee-51702345f015",
   "metadata": {},
   "source": [
    "# Display control buttons"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "9cdafaa0-59fa-45e7-89c7-043d8fffe054",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "8bd81ea584134d6c99a11b353d7a84f0",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "Button(button_style='success', description='click the watch', icon='hand-o-left', style=ButtonStyle(), tooltip…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "dea5b469ae0d401eb8cfd2924cec7221",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "Button(button_style='warning', description='Begin Test Cycle', icon='refresh', style=ButtonStyle(), tooltip='l…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "display(click_btn)\n",
    "display(loop_begin_btn)"
   ]
  }
 ],
 "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.7.3"
  },
  "toc-autonumbering": true,
  "toc-showcode": true,
  "toc-showmarkdowntxt": true
 },
 "nbformat": 4,
 "nbformat_minor": 5
}