fixes and improvements for the unified data model migration (be9bec3a) · Commits · trapper-project / trapper

trapper/trapper-project/trapper/apps/media_classification/migrations/0094_unified_classification_model_move_data.py

+517 −231

File changed.

Preview size limit exceeded, changes collapsed.

trapper/trapper-project/trapper/apps/media_classification/migrations/0095_object_frame_observation_model.py

+4 −20

Original line number	Diff line number	Diff line
		@@ -94,24 +94,8 @@ class Migration(migrations.Migration):
		),
		reverse_sql=migrations.RunSQL.noop,
		),
		# Enable native compression: order by time, segment by object identifiers for better locality
		migrations.RunSQL(
		sql=(
		"ALTER TABLE media_classification_objectframeobservation SET ("
		" timescaledb.compress,"
		" timescaledb.compress_orderby='ts',"
		" timescaledb.compress_segmentby='project_id,classification_id,dynamic_attrs_id'"
		");"
		),
		reverse_sql=migrations.RunSQL.noop,
		),
		# Add an automatic compression policy (tune interval as needed)
		migrations.RunSQL(
		sql=(
		"SELECT add_compression_policy('media_classification_objectframeobservation', INTERVAL '30 days')"
		" WHERE NOT EXISTS (SELECT 1 FROM timescaledb_information.jobs j JOIN timescaledb_information.job_stats s ON j.job_id=s.job_id"
		" WHERE j.proc_name='policy_compression' AND j.hypertable_name='media_classification_objectframeobservation');"
		),
		reverse_sql=migrations.RunSQL.noop,
		),
		# NOTE: Compression settings and policy are added in migration 0096b
		# AFTER the data migration completes. This prevents the compression job
		# from trying to compress chunks while we're still inserting historical data,
		# which would cause severe performance degradation.
		]

trapper/trapper-project/trapper/apps/media_classification/migrations/0096_migrate_bboxes_to_timeseries.py

+229 −59

Original line number	Diff line number	Diff line
		import logging

		from django.db import migrations
		import time
		from datetime import timedelta
		from numbers import Number

		from django.db import migrations, models

		from trapper.apps.storage.taxonomy import ResourceType
		from tqdm import tqdm

		logger = logging.getLogger("migrations")
		logger = logging.getLogger("main_stdout")

		# Batch size for accumulating rows before bulk insert to hypertable
		INSERT_BATCH_SIZE = 50000
		# Batch size for bulk updating flattened bboxes
		UPDATE_BATCH_SIZE = 5000
		# Chunk size for Django's iterator (server-side cursor)
		ITERATOR_CHUNK_SIZE = 2000
		# Log progress every N records processed
		LOG_PROGRESS_EVERY = 50000


		def _is_numeric(value):
		if isinstance(value, complex):
		return False
		return isinstance(value, Number)

		def _iter_dynamic_attrs(qs, chunk_size=1000):
		start = 0
		while True:
		batch = list(qs[start : start + chunk_size])
		if not batch:
		break
		for obj in batch:
		yield obj
		start += chunk_size

		def _sanitize_bbox(entry):
		"""Return bbox as list[float] if entry represents a valid bbox."""
		if not isinstance(entry, (list, tuple)):
		return None
		if len(entry) != 4:
		return None

		cleaned = []
		for value in entry:
		if _is_numeric(value):
		cleaned.append(float(value))
		continue
		try:
		cleaned.append(float(value))
		except (TypeError, ValueError):
		return None
		return cleaned


		def _normalize_bboxes(bboxes):
		"""
		Normalize bboxes into iterator of (frame_index, [x,y,w,h]). Skip empty items.
		Accepts shapes: [x,y,w,h] or [[x,y,w,h], ...] or list with None entries for empty frames.
		Normalize bboxes into list of (frame_index, [x,y,w,h]). Skip empty items.
		Accepts shapes: [x,y,w,h] or [[x,y,w,h], ...] or list with None entries.
		"""
		if not bboxes:
		return []
		if isinstance(bboxes, (list, tuple)):
		if len(bboxes) == 4 and all(isinstance(v, (int, float)) for v in bboxes):
		return [(0, bboxes)]
		# list of per-frame entries
		out = []
		for idx, item in enumerate(bboxes):
		if not item or not isinstance(item, (list, tuple)):
		@@ -41,45 +65,77 @@ def _normalize_bboxes(bboxes):
		return []


		def _determine_fps(classification, resource, is_video: bool = False):
		"""Determine FPS or raise for multi-frame (video-like) data when unknown.
		def _extract_first_bbox_and_index(bboxes):
		"""Return (bbox, frame_index) for first non-null bbox entry."""
		if not bboxes:
		return None, None

		- Returns None for single-frame images.
		- Returns float FPS for videos.
		- Raises ValueError if is_video is True and FPS cannot be determined.
		"""
		bbox = _sanitize_bbox(bboxes)
		if bbox:
		return bbox, 0

		if isinstance(bboxes, (list, tuple)):
		for idx, entry in enumerate(bboxes):
		if not entry:
		continue
		bbox = _sanitize_bbox(entry)
		if bbox:
		return bbox, idx

		return None, None


		def _determine_fps(classification, resource, is_video: bool = False):
		"""Determine FPS or raise for multi-frame (video-like) data when unknown."""
		fps = getattr(classification.project, "target_fps", None) or getattr(
		resource, "file_fps", None
		)
		try:
		if fps is not None:
		fps_val = float(fps)
		else:
		fps_val = None
		fps_val = float(fps) if fps is not None else None
		except Exception:
		fps_val = None

		if not is_video:
		# Image or single-frame data does not require FPS
		return None if not fps_val else fps_val
		return fps_val

		# Video/multi-frame requires FPS
		if not fps_val or fps_val <= 0:
		raise ValueError(
		"FPS is required for multi-frame data but could not be determined from "
		"project.target_fps or media file."
		"FPS is required for multi-frame data but could not be determined"
		)
		return fps_val


		def _compute_ts(base_ts, frame_idx, fps):
		if not fps or fps <= 0:
		# images or unknown fps; only frame 0 is meaningful
		return base_ts
		return base_ts + timedelta(seconds=(frame_idx / fps))


		def _flush_hypertable_batch(ObjectFrameObservation, rows):
		"""Sort rows by timestamp and bulk insert for better chunk locality."""
		if not rows:
		return 0
		rows.sort(key=lambda r: r.ts)
		ObjectFrameObservation.objects.bulk_create(rows, batch_size=10000)
		return len(rows)


		def _flush_update_batch(ClassificationDynamicAttrs, batch):
		"""Bulk update flattened bboxes and first_frame_index."""
		if batch:
		ClassificationDynamicAttrs.objects.bulk_update(
		batch, ["bboxes", "first_frame_index"], batch_size=2000
		)


		def forwards(apps, schema_editor):
		"""
		Combined migration that:
		1. Migrates bbox data to ObjectFrameObservation hypertable
		2. Flattens bboxes to single bbox and sets first_frame_index

		This is done in a single pass to avoid iterating millions of records twice.
		"""
		ObjectFrameObservation = apps.get_model(
		"media_classification", "ObjectFrameObservation"
		)
		@@ -87,35 +143,60 @@ def forwards(apps, schema_editor):
		"media_classification", "ClassificationDynamicAttrs"
		)

		# Migrate bboxes for BOTH AI and USER unified versions
		qs = ClassificationDynamicAttrs.objects.exclude(bboxes=None).select_related(
		# Process ALL dynamic attrs (not just those with bboxes) for the flattening
		# But only create hypertable rows for those with valid bboxes
		qs = ClassificationDynamicAttrs.objects.select_related(
		"classification", "classification__resource", "classification__project"
		)

		total = qs.count()
		bar = tqdm(total=total, desc="Migrating unified bboxes")
		for dyn in _iter_dynamic_attrs(qs):
		logger.info("=" * 70)
		logger.info("Migration 0096: Migrate bboxes to hypertable + flatten bboxes")
		logger.info("=" * 70)
		logger.info(f"Total ClassificationDynamicAttrs to process: {total}")

		start_time = time.time()

		# Stats
		stats = {
		"processed": 0,
		"hypertable_rows": 0,
		"hypertable_skipped": 0,
		"bbox_updated": 0,
		"bbox_cleared": 0,
		}

		# Batches
		hypertable_rows = []
		update_batch = []
		last_logged_at = 0

		for dyn in qs.iterator(chunk_size=ITERATOR_CHUNK_SIZE):
		stats["processed"] += 1
		original_bboxes = dyn.bboxes

		# --- Part 1: Migrate to hypertable (only if valid bbox data) ---
		if original_bboxes is not None:
		cls = dyn.classification
		if cls is not None:
		resource = cls.resource
		project_id = cls.project_id
		base_ts = cls.date_recorded
		entries = _normalize_bboxes(dyn.bboxes)
		if resource is not None and cls.date_recorded is not None:
		entries = _normalize_bboxes(original_bboxes)
		if entries:
		is_video = resource.resource_type == ResourceType.TYPE_VIDEO

		try:
		fps = _determine_fps(cls, resource, is_video=is_video)
		except ValueError as e:
		logger.warning(f"Skipping classification_id={cls.id} due to error: {e}")
		bar.update(1)
		continue

		rows = []
		confidence = (
		float(dyn.classification_confidence)
		if dyn.classification_confidence is not None
		else None
		)
		for frame_idx, bbox in entries:
		x, y, w, h = bbox
		ts = _compute_ts(base_ts, frame_idx, fps)
		rows.append(
		ts = _compute_ts(cls.date_recorded, frame_idx, fps)
		hypertable_rows.append(
		ObjectFrameObservation(
		project_id=project_id,
		project_id=cls.project_id,
		classification_id=cls.id,
		frame_index=frame_idx,
		ts=ts,
		@@ -123,22 +204,88 @@ def forwards(apps, schema_editor):
		y=y,
		w=w,
		h=h,
		confidence=(
		float(dyn.classification_confidence)
		if dyn.classification_confidence is not None
		else None
		),
		confidence=confidence,
		dynamic_attrs_id=dyn.id,
		)
		)
		if rows:
		ObjectFrameObservation.objects.bulk_create(rows, batch_size=10000)
		bar.update(1)
		bar.close()
		except ValueError:
		stats["hypertable_skipped"] += 1
		else:
		stats["hypertable_skipped"] += 1
		else:
		stats["hypertable_skipped"] += 1
		else:
		stats["hypertable_skipped"] += 1

		# --- Part 2: Flatten bbox and set first_frame_index ---
		first_bbox, frame_index = _extract_first_bbox_and_index(original_bboxes)
		new_frame_index = int(frame_index) if frame_index is not None else None

		changed = False

		if first_bbox is None:
		if dyn.bboxes is not None:
		dyn.bboxes = None
		changed = True
		stats["bbox_cleared"] += 1
		else:
		if dyn.bboxes != first_bbox:
		dyn.bboxes = first_bbox
		changed = True

		if new_frame_index != dyn.first_frame_index:
		dyn.first_frame_index = new_frame_index
		changed = True

		if changed:
		update_batch.append(dyn)
		stats["bbox_updated"] += 1

		# Flush hypertable batch
		if len(hypertable_rows) >= INSERT_BATCH_SIZE:
		stats["hypertable_rows"] += _flush_hypertable_batch(
		ObjectFrameObservation, hypertable_rows
		)
		hypertable_rows = []

		# Flush update batch
		if len(update_batch) >= UPDATE_BATCH_SIZE:
		_flush_update_batch(ClassificationDynamicAttrs, update_batch)
		update_batch = []

		# Log progress
		if stats["processed"] - last_logged_at >= LOG_PROGRESS_EVERY:
		elapsed = time.time() - start_time
		rate = stats["processed"] / elapsed if elapsed > 0 else 0
		logger.info(
		f" Progress: {stats['processed']}/{total} ({rate:.0f}/sec), "
		f"hypertable={stats['hypertable_rows']}, updated={stats['bbox_updated']}"
		)
		last_logged_at = stats["processed"]

		# Final flush
		if hypertable_rows:
		stats["hypertable_rows"] += _flush_hypertable_batch(
		ObjectFrameObservation, hypertable_rows
		)
		if update_batch:
		_flush_update_batch(ClassificationDynamicAttrs, update_batch)

		elapsed = time.time() - start_time
		rate = stats["processed"] / elapsed if elapsed > 0 else 0

		logger.info("=" * 70)
		logger.info(f"Migration 0096 completed in {elapsed:.1f}s ({rate:.0f}/sec)")
		logger.info("=" * 70)
		logger.info(f" Total processed: {stats['processed']}")
		logger.info(f" Hypertable rows created: {stats['hypertable_rows']}")
		logger.info(f" Hypertable skipped (no valid data): {stats['hypertable_skipped']}")
		logger.info(f" Bbox records updated: {stats['bbox_updated']}")
		logger.info(f" Bbox records cleared (invalid): {stats['bbox_cleared']}")
		logger.info("=" * 70)


		def reverse_code(apps, schema_editor):
		# No reverse data migration provided
		pass


		@@ -149,5 +296,28 @@ class Migration(migrations.Migration):
		]

		operations = [
		# Schema changes from 0100: add first_frame_index field and alter bboxes
		migrations.AddField(
		model_name="classificationdynamicattrs",
		name="first_frame_index",
		field=models.IntegerField(
		blank=True,
		default=0,
		help_text="Zero-based index of the first frame where the object appears.",
		null=True,
		verbose_name="First frame index",
		),
		),
		migrations.AlterField(
		model_name="classificationdynamicattrs",
		name="bboxes",
		field=models.JSONField(
		blank=True,
		default=list,
		help_text="BBox for the first occurrence of given object",
		null=True,
		),
		),
		# Combined data migration: hypertable + bbox flattening
		migrations.RunPython(forwards, reverse_code),
		]

trapper/trapper-project/trapper/apps/media_classification/migrations/0096b_enable_timescaledb_compression.py

0 → 100644

+58 −0

Original line number	Diff line number	Diff line
		"""
		Enable TimescaleDB compression on ObjectFrameObservation hypertable.

		This migration runs AFTER the data migration (0096) to avoid performance
		degradation. If compression is enabled before bulk data insertion, the
		automatic compression policy may try to compress chunks while data is still
		being written, causing severe slowdowns.
		"""

		from django.db import migrations


		class Migration(migrations.Migration):

		dependencies = [
		("media_classification", "0096_migrate_bboxes_to_timeseries"),
		]

		operations = [
		# Enable native compression: order by time, segment by object identifiers for better locality
		migrations.RunSQL(
		sql=(
		"ALTER TABLE media_classification_objectframeobservation SET ("
		" timescaledb.compress,"
		" timescaledb.compress_orderby='ts',"
		" timescaledb.compress_segmentby='project_id,classification_id,dynamic_attrs_id'"
		");"
		),
		reverse_sql=(
		"ALTER TABLE media_classification_objectframeobservation SET ("
		" timescaledb.compress=false"
		");"
		),
		),
		# Add an automatic compression policy: compress chunks older than 30 days
		# Using DO block to handle idempotency - only add if not already exists
		migrations.RunSQL(
		sql="""
		DO $$
		BEGIN
		IF NOT EXISTS (
		SELECT 1 FROM timescaledb_information.jobs
		WHERE proc_name = 'policy_compression'
		AND hypertable_name = 'media_classification_objectframeobservation'
		) THEN
		PERFORM add_compression_policy(
		'media_classification_objectframeobservation',
		INTERVAL '30 days'
		);
		END IF;
		END
		$$;
		""",
		reverse_sql="""
		SELECT remove_compression_policy('media_classification_objectframeobservation', if_exists => true);
		""",
		),
		]

trapper/trapper-project/trapper/apps/media_classification/migrations/0097_remove_megadetectorbatchprocessingprovider_aiprovider_ptr_and_more.py

+116 −73

File changed.

Preview size limit exceeded, changes collapsed.