59 lines
2.3 KiB
Python
59 lines
2.3 KiB
Python
from pathlib import Path
|
|
|
|
from celery import shared_task
|
|
from django.utils import timezone
|
|
|
|
from .algorithms import ProcessingError, average_images, histogram, subtract_images
|
|
from .models import ImageSession, ProcessingJob
|
|
from .storage import delete_relative_file, load_image_array, save_image_array
|
|
|
|
|
|
@shared_task(bind=True)
|
|
def run_batch_job(self, job_id, operation, session_ids):
|
|
job = ProcessingJob.objects.get(id=job_id)
|
|
job.status = ProcessingJob.STATUS_RUNNING
|
|
job.progress = 10
|
|
job.save(update_fields=["status", "progress", "updated_at"])
|
|
try:
|
|
sessions = list(ImageSession.objects.filter(id__in=session_ids))
|
|
if len(sessions) != len(session_ids):
|
|
raise ProcessingError("One or more sessions do not exist.")
|
|
images = [load_image_array(session.processed_image or session.original_image) for session in sessions]
|
|
job.progress = 45
|
|
job.save(update_fields=["progress", "updated_at"])
|
|
if operation == "average":
|
|
result = average_images(images)
|
|
elif operation == "subtract":
|
|
result = subtract_images(images[0], images[1])
|
|
else:
|
|
raise ProcessingError(f"Unsupported batch operation '{operation}'.")
|
|
relative_path = save_image_array(result, f"batch-{operation}")
|
|
job.status = ProcessingJob.STATUS_COMPLETE
|
|
job.progress = 100
|
|
job.result_image = relative_path
|
|
job.result_histogram = histogram(result)
|
|
job.error = ""
|
|
job.save(update_fields=["status", "progress", "result_image", "result_histogram", "error", "updated_at"])
|
|
except Exception as exc:
|
|
job.status = ProcessingJob.STATUS_FAILED
|
|
job.error = str(exc)
|
|
job.progress = 100
|
|
job.save(update_fields=["status", "error", "progress", "updated_at"])
|
|
raise
|
|
|
|
|
|
@shared_task
|
|
def cleanup_expired_sessions():
|
|
expired = ImageSession.objects.filter(expires_at__lt=timezone.now())
|
|
for session in expired:
|
|
delete_relative_file(session.original_image)
|
|
delete_relative_file(session.processed_image)
|
|
for state in session.states.all():
|
|
delete_relative_file(state.image)
|
|
expired.delete()
|
|
|
|
old_jobs = ProcessingJob.objects.filter(created_at__lt=timezone.now() - timezone.timedelta(hours=24))
|
|
for job in old_jobs:
|
|
delete_relative_file(job.result_image)
|
|
old_jobs.delete()
|