Refactor Python script
This commit is contained in:
+128
-133
@@ -1,8 +1,31 @@
|
||||
"""
|
||||
Automated Photogrammetry Capture Suite
|
||||
=====================================
|
||||
This module automates a 360-degree photography sequence using an Olympus OM-D
|
||||
EM-10 Mk III camera and an Arduino-controlled rotation stand.
|
||||
|
||||
The script synchronizes the two systems by:
|
||||
1. Initializing the camera's internal API to enter shutter mode.
|
||||
2. Establishing a persistent serial connection with the Arduino controller to
|
||||
manage rotation speed and movement.
|
||||
3. Executing a sequenced loop of image capture, delay cycles, and incremental
|
||||
rotation steps.
|
||||
|
||||
Hardware Requirements:
|
||||
-----------------------
|
||||
- Camera: Olympus OM-D EM-10 Mk III (connected via its internal WiFi network).
|
||||
- Stand: Arduino-based rotation table (connected via USB/Serial).
|
||||
|
||||
Usage Note:
|
||||
----------
|
||||
Ensure the Camera's WiFi is active and the script 'CAMERA_IP' matches its
|
||||
assigned address on your network. Ensure the correct 'ARDUINO_PORT' is mapped
|
||||
to your local system (e.g., /dev/ttyUSB0 or COMx).
|
||||
"""
|
||||
|
||||
import requests
|
||||
import serial
|
||||
import time
|
||||
import sys
|
||||
import re
|
||||
|
||||
# ================= CONFIGURATION =================
|
||||
CAMERA_IP = "192.168.0.10"
|
||||
@@ -15,167 +38,139 @@ ROTATION_STEP_SIZE = 15.0
|
||||
TOTAL_TARGET_DEGREES = 360.0
|
||||
|
||||
# Timing
|
||||
PHOTO_TO_MOVE_DELAY = 8.0 # Wait after photo before rotation starts
|
||||
SERIAL_TIMEOUT = 10.0 # Increased timeout for long moves
|
||||
PHOTO_TO_MOVE_DELAY = 8.0
|
||||
TIMEOUT = 10.0
|
||||
# ================================================
|
||||
|
||||
def emergency_shutdown(error_msg, ser=None):
|
||||
"""
|
||||
Failsafe routine: Attempts to switch camera to play mode and exits.
|
||||
"""
|
||||
print(f"\n[!] ERROR DETECTED: {error_msg}")
|
||||
print("Initiating failsafe: Switching camera back to default state...")
|
||||
|
||||
try:
|
||||
requests.get(f"http://{CAMERA_IP}/switch_cammode.cgi?mode=play", timeout=5)
|
||||
print("Camera switched to play mode.")
|
||||
except Exception as e:
|
||||
print(f"Failed to switch camera mode during shutdown: {e}")
|
||||
|
||||
if ser:
|
||||
try:
|
||||
ser.close()
|
||||
print("Serial connection closed.")
|
||||
except:
|
||||
class SerialTimeoutError(Exception):
|
||||
pass
|
||||
|
||||
print("System halted.")
|
||||
sys.exit(1)
|
||||
|
||||
def send_arduino_command(ser, command):
|
||||
def initialize_camera():
|
||||
"""
|
||||
Sends a command to the Arduino.
|
||||
Does NOT wait for a response here to keep the flow simple,
|
||||
but the response is consumed in the main loop.
|
||||
Configures the camera's hardware state for photography.
|
||||
"""
|
||||
try:
|
||||
ser.write(f"{command}\n".encode())
|
||||
# Small delay to ensure Arduino receives command before next action
|
||||
time.sleep(0.1)
|
||||
except Exception as e:
|
||||
raise Exception(f"Failed to send command '{command}': {e}")
|
||||
print("[CAMERA] Switching to SHUTTER mode.")
|
||||
requests.get(f"http://{CAMERA_IP}/switch_cammode.cgi?mode=shutter", timeout=TIMEOUT)
|
||||
time.sleep(0.3)
|
||||
|
||||
def wait_for_move_completion(ser, expected_angle, speed_dps):
|
||||
"""
|
||||
Waits until the Arduino reports 'Steps Left: 0'.
|
||||
Falls back to a calculated timeout if serial parsing fails.
|
||||
"""
|
||||
# Calculate expected time for the move as a safety fallback
|
||||
# angle / speed = seconds. Add 10% buffer.
|
||||
expected_time = (expected_angle / speed_dps) * 1.2
|
||||
start_time = time.time()
|
||||
|
||||
print(f"Waiting for movement ({expected_angle}°) to complete...")
|
||||
def initialize_stand(ser):
|
||||
"""
|
||||
Configures the Arduino-controlled rotation stand's parameters.
|
||||
|
||||
This is a blocking function. It sends commands to set the stand into
|
||||
MANUAL mode and defines the rotation speed (deg/sec). It will not return
|
||||
until the Arduino confirms these states have been successfully accepted.
|
||||
|
||||
Args:
|
||||
ser (serial.Serial): The open serial connection to the Arduino controller.
|
||||
"""
|
||||
print("[STAND] Switching to MANUAL mode.")
|
||||
ser.write(f"m1\n".encode())
|
||||
time.sleep(0.3)
|
||||
await_completion(ser, "Mode: MANUAL")
|
||||
|
||||
print(f"[STAND] Setting rotation speed to {ROTATION_SPEED_DPS} degrees/second.")
|
||||
ser.write(f"s{ROTATION_SPEED_DPS}\n".encode())
|
||||
time.sleep(0.3)
|
||||
await_completion(ser, f"Speed: {ROTATION_SPEED_DPS:.0f}")
|
||||
|
||||
|
||||
def take_photo():
|
||||
"""
|
||||
Triggers the camera's shutter and waits for capture completion.
|
||||
"""
|
||||
print("[CAMERA] Activating electronic shutter.")
|
||||
requests.get(f"http://{CAMERA_IP}/exec_shutter.cgi?type=1st2ndpush", timeout=TIMEOUT)
|
||||
time.sleep(0.3)
|
||||
|
||||
requests.get(f"http://{CAMERA_IP}/exec_shutter.cgi?com=2nd1strelease", timeout=TIMEOUT)
|
||||
time.sleep(0.3)
|
||||
|
||||
|
||||
def rotate_stand(ser):
|
||||
"""
|
||||
Commands the stand to perform a specific rotation.
|
||||
|
||||
This is a blocking function. It sends the degrees-to-rotate command
|
||||
and monitors the serial line until the Arduino confirms that the
|
||||
movement has finished or is ready for the next instruction.
|
||||
|
||||
Args:
|
||||
ser (serial.Serial): The open serial connection to the Arduino controller.
|
||||
"""
|
||||
print(f"[STAND] Rotating stand by {ROTATION_STEP_SIZE} degrees.")
|
||||
ser.write(f"a{ROTATION_STEP_SIZE}\n".encode())
|
||||
time.sleep(0.3)
|
||||
await_completion(ser, "Waiting")
|
||||
|
||||
|
||||
def await_completion(ser, s):
|
||||
"""
|
||||
Blocks the execution of the script until a specific confirmation
|
||||
string is detected in the serial output from the Arduino.
|
||||
|
||||
Args:
|
||||
ser (serial.Serial): The open serial connection to the Arduino.
|
||||
s (str): The substring expected in the response to indicate completion.
|
||||
|
||||
Raises:
|
||||
SerialTimeoutError: If no data is received or if a
|
||||
non-matching line is found repeatedly.
|
||||
"""
|
||||
while True:
|
||||
if ser.in_waiting > 0:
|
||||
try:
|
||||
line = ser.readline().decode('utf-8').strip()
|
||||
if not line:
|
||||
continue
|
||||
line_bytes = ser.readline()
|
||||
line_str = line_bytes.decode('utf-8').strip()
|
||||
|
||||
# Check for "Steps Left: 0" in the Arduino status output
|
||||
# Your Arduino prints: " | Steps Left: 0"
|
||||
if "Steps Left: 0" in line:
|
||||
print("Movement complete (Steps Left: 0 detected).")
|
||||
return True
|
||||
if not line_str:
|
||||
raise SerialTimeoutError
|
||||
|
||||
# Optional: Debug print if needed
|
||||
# print(f"Arduino: {line}")
|
||||
except Exception as e:
|
||||
print(f"Warning: Error reading serial: {e}")
|
||||
if s in line_str:
|
||||
break
|
||||
|
||||
# Fallback timeout
|
||||
elapsed = time.time() - start_time
|
||||
if elapsed > max(expected_time + 5, 30): # Max 30s or calculated time + 5s
|
||||
print(f"Timeout reached ({elapsed:.1f}s). Assuming move finished.")
|
||||
return False
|
||||
time.sleep(0.1)
|
||||
|
||||
time.sleep(0.1) # Prevent CPU spinning
|
||||
|
||||
def main():
|
||||
ser = None
|
||||
with serial.Serial(ARDUINO_PORT, BAUD_RATE, timeout=TIMEOUT) as ser:
|
||||
try:
|
||||
# 1. Initialize Serial Connection (Persistent)
|
||||
print("Opening serial port...")
|
||||
ser = serial.Serial(ARDUINO_PORT, BAUD_RATE, timeout=1)
|
||||
time.sleep(2) # Wait for Arduino to reset and boot
|
||||
time.sleep(2.0) # Buffer to account for Arduino startup
|
||||
|
||||
# Flush any garbage data from boot
|
||||
print("Flushing serial buffers.")
|
||||
ser.reset_input_buffer()
|
||||
ser.reset_output_buffer()
|
||||
|
||||
# 2. Initialize Table Parameters
|
||||
print("Initializing table parameters...")
|
||||
send_arduino_command(ser, f"s{ROTATION_SPEED_DPS}")
|
||||
time.sleep(0.5)
|
||||
send_arduino_command(ser, "m1") # Set to Manual mode (controlled by script)
|
||||
time.sleep(0.5)
|
||||
|
||||
# 3. Prepare Camera State
|
||||
print("Switching camera to shutter mode...")
|
||||
requests.get(f"http://{CAMERA_IP}/switch_cammode.cgi?mode=shutter", timeout=5)
|
||||
print("System ready.")
|
||||
initialize_stand(ser)
|
||||
initialize_camera()
|
||||
time.sleep(2.0) # Buffer to ensure the display is not moving
|
||||
|
||||
current_angle = 0.0
|
||||
total_steps = int(TOTAL_TARGET_DEGREES / ROTATION_STEP_SIZE)
|
||||
|
||||
# Ensure we cover the full circle if 360 is the target
|
||||
# If 360/15 = 24 steps, we do 24 moves.
|
||||
# The loop runs while current_angle < TOTAL (0 to 345), then we do one last check?
|
||||
# Actually, simpler: Loop for exactly the number of steps needed.
|
||||
|
||||
steps_taken = 0
|
||||
while steps_taken <= total_steps:
|
||||
# A. Capture Image
|
||||
angle_label = int(current_angle)
|
||||
print(f"\n--- Step {steps_taken + 1}/{total_steps + 1} ---")
|
||||
print(f"Capturing image at {angle_label}°...")
|
||||
|
||||
try:
|
||||
# 1. Push Shutter (Start exposure)
|
||||
req1 = requests.get(f"http://{CAMERA_IP}/exec_shutter.cgi?type=1st2ndpush", timeout=10)
|
||||
time.sleep(0.1)
|
||||
|
||||
# 2. Release Shutter (End exposure) - CRITICAL STEP
|
||||
# The user specified this command is required to actually take the photo
|
||||
req2 = requests.get(f"http://{CAMERA_IP}/exec_shutter.cgi?com=2nd1strelease", timeout=10)
|
||||
|
||||
if req1.status_code != 200 or req2.status_code != 200:
|
||||
raise Exception(f"Camera returned status {req1.status_code} / {req2.status_code}")
|
||||
|
||||
print("Image captured successfully.")
|
||||
except Exception as e:
|
||||
raise Exception(f"Camera shutter sequence failed: {e}")
|
||||
|
||||
# If this is the last step, we don't need to move or wait
|
||||
if steps_taken == total_steps:
|
||||
print("All steps completed.")
|
||||
break
|
||||
|
||||
# B. Wait for camera processing
|
||||
print(f"Waiting {PHOTO_TO_MOVE_DELAY}s for camera processing...")
|
||||
time.sleep(PHOTO_TO_MOVE_DELAY)
|
||||
|
||||
# C. Rotate Table
|
||||
print(f"Rotating {ROTATION_STEP_SIZE}°...")
|
||||
send_arduino_command(ser, f"a{ROTATION_STEP_SIZE}")
|
||||
|
||||
# D. Wait for movement to finish
|
||||
# We pass the step size and speed to calculate a safe wait time
|
||||
wait_for_move_completion(ser, ROTATION_STEP_SIZE, ROTATION_SPEED_DPS)
|
||||
while current_angle < TOTAL_TARGET_DEGREES:
|
||||
take_photo()
|
||||
time.sleep(PHOTO_TO_MOVE_DELAY) # Buffer to account for shutter speed
|
||||
rotate_stand(ser)
|
||||
|
||||
current_angle += ROTATION_STEP_SIZE
|
||||
steps_taken += 1
|
||||
print(f"Progress: {int(current_angle)}/{TOTAL_TARGET_DEGREES} degrees.")
|
||||
|
||||
print("\n[SUCCESS] Full rotation completed successfully.")
|
||||
except requests.Timeout:
|
||||
print("Connection to camera API lost.")
|
||||
|
||||
except SerialTimeoutError:
|
||||
print("Arduino failed to respond in time.")
|
||||
|
||||
except Exception as e:
|
||||
emergency_shutdown(str(e), ser)
|
||||
finally:
|
||||
if ser and ser.is_open:
|
||||
ser.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
|
||||
# Send `m1` command
|
||||
# Response: "Mode: MANUAL | State: WAIT | Steps Left: 0 | Speed: 45.00 deg/s | Wait: 8.00s"
|
||||
|
||||
# Send `s45` command
|
||||
# Response: "Mode: MANUAL | State: WAIT | Steps Left: 0 | Speed: 45.00 deg/s | Wait: 8.00s"
|
||||
|
||||
# Send `a15` command
|
||||
# Response: "Waiting indefinitely for new command (MANUAL mode)."
|
||||
+182
@@ -0,0 +1,182 @@
|
||||
|
||||
import requests
|
||||
import serial
|
||||
import time
|
||||
import sys
|
||||
import re
|
||||
|
||||
# ================= CONFIGURATION =================
|
||||
CAMERA_IP = "192.168.0.10"
|
||||
ARDUINO_PORT = "/dev/ttyUSB0"
|
||||
BAUD_RATE = 9600
|
||||
|
||||
# Motion Control
|
||||
ROTATION_SPEED_DPS = 45.0
|
||||
ROTATION_STEP_SIZE = 15.0
|
||||
TOTAL_TARGET_DEGREES = 360.0
|
||||
|
||||
# Timing
|
||||
PHOTO_TO_MOVE_DELAY = 8.0 # Wait after photo before rotation starts
|
||||
SERIAL_TIMEOUT = 10.0 # Increased timeout for long moves
|
||||
# ================================================
|
||||
|
||||
def emergency_shutdown(error_msg, ser=None):
|
||||
"""
|
||||
Failsafe routine: Attempts to switch camera to play mode and exits.
|
||||
"""
|
||||
print(f"\n[!] ERROR DETECTED: {error_msg}")
|
||||
print("Initiating failsafe: Switching camera back to default state...")
|
||||
|
||||
try:
|
||||
requests.get(f"http://{CAMERA_IP}/switch_cammode.cgi?mode=play", timeout=5)
|
||||
print("Camera switched to play mode.")
|
||||
except Exception as e:
|
||||
print(f"Failed to switch camera mode during shutdown: {e}")
|
||||
|
||||
if ser:
|
||||
try:
|
||||
ser.close()
|
||||
print("Serial connection closed.")
|
||||
except:
|
||||
pass
|
||||
|
||||
print("System halted.")
|
||||
sys.exit(1)
|
||||
|
||||
def send_arduino_command(ser, command):
|
||||
"""
|
||||
Sends a command to the Arduino.
|
||||
Does NOT wait for a response here to keep the flow simple,
|
||||
but the response is consumed in the main loop.
|
||||
"""
|
||||
try:
|
||||
ser.write(f"{command}\n".encode())
|
||||
# Small delay to ensure Arduino receives command before next action
|
||||
time.sleep(0.1)
|
||||
except Exception as e:
|
||||
raise Exception(f"Failed to send command '{command}': {e}")
|
||||
|
||||
def wait_for_move_completion(ser, expected_angle, speed_dps):
|
||||
"""
|
||||
Waits until the Arduino reports 'Steps Left: 0'.
|
||||
Falls back to a calculated timeout if serial parsing fails.
|
||||
"""
|
||||
# Calculate expected time for the move as a safety fallback
|
||||
# angle / speed = seconds. Add 10% buffer.
|
||||
expected_time = (expected_angle / speed_dps) * 1.2
|
||||
start_time = time.time()
|
||||
|
||||
print(f"Waiting for movement ({expected_angle}°) to complete...")
|
||||
|
||||
while True:
|
||||
if ser.in_waiting > 0:
|
||||
try:
|
||||
line = ser.readline().decode('utf-8').strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
# Check for "Steps Left: 0" in the Arduino status output
|
||||
# Your Arduino prints: " | Steps Left: 0"
|
||||
if "Steps Left: 0" in line:
|
||||
print("Movement complete (Steps Left: 0 detected).")
|
||||
return True
|
||||
|
||||
# Optional: Debug print if needed
|
||||
# print(f"Arduino: {line}")
|
||||
except Exception as e:
|
||||
print(f"Warning: Error reading serial: {e}")
|
||||
|
||||
# Fallback timeout
|
||||
elapsed = time.time() - start_time
|
||||
if elapsed > max(expected_time + 5, 30): # Max 30s or calculated time + 5s
|
||||
print(f"Timeout reached ({elapsed:.1f}s). Assuming move finished.")
|
||||
return False
|
||||
|
||||
time.sleep(0.1) # Prevent CPU spinning
|
||||
|
||||
def main():
|
||||
ser = None
|
||||
try:
|
||||
# 1. Initialize Serial Connection (Persistent)
|
||||
print("Opening serial port...")
|
||||
ser = serial.Serial(ARDUINO_PORT, BAUD_RATE, timeout=1)
|
||||
time.sleep(2) # Wait for Arduino to reset and boot
|
||||
|
||||
# Flush any garbage data from boot
|
||||
ser.reset_input_buffer()
|
||||
ser.reset_output_buffer()
|
||||
|
||||
# 2. Initialize Table Parameters
|
||||
print("Initializing table parameters...")
|
||||
send_arduino_command(ser, f"s{ROTATION_SPEED_DPS}")
|
||||
time.sleep(0.5)
|
||||
send_arduino_command(ser, "m1") # Set to Manual mode (controlled by script)
|
||||
time.sleep(0.5)
|
||||
|
||||
# 3. Prepare Camera State
|
||||
print("Switching camera to shutter mode...")
|
||||
requests.get(f"http://{CAMERA_IP}/switch_cammode.cgi?mode=shutter", timeout=5)
|
||||
print("System ready.")
|
||||
|
||||
current_angle = 0.0
|
||||
total_steps = int(TOTAL_TARGET_DEGREES / ROTATION_STEP_SIZE)
|
||||
|
||||
# Ensure we cover the full circle if 360 is the target
|
||||
# If 360/15 = 24 steps, we do 24 moves.
|
||||
# The loop runs while current_angle < TOTAL (0 to 345), then we do one last check?
|
||||
# Actually, simpler: Loop for exactly the number of steps needed.
|
||||
|
||||
steps_taken = 0
|
||||
while steps_taken <= total_steps:
|
||||
# A. Capture Image
|
||||
angle_label = int(current_angle)
|
||||
print(f"\n--- Step {steps_taken + 1}/{total_steps + 1} ---")
|
||||
print(f"Capturing image at {angle_label}°...")
|
||||
|
||||
try:
|
||||
# 1. Push Shutter (Start exposure)
|
||||
req1 = requests.get(f"http://{CAMERA_IP}/exec_shutter.cgi?type=1st2ndpush", timeout=10)
|
||||
time.sleep(0.1)
|
||||
|
||||
# 2. Release Shutter (End exposure) - CRITICAL STEP
|
||||
# The user specified this command is required to actually take the photo
|
||||
req2 = requests.get(f"http://{CAMERA_IP}/exec_shutter.cgi?com=2nd1strelease", timeout=10)
|
||||
|
||||
if req1.status_code != 200 or req2.status_code != 200:
|
||||
raise Exception(f"Camera returned status {req1.status_code} / {req2.status_code}")
|
||||
|
||||
print("Image captured successfully.")
|
||||
except Exception as e:
|
||||
raise Exception(f"Camera shutter sequence failed: {e}")
|
||||
|
||||
# If this is the last step, we don't need to move or wait
|
||||
if steps_taken == total_steps:
|
||||
print("All steps completed.")
|
||||
break
|
||||
|
||||
# B. Wait for camera processing
|
||||
print(f"Waiting {PHOTO_TO_MOVE_DELAY}s for camera processing...")
|
||||
time.sleep(PHOTO_TO_MOVE_DELAY)
|
||||
|
||||
# C. Rotate Table
|
||||
print(f"Rotating {ROTATION_STEP_SIZE}°...")
|
||||
send_arduino_command(ser, f"a{ROTATION_STEP_SIZE}")
|
||||
|
||||
# D. Wait for movement to finish
|
||||
# We pass the step size and speed to calculate a safe wait time
|
||||
wait_for_move_completion(ser, ROTATION_STEP_SIZE, ROTATION_SPEED_DPS)
|
||||
|
||||
current_angle += ROTATION_STEP_SIZE
|
||||
steps_taken += 1
|
||||
print(f"Progress: {int(current_angle)}/{TOTAL_TARGET_DEGREES} degrees.")
|
||||
|
||||
print("\n[SUCCESS] Full rotation completed successfully.")
|
||||
|
||||
except Exception as e:
|
||||
emergency_shutdown(str(e), ser)
|
||||
finally:
|
||||
if ser and ser.is_open:
|
||||
ser.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user