steam-subtraction-v4

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "initial_id",
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import cv2\n",
    "import datetime\n",
    "import matplotlib.animation as animation\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import pybgs as bgs\n",
    "\n",
    "from IPython.display import HTML\n",
    "\n",
    "from breathecam import BreatheCam\n",
    "from common import get_previous_frame_time\n",
    "from components import View\n",
    "from motion import temporal_events, mask_background, get_event\n",
    "\n",
    "\n",
    "def display_event(video, flattened=False):\n",
    "    fig, ax = plt.subplots(1, 1)\n",
    "    fig.tight_layout()\n",
    "\n",
    "    ax.set_axis_off()\n",
    "\n",
    "    ims = []\n",
    "\n",
    "    if len(video.shape) < 4 and flattened:\n",
    "        ims.append([ax.imshow(video)])\n",
    "    else:\n",
    "        for i in range(len(video)):\n",
    "            ims.append([ax.imshow(video[i], animated=True)])\n",
    "\n",
    "    anim = animation.ArtistAnimation(fig, ims, interval=175, blit=True, repeat_delay=1000)\n",
    "\n",
    "    plt.close()\n",
    "\n",
    "    display(HTML(anim.to_jshtml()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "87a8a63d6e2bb3c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "#expertiment settings and avoid midnight\n",
    "day = datetime.date.fromisoformat(\"2024-05-19\")\n",
    "time = datetime.time.fromisoformat(\"09:49:00\")\n",
    "previous_frame_time = get_previous_frame_time(time, 3)\n",
    "nframes = 80\n",
    "nlevels = 4\n",
    "view = View(2307, 1914, 6814, 2515)\n",
    "\n",
    "print(f\"Size: {view.width}x{view.height} ({view.width * view.height * nframes} pixels)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3e5c1e77aa8e8d5e",
   "metadata": {},
   "outputs": [],
   "source": [
    "camera = BreatheCam.init_from(\"Clairton Coke Works\", day)\n",
    "breathecam_video = camera.download_video(previous_frame_time, nframes+1, view, nlevels)\n",
    "fullres_video = camera.download_video(time, nframes, view, 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ff50861f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#modified mask_background from motion.py so that can take bgs background subtractor instances\n",
    "def mask_background(video: np.ndarray, background_subtractor) -> np.ndarray:\n",
    "    \"\"\"Generates, by background subtraction, a video of foreground masks\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    * video - a collection of frames\n",
    "    * background_subtractor - a mask generating function to apply to each frame\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "\n",
    "    A collection of frames representing foreground objects in each frame\n",
    "    \"\"\"\n",
    "\n",
    "    if isinstance(background_subtractor, cv2.BackgroundSubtractor):\n",
    "        print(\"using cv background subtractor...\")\n",
    "        f = background_subtractor.apply\n",
    "    elif isinstance(background_subtractor, bgs.SuBSENSE):\n",
    "        print(\"using bgs SuBENSE background subtractor\")\n",
    "        f = background_subtractor.apply\n",
    "    elif isinstance(background_subtractor, bgs.PixelBasedAdaptiveSegmenter):\n",
    "        print(\" PixelBasedAdaptiveSegmenter\")\n",
    "        f = background_subtractor.apply\n",
    "    elif isinstance(background_subtractor, bgs.LOBSTER):\n",
    "        print(\"LOBSTER\")\n",
    "        f = background_subtractor.apply\n",
    "    else:\n",
    "        #Also tried ViBe algorithm but couldn't find name\n",
    "        print(\"couldn't find name of bgs background subtractor class\")\n",
    "        f = background_subtractor.apply\n",
    "\n",
    "    if len(video.shape) == 3:\n",
    "        return np.array([f(video)])\n",
    "    else:\n",
    "        return np.array([f(v) for v in video])\n",
    "\n",
    "#instantiate background_subtractor instance here\n",
    "#name of class sometimes hard to find\n",
    "#find in this GitHub: https://github.com/andrewssobral/bgslibrary/tree/master/bgslibrary/algorithms\n",
    "background_subtractor = bgs.PixelBasedAdaptiveSegmenter()\n",
    "masks = mask_background(breathecam_video[1:], background_subtractor)\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d4efba8495b71f76",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "eventspace = temporal_events(masks, neighbors=8, depth=3, threshold=127)\n",
    "event_filter = lambda e : e.number_of_frames > 3 and e.region.height > 10 and e.region.width > 10\n",
    "eventspace = list(filter(event_filter, eventspace))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3470d1b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "#more of a type than a class. Acutall functionality implemented in sub-classes\n",
    "\n",
    "#init method. set nessisary hyperparameters alpha, beta etc... \n",
    "\n",
    "#update method. mutates the passed estimates array. \n",
    "\n",
    "#TODO define estimators as subclasses?\n",
    "class Estimator():\n",
    "    def __init__(self):\n",
    "        pass\n",
    "    def update(self, estimates, test_results):\n",
    "        pass\n",
    "\n",
    "class Exponential_Smoothing_Estimator():\n",
    "    def __init__(self, alpha):\n",
    "        self.alpha = alpha\n",
    "    \n",
    "    def update(self, estimates, test_results):\n",
    "        assert(estimates.shape == test_results.shape)\n",
    "        rows, cols = estimates.shape\n",
    "\n",
    "        for r in range(rows):\n",
    "            for c in range(cols):\n",
    "                alpha_estimate = self.alpha * test_results[r, c]\n",
    "                prev_estimate = (1 - self.alpha) * estimates[r, c]\n",
    "                estimates[r, c] = alpha_estimate + prev_estimate\n",
    "\n",
    "class Moving_Average_Estimator():\n",
    "    #Skip this one because doesn't fit into the Estimator class interface well\n",
    "    #Also is prob not as good as others\n",
    "    def __init__(self):\n",
    "        return\n",
    "    \n",
    "    def update(self, estimates, test_results):\n",
    "        assert(estimates.shape == test_results.shape)\n",
    "        rows, cols = estimates.shapes\n",
    "        for r in range(rows):\n",
    "            for c in range(cols):\n",
    "                #If continue implementing. Maybe implement as running average\n",
    "                pass\n",
    "class Double_Exponential_Smoothing_Estimator():\n",
    "    def __init__(self, alpha, beta):\n",
    "        self.alpha = alpha\n",
    "        self.beta = beta\n",
    "    \n",
    "    #TODO don't understand how exactly how to implement this one ask Harry "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e2eac6ed",
   "metadata": {},
   "outputs": [],
   "source": [
    "def naive_steam_approx(masks, video, estimator, steam_probability, intensity_tresh = 70):\n",
    "    #Non-mutating. Returns a copy of video with steam pixels subtracted\n",
    "    \n",
    "    estimates = np.full(video.shape[1:3], 0.5)\n",
    "    for frameIdx in range(video.shape[0]):\n",
    "        gray_frame = cv2.cvtColor(video[frameIdx], cv2.COLOR_BGR2GRAY)\n",
    "        curr_mask = masks[frameIdx]\n",
    "        selected_pixels = np.where(curr_mask == 255, gray_frame, 0)\n",
    "        test_results = selected_pixels > intensity_tresh\n",
    "        test_results = np.where(test_results, 1, 0)\n",
    "        estimator.update(estimates, test_results)\n",
    "    \n",
    "    #Remove any pixels with a high probability of being steam\n",
    "    steam_mask = estimates > steam_probability\n",
    "    steam_mask = steam_mask[np.newaxis, :, :, np.newaxis]\n",
    "    steam_mask = np.broadcast_to(steam_mask, video.shape)\n",
    "    non_steam_video = video.copy()\n",
    "    non_steam_video[steam_mask] = 0\n",
    "    return non_steam_video"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9f311fde",
   "metadata": {},
   "outputs": [],
   "source": [
    "#exclude 0th frame of video because that frame is an extra framed just added \n",
    "#to init background subtractor\n",
    "estimator = Exponential_Smoothing_Estimator(0.45)\n",
    "non_steam_video = naive_steam_approx(masks, breathecam_video[1:], estimator, 0.55, intensity_tresh=70)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "946cbd80",
   "metadata": {},
   "outputs": [],
   "source": [
    "display_event(fullres_video)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e96c1e5b",
   "metadata": {},
   "outputs": [],
   "source": [
    "display_event(masks)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a3f9e3ce",
   "metadata": {},
   "outputs": [],
   "source": [
    "display_event(non_steam_video)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "be7b482e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#event shape: (num_frames, _, _,channels)\n",
    "events_cropped_vids = [] \n",
    "for i in range(10):\n",
    "    event = get_event(eventspace[i], fullres_video[], nlevels=4)\n",
    "    events_cropped_vids.append(event)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8df27c56",
   "metadata": {},
   "outputs": [],
   "source": [
    "for event_vid in events_cropped_vids:\n",
    "    display_event(event_vid)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a84fab78",
   "metadata": {},
   "outputs": [],
   "source": [
    "#code for cv2 optical flow usage from:\n",
    "#1) https://docs.opencv.org/4.x/d4/dee/tutorial_optical_flow.html\n",
    "#2) https://www.geeksforgeeks.org/python-opencv-dense-optical-flow/\n",
    "\n",
    "firstframe = event[0]\n",
    "prev_gray = cv2.cvtColor(firstframe, cv2.COLOR_BGR2GRAY)\n",
    "\n",
    "hsv_mask = np.zeros_like(firstframe)\n",
    "hsv_mask[..., 1] = 255\n",
    "\n",
    "flow_vid = []\n",
    "\n",
    "for frame in event[1:]:\n",
    "    curr_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)\n",
    "    flow = cv2.calcOpticalFlowFarneback(prev_gray, curr_gray, None,\n",
    "                                        0.5, 3, 15, 3, 5, 1.2, 0)\n",
    "\n",
    "    magnitude, angle = cv2.cartToPolar(flow[..., 0], flow[..., 1])\n",
    "\n",
    "    hsv_mask[..., 0] = angle * 180 / np.pi / 2\n",
    "\n",
    "    hsv_mask[..., 2] = cv2.normalize(magnitude, None, 0, 255, cv2.NORM_MINMAX)\n",
    "    rgb = cv2.cvtColor(hsv_mask, cv2.COLOR_HSV2BGR)\n",
    "    flow_vid.append(rgb)\n",
    "\n",
    "    prev_gray = curr_gray\n",
    "\n",
    "flow_vid = np.array(flow_vid)\n",
    "display_event(flow_vid)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}