Data compression for screen data

.github/workflows/test.yml

@@ -28,6 +28,11 @@ jobs:
         with:
           python-version: '3.12.8'
 
+      - name: Install FFmpeg
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y ffmpeg
+
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip

.gitignore

@@ -163,3 +163,6 @@ cython_debug/
 
 *.sif
 *.bak
+
+# local workflow testing
+bin/ 

examples/allen_demo.ipynb

@@ -0,0 +1,215 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "5dc8ad24",
+   "metadata": {},
+   "source": [
+    "### Showcase of interpolation on Experiment level for encoded videos and images in greyscale and RGB."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "e4aecde6",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# import dependencies\n",
+    "\n",
+    "%load_ext autoreload\n",
+    "%autoreload 2\n",
+    "%matplotlib inline\n",
+    "\n",
+    "import numpy as np\n",
+    "import matplotlib.pyplot as plt\n",
+    "import matplotlib.animation as animation\n",
+    "from IPython.display import HTML"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "9bc17dbb",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# get the experiment class from experanto \n",
+    "from experanto.experiment import Experiment\n",
+    "\n",
+    "# set experiment folder as root\n",
+    "root_folder = '../../allen-exporter/data/allen_data/experiment_951980471'\n",
+    "\n",
+    "# initialize experiment object\n",
+    "e = Experiment(root_folder)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2447e233",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "times = np.arange(310., 340., 0.5)\n",
+    "video = e.interpolate(times, device=\"screen\")  # shape: (T, C, H, W)\n",
+    "\n",
+    "# Clip and convert to uint8 to avoid matplotlib clipping warnings\n",
+    "video = np.clip(video, 0, 255).astype(np.uint8)\n",
+    "video = video.transpose(0, 2, 3, 1)  # (T, H, W, C)\n",
+    "\n",
+    "n_frames, height, width, channels = video.shape\n",
+    "print(f\"Video shape: {video.shape}\")\n",
+    "\n",
+    "# Handle grayscale vs color\n",
+    "is_grayscale = (channels == 1)\n",
+    "if is_grayscale:\n",
+    "    video = video[..., 0]  # Now shape: (T, H, W)\n",
+    "\n",
+    "fig, ax = plt.subplots()\n",
+    "\n",
+    "# Initialize with appropriate cmap\n",
+    "if is_grayscale:\n",
+    "    img = ax.imshow(video[0], cmap='gray', vmin=0, vmax=255)\n",
+    "else:\n",
+    "    img = ax.imshow(video[0])\n",
+    "\n",
+    "ax.axis('off')\n",
+    "\n",
+    "def update(frame):\n",
+    "    img.set_array(video[frame])\n",
+    "    ax.set_title(f'Frame {frame}')\n",
+    "    return [img]\n",
+    "\n",
+    "ani = animation.FuncAnimation(fig, update, frames=n_frames, interval=50, blit=True)\n",
+    "\n",
+    "plt.close(fig)\n",
+    "HTML(ani.to_jshtml())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "129ae368",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%%timeit\n",
+    "video = e.interpolate(times, device=\"screen\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "d58eaaeb",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "times = np.arange(4484., 4500., 0.5)\n",
+    "video = e.interpolate(times, device=\"screen\")  # shape: (T, C, H, W)\n",
+    "\n",
+    "# Clip and convert to uint8 to avoid matplotlib clipping warnings\n",
+    "video = np.clip(video, 0, 255).astype(np.uint8)\n",
+    "video = video.transpose(0, 2, 3, 1)  # (T, H, W, C)\n",
+    "\n",
+    "n_frames, height, width, channels = video.shape\n",
+    "print(f\"Video shape: {video.shape}\")\n",
+    "\n",
+    "# Handle grayscale vs color\n",
+    "is_grayscale = (channels == 1)\n",
+    "if is_grayscale:\n",
+    "    video = video[..., 0]  # Now shape: (T, H, W)\n",
+    "\n",
+    "fig, ax = plt.subplots()\n",
+    "\n",
+    "# Initialize with appropriate cmap\n",
+    "if is_grayscale:\n",
+    "    img = ax.imshow(video[0], cmap='gray', vmin=0, vmax=255)\n",
+    "else:\n",
+    "    img = ax.imshow(video[0])\n",
+    "\n",
+    "ax.axis('off')\n",
+    "\n",
+    "def update(frame):\n",
+    "    img.set_array(video[frame])\n",
+    "    ax.set_title(f'Frame {frame}')\n",
+    "    return [img]\n",
+    "\n",
+    "ani = animation.FuncAnimation(fig, update, frames=n_frames, interval=50, blit=True)\n",
+    "\n",
+    "plt.close(fig)\n",
+    "HTML(ani.to_jshtml())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "9a74eb40",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%%timeit\n",
+    "video = e.interpolate(times, device=\"screen\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "ba9c51eb",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Download a test_rgb_image.jpg (1200x1900!) put it into the folder with other stimuli and run this to test rgb interpolation in the third cell.\n",
+    "# My test image : https://wallpapercave.com/download/1900x1200-wallpapers-wp9725873\n",
+    "# Also adjust number_channels to 3 inside the default.yaml config file\n",
+    "\n",
+    "import cv2\n",
+    "import numpy as np\n",
+    "import matplotlib.pyplot as plt\n",
+    "\n",
+    "# Load the JPEG image (in BGR format)\n",
+    "img = cv2.imread('../../../allen-exporter/data/allen_data/experiment_951980471/stimuli/test_rgb_image.jpg')\n",
+    "\n",
+    "# Optional: convert from BGR to RGB if you want\n",
+    "img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)\n",
+    "\n",
+    "# Plot the image\n",
+    "plt.imshow(img_rgb)\n",
+    "plt.axis('off')  # hide axes\n",
+    "plt.show()\n",
+    "\n",
+    "# Save the image as a .npy file\n",
+    "np.save('../../../allen-exporter/data/allen_data/experiment_951980471/stimuli/im065.npy', img_rgb)\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2c52e9e3",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "experanto",
+   "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.12.8"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

experanto/datasets.py

@@ -316,11 +316,14 @@ def initialize_transforms(self):
                             transform_list.append(module)
 
                 transform_list.insert(0, add_channel)
-            else:
 
-                transform_list = [ToTensor()]
+            else:
+                transform_list = [
+                    lambda x: torch.from_numpy(x)
+                    .float()
+                    .unsqueeze(0)  # Add C dim to use torchvision transform
+                ]
 
-            # Normalization.
             if self.modality_config[device_name].transforms.get("normalization", False):
                 transform_list.append(
                     torchvision.transforms.Normalize(
@@ -330,6 +333,7 @@ def initialize_transforms(self):
                 )
 
             transforms[device_name] = Compose(transform_list)
+
         return transforms
 
     def _get_callable_filter(self, filter_config):
@@ -636,7 +640,7 @@ def __getitem__(self, idx) -> dict:
             # scale everything back to truncated values
             times = times.astype(np.float64) / self.scale_precision
 
-            data, _ = self._experiment.interpolate(times, device=device_name)
+            data = self._experiment.interpolate(times, device=device_name)
             out[device_name] = self.transforms[device_name](data).squeeze(
                 0
             )  # remove dim0 for response/eye_tracker/treadmill

experanto/experiment.py

@@ -86,16 +86,15 @@ def _load_devices(self) -> None:
     def device_names(self):
         return tuple(self.devices.keys())
 
-    def interpolate(self, times: slice, device=None) -> tuple[np.ndarray, np.ndarray]:
+    def interpolate(self, times: slice, device=None) -> np.ndarray:
         if device is None:
             values = {}
-            valid = {}
             for d, interp in self.devices.items():
-                values[d], valid[d] = interp.interpolate(times)
+                values[d] = interp.interpolate(times)
         elif isinstance(device, str):
             assert device in self.devices, "Unknown device '{}'".format(device)
-            values, valid = self.devices[device].interpolate(times)
-        return values, valid
+            values = self.devices[device].interpolate(times)
+        return values
 
     def get_valid_range(self, device_name) -> tuple:
         return tuple(self.devices[device_name].valid_interval)