commit before cleanup

vision_v3/10_do_obj_det_and_clip/auto_compact_milvus.py

@@ -0,0 +1,136 @@
+#!/home/thebears/envs/vector_search/bin/python
+
+"""Trigger and (optionally) wait on Milvus compaction for one or more collections.
+
+Each collection is flushed before compaction is triggered, since Milvus only
+compacts sealed segments -- flushing seals any growing segments so recently
+inserted data is actually eligible.
+
+By default this compacts every collection on the server once and exits, which is
+the right shape for a cron job / systemd oneshot timer. Pass --loop to instead run
+forever, re-compacting every --interval seconds (for a long-running systemd service).
+"""
+
+from pymilvus import MilvusClient
+from fnmatch import fnmatch
+import argparse
+import logging
+import time
+import random
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_URI = "http://192.168.1.242:19530"
+
+
+def compact_collections(
+    client,
+    pattern="*",
+    is_clustering=False,
+    wait=True,
+    poll_interval=60,
+    timeout=3600*10,
+    dry_run=False,
+):
+    """Trigger compaction for every collection whose name matches `pattern`.
+
+    Returns a dict of {collection_name: True/False/None}, where None means the
+    job was triggered but not waited on (wait=False).
+    """
+    results = dict()
+    collections = [c for c in client.list_collections() if fnmatch(c, pattern)]
+
+    if not collections:
+        logger.info(f"No collections matched pattern {pattern!r}")
+        return results
+
+    random.shuffle(collections)
+    for coll_name in collections:
+        if dry_run:
+            logger.info(f"[dry-run] would flush and compact {coll_name}")
+            continue
+
+        try:
+            client.flush(coll_name)
+        except Exception as e:
+            logger.error(f"Failed to flush {coll_name} before compaction: {e}")
+            results[coll_name] = False
+            continue
+
+        try:
+            job_id = client.compact(coll_name, is_clustering=is_clustering)
+        except Exception as e:
+            logger.error(f"Failed to trigger compaction for {coll_name}: {e}")
+            results[coll_name] = False
+            continue
+
+        logger.info(f"Triggered compaction job {job_id} for {coll_name}")
+
+        if not wait:
+            results[coll_name] = None
+            continue
+
+        waited = 0
+        state = None
+        while waited < timeout:
+            try:
+                state = client.get_compaction_state(job_id)
+            except Exception as e:
+                logger.error(f"Failed to poll compaction state for {coll_name} (job {job_id}): {e}")
+                break
+            if state == "Completed":
+                break
+            time.sleep(poll_interval)
+            waited += poll_interval
+
+        success = state == "Completed"
+        if success:
+            logger.info(f"Compaction of {coll_name} completed (job {job_id})")
+        else:
+            logger.warning(f"Compaction of {coll_name} (job {job_id}) ended with state={state} after {waited}s")
+        results[coll_name] = success
+
+    return results
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        prog="Auto-compact Milvus collections"
+    )
+    parser.add_argument("--uri", default=DEFAULT_URI, help="Milvus server URI")
+    parser.add_argument("--pattern", default="*", help="fnmatch glob to select which collections to compact")
+    parser.add_argument("--is-clustering", action="store_true", dest="is_clustering",
+                         help="Trigger clustering compaction instead of a regular merge compaction")
+    parser.add_argument("--no-wait", action="store_false", dest="wait",
+                         help="Trigger compaction jobs without waiting for them to finish")
+    parser.add_argument("--poll-interval", type=float, default=5, help="Seconds between compaction state polls")
+    parser.add_argument("--timeout", type=float, default=3600, help="Max seconds to wait per collection's compaction job")
+    parser.add_argument("--dry-run", action="store_true", help="List collections that would be compacted and exit")
+    parser.add_argument("--loop", action="store_true", help="Run forever, re-compacting on a fixed interval")
+    parser.add_argument("--interval", type=float, default=6 * 3600, help="Seconds between runs when --loop is set")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s|%(levelname)s|%(message)s")
+
+    client = MilvusClient(uri=args.uri)
+    try:
+        while True:
+            results = compact_collections(
+                client,
+                pattern=args.pattern,
+                is_clustering=args.is_clustering,
+                wait=args.wait,
+                poll_interval=args.poll_interval,
+                timeout=args.timeout,
+                dry_run=args.dry_run,
+            )
+            for coll_name, ok in results.items():
+                if ok is False:
+                    print(f"Failed: {coll_name}")
+
+            if not args.loop:
+                break
+            logger.info(f"Sleeping {args.interval}s before next compaction pass")
+            time.sleep(args.interval)
+    finally:
+        client.close()

vision_v3/10_do_obj_det_and_clip/upload_npz_to_milvus.py

@@ -1,4 +1,6 @@
-from common_code.vector_utils import upload_vectors_to_db
+#!/home/thebears/envs/vector_search/bin/python
+
+from common_code.vector_utils import bulk_insert_vectors_to_db, bulk_import_vectors_to_db
 from common_code import kwq, util, file_names, vector_utils, settings
 from multiprocessing import Pool
 from tqdm import tqdm
@@ -10,7 +12,7 @@ logger = logging.getLogger(__name__)
 
 
 topic_subscribe = 'upload_to_vector_db'
-def add_files_to_queue(paths, dry_run = False, force = False, do_upload = True, do_queue = False):
+def add_files_to_queue(paths, dry_run = False, force = False, do_upload = True, do_queue = False, use_bulk_import = False):
     queued = set()
     for rt in paths:
         for root, dirs, files in os.walk(rt):
@@ -20,20 +22,19 @@ def add_files_to_queue(paths, dry_run = False, force = False, do_upload = True,
 #                if decide_to_put_in_queue(new_path, force = force):
                     queued.add(new_path)
 
-    for x in tqdm(queued):
-        try:
-            if do_upload:
-                if os.path.exists(x+'.in_queue'):
-                    pass
-                else:
-                    upload_vectors_to_db(x)
-                    with open(x+'.in_queue','w') as ff:
-                        pass
-            if do_queue:
-                kwq.publish(topic_subscribe, x, x)
+    if do_upload:
+        upload_fn = bulk_import_vectors_to_db if use_bulk_import else bulk_insert_vectors_to_db
+        results = upload_fn(list(tqdm(queued)))
+        for x, ok in results.items():
+            if not ok:
+                print('Failed: ' + str(x))
 
-        except Exception as e:
-            print('Failed: '+str(x)+'with '+str(e))
+    if do_queue:
+        for x in tqdm(queued):
+            try:
+                kwq.publish(topic_subscribe, x, x)
+            except Exception as e:
+                print('Failed: '+str(x)+'with '+str(e))
 
 
 
@@ -49,14 +50,25 @@ if __name__ == "__main__":
     parser.add_argument("--force", action='store_true', help='Force adding (do not check if already done)')
     parser.add_argument("--skip_upload", action='store_true', help='Force adding (do not check if already done)')
     parser.add_argument("--do_queue", action='store_true', help='Force adding (do not check if already done)')
+    parser.add_argument("--bulk-import", action='store_true', dest='bulk_import',
+                         help='EXPERIMENTAL/UNSAFE: use the Milvus server-side Bulk Import job API '
+                              'instead of batched inserts. Known to crash this Milvus build when the '
+                              'imported segment is loaded -- do not use against production collections.')
     args, _ = parser.parse_known_args()
     paths = args.paths
     force = args.force
     dry_run = args.dry_run
     do_queue = args.do_queue
     do_upload = not args.skip_upload
+    use_bulk_import = args.bulk_import
     if len(paths) == 0:
         paths = settings.dir_watch
 
-    add_files_to_queue(paths, dry_run = dry_run, force=force, do_upload = do_upload, do_queue = do_queue)
+    print(paths)
+    add_files_to_queue(paths, dry_run = dry_run, force=force, do_upload = do_upload, do_queue = do_queue, use_bulk_import = use_bulk_import)
 
+
+# %%
+# from common_code.vector_utils import upload_vectors_to_db
+# ff = '/srv/ftp/hummingbird/2026/07/18/hummingbird_00_20260718144349.oclip_embeds.npz'
+# upload_vectors_to_db(ff)

vision_v3/20_delete_files_and_create_other_videos/20_delete_files_and_create_other_videos.py

@@ -56,8 +56,6 @@ def get_ok_to_delete(file_path):
 # %%
 
 def exec_file_remove_logic(cset, dry_run = False):
-
-    
     did_purge = False
     if isinstance(cset, dict):
         file_path = cset['.mp4']
@@ -92,6 +90,7 @@ def exec_file_remove_logic(cset, dry_run = False):
         n_frames_scored = len(det_results['scored_frames'])
 
         frac_frames_scored = n_frames_above / n_frames_scored
+        logger.info(f'Frac frames scored: {frac_frames_scored}')
         do_keep = frac_frames_scored > settings['frames_with_dets_threshold']
 
 
@@ -138,6 +137,7 @@ def perform_kafka_loop():
             return
         logger.info(f"Executing file operations {pfm(file_to_score)}")
         did_purge = exec_file_remove_logic(get_cset_match(file_to_score))
+        logger.info(f"Did purge? {did_purge}   {get_cset_match(file_to_score)}")
         if not did_purge:
             did_convert = convert_file_with_check(file_to_score)
             FTPVideo(file_to_score).frames_info;