Take Photos with Specified Exposure Settings

.github/workflows/pylint.yml

@@ -7,7 +7,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.13"]
+        python-version: ["3.9"]
     steps:
     - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}

.gitignore

@@ -205,3 +205,4 @@ cython_debug/
 marimo/_static/
 marimo/_lsp/
 __marimo__/
+.vscode/settings.json

theta_capture.py

@@ -3,16 +3,10 @@
 import time
 import datetime
 
-import requests
+import theta_wireless as tw
+from theta_wireless import Options
+import theta_init
 
-# pylint: disable=duplicate-code
-
-# カメラIP
-THETA_IP = "192.168.1.1"
-EXECUTE_URL = f"http://{THETA_IP}/osc/commands/execute"
-STATUS_URL = f"http://{THETA_IP}/osc/commands/status"
-
-HEADERS = {"Content-Type": "application/json;charset=utf-8"}
 
 # 撮影設定リスト (ISO, shutter_speed, ColorTemperature)
 settings_list = [
@@ -33,38 +27,28 @@
 
 def set_options(iso, shutter_speed, white_balance):
     """オプションを設定"""
-    options_command = {
-        "name": "camera.setOptions",
-        "parameters": {
-            "options": {
-                "iso": iso,
-                "shutterSpeed": shutter_speed,
-                "whiteBalance": "_colorTemperature",
-                "colorTemperature": white_balance,
-            }
-        },
-    }
-    resp = requests.post(EXECUTE_URL, json=options_command, headers=HEADERS, timeout=10)
-    resp.raise_for_status()
-    return resp.json()
-
-
-def take_picture():
-    """Theta Web APIを用いて撮影"""
-    take_command = {"name": "camera.takePicture"}
-    resp = requests.post(EXECUTE_URL, json=take_command, headers=HEADERS, timeout=10)
-    resp.raise_for_status()
-    return resp.json()
+    tw.set_options(
+        {
+            Options.ISO:iso,
+            Options.SHUTTER_SPEED: shutter_speed,
+            Options.WHITE_BALANCE: Options.COLOR_TEMPERATURE,
+            Options.COLOR_TEMPERATURE: white_balance
+        }
+    )
+    resp = tw.get_options(
+        {
+            Options.ISO,
+            Options.SHUTTER_SPEED,
+            Options.COLOR_TEMPERATURE
+        }
+    )
+    return resp
 
 
 def wait_for_completion(command_id):
     """撮影完了まで処理を停止"""
     while True:
-        status_resp = requests.post(
-            STATUS_URL, json={"id": command_id}, headers=HEADERS, timeout=10
-        )
-        status_resp.raise_for_status()
-        status = status_resp.json()
+        status = tw.check_status(command_id=command_id)
         if status.get("state") == "done":
             return status.get("results")
         time.sleep(0.5)
@@ -77,7 +61,7 @@ def capture_12():
         set_options(**s)
         print("設定完了:", s)
 
-        result = take_picture()
+        result = tw.take_picture()
         print("撮影コマンド送信:", result)
 
         if "id" in result:
@@ -119,5 +103,11 @@ def schedule_shoots():
             time.sleep(wait_sec / 60)
 
 
-if __name__ == "__main__":
+def main():
+    """メイン関数"""
+    theta_init.init()
     schedule_shoots()
+
+
+if __name__ == "__main__":
+    main()

theta_init.py

@@ -1,24 +1,34 @@
-"""Script for controlling RICOH THETA camera."""
+"""Thetaカメラを初期化します"""
 
-import requests
+import theta_wireless
+from theta_wireless import Options
 
-# pylint: disable=duplicate-code
+def init():
+    """Thetaの初期化処理を行います。"""
+    try:
+        theta_wireless.set_options(
+            {
+                Options.CAPTURE_MODE:"image",
+                Options.EXPOSURE_PROGRAM:1,
+                Options.SLEEP_DELAY:65535,
+                Options.SHUTTER_VOLUME:0
+            }
+        )
+        response = theta_wireless.get_options(
+            {
+                Options.CAPTURE_MODE,
+                Options.EXPOSURE_PROGRAM,
+                Options.SLEEP_DELAY,
+                Options.SHUTTER_VOLUME
+            }
+        )
+        print(response)
 
-URL = "http://192.168.1.1/osc/commands/execute"
-HEADERS = {"Content-Type": "application/json;charset=utf-8"}
+    except theta_wireless.CameraInternalError as e:
+        print(f"Caught OSC API error: {e.code} -> {e.message}")
 
-payload = {
-    "name": "camera.setOptions",
-    "parameters": {"options": {"captureMode": "image"}},
-}
+    except theta_wireless.CameraTimeout as e:
+        print(f"Timeoutしました: {e}")
 
-resp = requests.post(URL, json=payload, headers=HEADERS, timeout=10)
-print(resp.json())
-
-payload = {
-    "name": "camera.setOptions",
-    "parameters": {"options": {"exposureProgram": 1}},
-}
-
-resp = requests.post(URL, json=payload, headers=HEADERS, timeout=10)
-print(resp.json())
+if __name__ == "__main__":
+    init()

theta_test_capture.py

@@ -2,16 +2,11 @@
 
 import time
 
-import requests
+import theta_wireless as tw
+from theta_wireless import Options
 
 # pylint: disable=duplicate-code
 
-# カメラIP
-THETA_IP = "192.168.1.1"
-EXECUTE_URL = f"http://{THETA_IP}/osc/commands/execute"
-STATUS_URL = f"http://{THETA_IP}/osc/commands/status"
-
-HEADERS = {"Content-Type": "application/json;charset=utf-8"}
 
 # 撮影設定リスト (ISO, shutter_speed, f, ColorTemperature)
 settings_list = [
@@ -32,37 +27,28 @@
 
 def set_options(iso, shutter_speed, white_balance):
     """オプションを設定"""
-    options_command = {
-        "name": "camera.setOptions",
-        "parameters": {
-            "options": {
-                "iso": iso,
-                "shutterSpeed": shutter_speed,
-                "_colorTemperature": white_balance,
-            }
-        },
-    }
-    resp = requests.post(EXECUTE_URL, json=options_command, headers=HEADERS, timeout=10)
-    resp.raise_for_status()
-    return resp.json()
-
-
-def take_picture():
-    """Take a picture using the Theta API"""
-    take_command = {"name": "camera.takePicture"}
-    resp = requests.post(EXECUTE_URL, json=take_command, headers=HEADERS, timeout=10)
-    resp.raise_for_status()
-    return resp.json()
+    tw.set_options(
+        {
+            Options.ISO:iso,
+            Options.SHUTTER_SPEED: shutter_speed,
+            Options.WHITE_BALANCE: Options.COLOR_TEMPERATURE,
+            Options.COLOR_TEMPERATURE: white_balance
+        }
+    )
+    resp = tw.get_options(
+        {
+            Options.ISO,
+            Options.SHUTTER_SPEED,
+            Options.COLOR_TEMPERATURE
+        }
+    )
+    return resp
 
 
 def wait_for_completion(command_id):
     """Wait a few seconds to complete"""
     while True:
-        status_resp = requests.post(
-            STATUS_URL, json={"id": command_id}, headers=HEADERS, timeout=10
-        )
-        status_resp.raise_for_status()
-        status = status_resp.json()
+        status = tw.check_status(command_id=command_id)
         if status.get("state") == "done":
             return status.get("results")
         time.sleep(0.5)
@@ -74,7 +60,7 @@ def wait_for_completion(command_id):
     set_options(**s)
     print("設定完了:", s)
 
-    result = take_picture()
+    result = tw.take_picture()
     print("撮影コマンド送信:", result)
 
     if "id" in result: