73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from apscheduler.triggers.interval import IntervalTrigger
|
|
import requests
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
import re
|
|
|
|
scheduler = BackgroundScheduler()
|
|
jobs = {}
|
|
|
|
def capture_image(project_id, ip_address, port, username, password, capture_folder):
|
|
"""Capture an image from the camera"""
|
|
try:
|
|
# Build URL with optional auth
|
|
if username and password:
|
|
url = f"http://{username}:{password}@{ip_address}:{port}/axis-cgi/jpg/image.cgi"
|
|
else:
|
|
url = f"http://{ip_address}:{port}/axis-cgi/jpg/image.cgi"
|
|
|
|
# Create timestamped filename
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
safe_name = re.sub(r'[^a-zA-Z0-9_-]', '_', capture_folder)
|
|
folder_path = Path(f"captures/{safe_name}")
|
|
folder_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
output_path = folder_path / f"{timestamp}.jpg"
|
|
|
|
# Fetch image
|
|
response = requests.get(url, timeout=10)
|
|
response.raise_for_status()
|
|
|
|
# Save image
|
|
with open(output_path, 'wb') as f:
|
|
f.write(response.content)
|
|
|
|
print(f"Captured image for project {project_id}: {output_path}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Failed to capture image for project {project_id}: {e}")
|
|
return False
|
|
|
|
def start_job(project_id, ip_address, port, period_minutes, username, password, capture_folder):
|
|
"""Start a scheduled capture job"""
|
|
# Remove existing job if any
|
|
stop_job(project_id)
|
|
|
|
# Add new job
|
|
job = scheduler.add_job(
|
|
capture_image,
|
|
trigger=IntervalTrigger(minutes=period_minutes),
|
|
id=f"project_{project_id}",
|
|
args=[project_id, ip_address, port, username, password, capture_folder],
|
|
replace_existing=True
|
|
)
|
|
jobs[project_id] = job
|
|
print(f"Started job for project {project_id} (every {period_minutes} minutes)")
|
|
|
|
def stop_job(project_id):
|
|
"""Stop a scheduled capture job"""
|
|
job_id = f"project_{project_id}"
|
|
if scheduler.get_job(job_id):
|
|
scheduler.remove_job(job_id)
|
|
print(f"Stopped job for project {project_id}")
|
|
if project_id in jobs:
|
|
del jobs[project_id]
|
|
|
|
def start_all_jobs(projects):
|
|
"""Start jobs for all active projects"""
|
|
for p in projects:
|
|
project_id, name, ip, port, period, username, password, created, last_capture, active = p
|
|
start_job(project_id, ip, port, period, username, password, name)
|