Update to Control System to handle non-dome cameras with non-linear r…

axis-ptz-controller.env

@@ -46,3 +46,5 @@ INCLUDE_AGE=True
 LOG_TO_MQTT=False
 LOG_LEVEL=INFO
 CONTINUE_ON_EXCEPTION=False
+
+IS_DOME = True

axis-ptz-controller/axis_ptz_controller.py

@@ -90,6 +90,7 @@ def __init__(
         log_to_mqtt: bool = False,
         log_level: str = "INFO",
         continue_on_exception: bool = False,
+        is_dome: bool = True,
         **kwargs: Any,
     ):
         """Instantiate the PTZ controller by connecting to the camera
@@ -186,6 +187,8 @@ def __init__(
         continue_on_exception: bool
             Continue on unhandled exceptions if True, raise exception
             if False (the default)
+        is_dome: bool
+            flag for if this is a Dome type PTZ camera or a fully articulating camera
 
         Returns
         -------
@@ -224,6 +227,7 @@ def __init__(
         self.log_to_mqtt = log_to_mqtt
         self.log_level = log_level
         self.continue_on_exception = continue_on_exception
+        self.is_dome = is_dome
 
         # Always construct camera configuration and control since
         # instantiation only assigns arguments
@@ -290,6 +294,7 @@ def __init__(
             hyperfocal_distance=hyperfocal_distance,
             use_camera=use_camera,
             auto_focus=auto_focus,
+            is_dome=is_dome,
         )
 
         # Object to track
@@ -931,7 +936,7 @@ def _object_callback(
             self.object.update_from_msg(data)
             #self.object.recompute_location()
 
-        if self.use_camera and self.object.tau < 0:
+        if self.use_camera and self.is_dome and self.object.tau < 0:
             logging.info(f"Stopping image capture of object: {self.object.object_id}")
             self.object = None
             self.do_capture = False
@@ -1325,6 +1330,7 @@ def make_controller() -> AxisPtzController:
         continue_on_exception=ast.literal_eval(
             os.environ.get("CONTINUE_ON_EXCEPTION", "False")
         ),
+        is_dome=ast.literal_eval(os.environ.get("IS_DOME", "True")),
     )
 
 

axis-ptz-controller/camera.py

@@ -37,6 +37,7 @@ def __init__(
         focus_min: int = 7499,
         focus_max: int = 9999,
         hyperfocal_distance: float = 22500.0,
+        is_dome: bool = True,
     ) -> None:
         """Initializes the instance of the Camera class.
 
@@ -76,6 +77,7 @@ def __init__(
         self.hyperfocal_distance = hyperfocal_distance
         self.use_camera = use_camera
         self.auto_focus = auto_focus
+        self.is_dome = is_dome
 
         self.rho = 0.0
         self.tau = 0.0
@@ -416,14 +418,18 @@ def _compute_pan_rate_index(self, rho_dot: float) -> None:
         Returns
         -------
         """
-        if rho_dot < -self.pan_rate_max:
-            self.pan_rate_index = -100
-
-        elif self.pan_rate_max < rho_dot:
-            self.pan_rate_index = +100
-
-        else:
-            self.pan_rate_index = (100 / self.pan_rate_max) * rho_dot
+        if self.is_dome:  # Dome style cameras have linear behavior
+            self.pan_rate_index = rho_dot/self.pan_rate_max*100.0
+        else:  # Articulating style cameras have exponential behavior, gives more precise control at low numbers
+            self.pan_rate_index = (abs(rho_dot)/0.002)**(1/2.38824)*(rho_dot/abs(rho_dot))
+
+        if self.pan_rate_index<-100:
+            self.pan_rate_index=-100
+
+        if self.pan_rate_index>100:
+            self.pan_rate_index=100
+
+        # logging.info(f'{self.pan_rate_index} | {rho_dot}')
         # Even though the VAPIX API says it only supports INT, it seems to handle floats just fine
 
     def _compute_tilt_rate_index(self, tau_dot: float) -> None:
@@ -439,14 +445,18 @@ def _compute_tilt_rate_index(self, tau_dot: float) -> None:
         Returns
         -------
         """
-        if tau_dot < -self.tilt_rate_max:
-            self.tilt_rate_index = -100
-
-        elif self.tilt_rate_max < tau_dot:
-            self.tilt_rate_index = 100
-
-        else:
-            self.tilt_rate_index = (100 / self.tilt_rate_max) * tau_dot
+        if self.is_dome:  # Dome style cameras have linear behavior
+            self.tilt_rate_index = tau_dot/self.tilt_rate_max*100.0
+        else:  # Articulating style cameras have exponential behavior
+            self.tilt_rate_index = (abs(tau_dot)/0.002)**(1/2.38824)*(tau_dot/abs(tau_dot))
+
+        if self.tilt_rate_index<-100:
+            self.tilt_rate_index=-100
+
+        if self.tilt_rate_index>100:
+            self.tilt_rate_index=100
+
+        # logging.info(f'{self.tilt_rate_index} | {tau_dot}')
         # Even though the VAPIX API says it only supports INT, it seems to handle floats just fine
 
     def get_yaw_pitch_roll(self) -> Tuple[float, float, float]:

utils/Analysis-PanRate.ipynb

@@ -0,0 +1,233 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "d14779b0",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import requests\n",
+    "from typing import Any, Dict\n",
+    "import schedule\n",
+    "import time\n",
+    "import threading\n",
+    "import numpy as np\n",
+    "import matplotlib.pyplot as plt\n",
+    "import csv\n",
+    "# Replace these with your camera's IP address, username, and password\n",
+    "\n",
+    "camera_ip = \"xx.xx.xx.xx\"\n",
+    "username = \"username\"\n",
+    "password = \"password\"\n",
+    "\n",
+    "\n",
+    "class cameraTestSuite():\n",
+    "    def __init__(\n",
+    "        self: Any,\n",
+    "        camera_ip: str,\n",
+    "        username: str,\n",
+    "        password: str,\n",
+    "    ):\n",
+    "        self.url = f\"http://{camera_ip}/axis-cgi/com/ptz.cgi?\"\n",
+    "        self.auth = requests.auth.HTTPDigestAuth(username, password)\n",
+    "        self.pan = 0\n",
+    "        self.tilt = 0\n",
+    "        self.zoom = 0\n",
+    "        self.data = []\n",
+    "\n",
+    "    def reset(self):\n",
+    "        self.pan = 0\n",
+    "        self.tilt = 0\n",
+    "        self.data = []\n",
+    "\n",
+    "    # Axis PTZ control URL and parameters\n",
+    "    def moveCamera(self,pan, tilt):\n",
+    "        response = requests.get(f'{self.url}pan={pan}&tilt={tilt}', auth=self.auth)\n",
+    "        # Check response\n",
+    "        if response.status_code != 204:\n",
+    "            print(f\"Failed to move camera: {response.status_code}, Response: {response.text}\")\n",
+    "\n",
+    "    def zoomCamera(self,zoom):\n",
+    "        response = requests.get(f'{self.url}zoom={zoom}', auth=self.auth)\n",
+    "        # Check response\n",
+    "        if response.status_code != 204:\n",
+    "            print(f\"Failed to move camera: {response.status_code}, Response: {response.text}\")\n",
+    "        \n",
+    "    \n",
+    "    def moveCameraRate(self, panRate=0, tiltRate=0):\n",
+    "        response = requests.get(f'{self.url}continuouspantiltmove={panRate},{tiltRate}', auth=self.auth)\n",
+    "        # Check response\n",
+    "        if response.status_code != 204:\n",
+    "            print(f\"Failed to move camera: {response.status_code}, Response: {response.text}\")\n",
+    "\n",
+    "    def getCurrentPosition(self):\n",
+    "        params = {\n",
+    "            \"query\": \"position\"  # This query parameter should be adjusted based on camera API documentation\n",
+    "        }\n",
+    "        # Send the request to get the camera's current position\n",
+    "        response_start = time.time()\n",
+    "        response = requests.get(self.url, params=params, auth=self.auth)\n",
+    "        response_time = time.time() - response_start\n",
+    "        if response.status_code != 200:\n",
+    "            print(f\"Failed to get camera position: {response.status_code}, Response: {response.text}\")\n",
+    "        \n",
+    "        response_str = response.content.decode('UTF-8').split('\\r\\n')\n",
+    "        response_dict = {}\n",
+    "        for line in response_str:\n",
+    "            if line:\n",
+    "                key, value = line.split('=')\n",
+    "                response_dict[key] = value\n",
+    "        self.pan = float(response_dict['pan'])\n",
+    "        self.tilt = float(response_dict['tilt'])\n",
+    "        self.zoom = float(response_dict['zoom'])\n",
+    "        response_dict['timestamp'] = time.time()\n",
+    "        response_dict['response_time'] = response_time\n",
+    "        self.data.append(response_dict)\n",
+    "        # Check response\n",
+    "        return response_dict\n",
+    "    \n",
+    "    def run_schedule(self):\n",
+    "        while True:\n",
+    "            schedule.run_pending()\n",
+    "            time.sleep(0.01)\n",
+    "\n",
+    "    def start_pan(self, panRate=0, tiltRate=0):\n",
+    "        \n",
+    "        self.moveCameraRate(panRate, tiltRate)\n",
+    "        self.moveCamera(pan=0, tilt=0)\n",
+    "        #self.zoomCamera(1)\n",
+    "        time.sleep(10)\n",
+    "        self.moveCamera(pan=0, tilt=0)\n",
+    "        time.sleep(10)\n",
+    "        self.reset()\n",
+    "        schedule.every(0.1).seconds.do(self.getCurrentPosition)\n",
+    "        #print(\"Moving camera at panRate: \", panRate, \"tiltRate: \", tiltRate)\n",
+    "        self.moveCameraRate(tiltRate=0, panRate=panRate)\n",
+    "        \n",
+    "    def start_tilt(self, panRate=0, tiltRate=0):\n",
+    "        \n",
+    "        self.moveCameraRate(panRate, tiltRate)\n",
+    "        self.moveCamera(pan=0, tilt=-90)\n",
+    "        #self.zoomCamera(1)\n",
+    "        time.sleep(10)\n",
+    "        self.moveCamera(pan=0, tilt=-90)\n",
+    "        time.sleep(10)\n",
+    "        self.reset()\n",
+    "        schedule.every(0.1).seconds.do(self.getCurrentPosition)\n",
+    "        print(\"Moving camera at panRate: \", panRate, \"tiltRate: \", tiltRate)\n",
+    "        self.moveCameraRate(tiltRate=tiltRate, panRate=0)\n",
+    "    \n",
+    "    def stop(self):\n",
+    "        self.moveCameraRate(0, 0)\n",
+    "        schedule.clear()\n",
+    "\n",
+    "    def main(self):\n",
+    "        self.moveCamera(pan=0, tilt=0)\n",
+    "        self.zoomCamera(1)\n",
+    "        #schedule.every(0.1).seconds.do(self.getCurrentPosition)\n",
+    "\n",
+    "        # Start the scheduler thread\n",
+    "        scheduler_thread = threading.Thread(target=self.run_schedule)\n",
+    "        scheduler_thread.daemon = True\n",
+    "        scheduler_thread.start()\n",
+    "        \n",
+    "camera = cameraTestSuite(camera_ip=camera_ip, username=username, password=password)\n",
+    "camera.main()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "e619e90b",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "zoom=0\n",
+    "experiment_time = 120\n",
+    "\n",
+    "pan_rate_range = range(100, -1, -1)\n",
+    "tilt_rate = 0\n",
+    "camera.zoomCamera(zoom)\n",
+    "results = []\n",
+    "for pan_rate in pan_rate_range:\n",
+    "    #camera.start_tilt(tiltRate=pan_rate)\n",
+    "    camera.start_pan(panRate=pan_rate)\n",
+    "    timestart = time.time()\n",
+    "    while time.time()-timestart < experiment_time:\n",
+    "        time.sleep(0.1)\n",
+    "    camera.stop()\n",
+    "    \n",
+    "    # Calculate pan rate from last 2 measurement\n",
+    "    # Get the last two position measurements\n",
+    "    timestamps = [d['timestamp'] - camera.data[0]['timestamp'] for d in camera.data]\n",
+    "    #pan_values = np.unwrap([float(d['tilt']) for d in camera.data], period=360)\n",
+    "    pan_values = np.unwrap([float(d['pan']) for d in camera.data], period=360)\n",
+    "\n",
+    "    measured_pan_rate = (pan_values[-1]-pan_values[0])/(timestamps[-1]-timestamps[0])\n",
+    "    #print(f\"Measured tilt rate for {pan_rate}%: {measured_pan_rate:.3f} degrees/second\")\n",
+    "    print(f\"Measured pan rate for {pan_rate}%: {measured_pan_rate:.3f} degrees/second\")\n",
+    "    measurement = {\n",
+    "        'pan_rate': pan_rate,\n",
+    "        'measured_pan_rate': measured_pan_rate\n",
+    "    }\n",
+    "    results.append(measurement)\n",
+    "pan_rates = [result['pan_rate'] for result in results]\n",
+    "measured_pan_rates = [result['measured_pan_rate'] for result in results]"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "c9989437",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Save data to CSV\n",
+    "csv_filename = \"pan-rate-analysis.csv\"\n",
+    "with open(csv_filename, mode=\"w\", newline=\"\") as file:\n",
+    "    writer = csv.writer(file)\n",
+    "    writer.writerow([\"Pan Rate (%)\", \"Measured Pan Rate (degrees/second)\"])\n",
+    "    writer.writerows(zip(pan_rates, measured_pan_rates))\n",
+    "\n",
+    "print(f\"Data saved to '{csv_filename}'.\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "a45bfba6",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "plt.figure(figsize=(10, 6))\n",
+    "plt.plot(pan_rates, measured_pan_rates, marker='o')\n",
+    "plt.xlabel('Pan Rate (%)')\n",
+    "plt.ylabel('Measured Pan Rate (degrees/second)')\n",
+    "plt.title('Measured Pan Rate vs. Pan Rate Setting')\n",
+    "plt.grid(True)"
+   ]
+  }
+ ],
+ "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.10.12"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}