diff --git a/vision_v3/10_do_obj_det_and_clip/__pycache__/auto_compact_milvus.cpython-314.pyc b/vision_v3/10_do_obj_det_and_clip/__pycache__/auto_compact_milvus.cpython-314.pyc new file mode 100644 index 0000000..dbcd098 Binary files /dev/null and b/vision_v3/10_do_obj_det_and_clip/__pycache__/auto_compact_milvus.cpython-314.pyc differ diff --git a/vision_v3/10_do_obj_det_and_clip/__pycache__/upload_npz_to_milvus.cpython-314.pyc b/vision_v3/10_do_obj_det_and_clip/__pycache__/upload_npz_to_milvus.cpython-314.pyc new file mode 100644 index 0000000..367a996 Binary files /dev/null and b/vision_v3/10_do_obj_det_and_clip/__pycache__/upload_npz_to_milvus.cpython-314.pyc differ diff --git a/vision_v3/10_do_obj_det_and_clip/auto_compact_milvus.py b/vision_v3/10_do_obj_det_and_clip/auto_compact_milvus.py new file mode 100755 index 0000000..ad3c8c5 --- /dev/null +++ b/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() diff --git a/vision_v3/10_do_obj_det_and_clip/upload_npz_to_milvus.py b/vision_v3/10_do_obj_det_and_clip/upload_npz_to_milvus.py old mode 100644 new mode 100755 index dfeabc7..bc21c59 --- a/vision_v3/10_do_obj_det_and_clip/upload_npz_to_milvus.py +++ b/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,22 +22,21 @@ 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: + 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)) + + 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)) - except Exception as e: - print('Failed: '+str(x)+'with '+str(e)) - @@ -47,16 +48,27 @@ if __name__ == "__main__": parser.add_argument("paths", nargs="*", help="Paths to monitor", default=()) parser.add_argument("--dry-run", action='store_true', help='Dry Run') 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("--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) diff --git a/vision_v3/20_delete_files_and_create_other_videos/20_delete_files_and_create_other_videos.py b/vision_v3/20_delete_files_and_create_other_videos/20_delete_files_and_create_other_videos.py index b0cc4e1..324ba35 100644 --- a/vision_v3/20_delete_files_and_create_other_videos/20_delete_files_and_create_other_videos.py +++ b/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;