bump

vision_v3/20_delete_files_and_create_other_videos/20_add_into_queue.py

@@ -0,0 +1,42 @@
+
+
+import os
+import sys
+import time
+from common_code import kwq
+from common_code.util import get_cset_match
+from common_code.file_names import resolve_file_location
+import json
+import sys
+import logging
+from common_code.settings import get_logger, LogColorize
+sys.path.insert(0, "/home/thebears/Source/ml/sources_mltraining/model_runner/")
+from video_reduce_resolution import convert_file_with_check
+from common_code import settings
+default_dir_watch = settings.dir_watch
+
+
+pfm = LogColorize.file_modifications
+logger = get_logger(__name__,'/var/log/ml_vision_logs/20_add_into_queue', stdout=True, systemd=False, level = logging.INFO)
+
+success_det_and_embed_topic = kwq.TOPICS.exit_60_videos_embedded
+
+
+
+f_nav = list()
+for c_dir in settings.dir_watch_archived():
+    for root, _, files in  os.walk(c_dir):
+        for f in files:
+            if f.endswith('.mp4') and "reduced" not in f:
+                cfile = os.path.join( root, f)
+                f_nav.append(cfile)
+
+
+# %%
+#for f, _ in zip(f_nav, range(10)):
+from tqdm import tqdm
+
+    kwq.publish(success_det_and_embed_topic, f)
+
+# %%
+

vision_v3/20_delete_files_and_create_other_videos/20_delete_files_and_create_other_videos.py

@@ -0,0 +1,154 @@
+
+import os
+import sys
+import time
+from common_code import kwq
+from common_code.util import get_cset_match
+from common_code.file_names import resolve_file_location
+import json
+import sys
+import logging
+from common_code.settings import get_logger, LogColorize
+sys.path.insert(0, "/home/thebears/Source/ml/sources_mltraining/model_runner/")
+from video_reduce_resolution import convert_file_with_check
+import numpy as np
+
+pfm = LogColorize.file_modifications
+logger = get_logger(__name__,'/var/log/ml_vision_logs/20_delete_files_and_create_other_videos', stdout=True, systemd=False, level = logging.INFO)
+
+success_det_and_embed_topic = kwq.TOPICS.exit_60_videos_embedded
+
+group_id = "vision_v3"
+client_id = "file_converter"
+
+if not ('__file__' in vars() or '__file__' in globals()):
+    __file__ = '/home/thebears/Source/pipelines/vision_v3/20_delete_files_and_create_other_videos/20_delete_files_and_create_other_videos.py'
+
+
+logger.info(pfm(f"Starting file modifier @ {__file__}"))
+
+
+queue = kwq.create_consumer(group_id = group_id, client_id = client_id, auto_offset_reset = 'earliest')
+queue.subscribe([success_det_and_embed_topic])
+
+# %%
+def get_ok_to_delete(file_path):
+    max_look = 5
+    c_path = file_path
+    settings_default = {'class_threshold':10000, 'frames_with_dets_threshold': -10000, 'species_ignore': [], 'keep_reduced_mp4': False}
+
+    depth_set = dict()
+    for i in range(max_look):
+        c_path  = os.path.abspath(os.path.join(c_path, '..'))
+        test_path = os.path.join(c_path, 'settings.json')
+        if os.path.exists(test_path):
+            with open(test_path, 'r') as rr:
+                depth_set[i] = json.load(rr)
+
+    depth_set = dict(sorted(depth_set.items(), reverse=True))
+
+    for depth, val in depth_set.items():
+        settings_default.update(val)
+
+
+    settings_default['species_ignore'] = set(settings_default['species_ignore'])
+    return settings_default
+# %%
+
+def exec_file_remove_logic(cset, dry_run = False):
+
+    
+    did_purge = False
+    if isinstance(cset, dict):
+        file_path = cset['.mp4']
+    else:
+        file_path = cset
+        cset = get_cset_match(file_path)
+    
+    settings = get_ok_to_delete(file_path)
+
+    if settings['class_threshold'] > 1 or settings['class_threshold'] <= 0:
+        print(f"THRESHOLD SET TO ABOVE 1, SKIPPING: {file_path}")
+        return False
+
+
+    if '.detection.npz' in cset and os.path.exists(cset['.detection.npz']):
+        ff = np.load(cset['.detection.npz'], allow_pickle = True)
+        det_results = dict(ff)
+
+        del_idx = list()
+        for idx, val in det_results['species_label_map'].item().items():
+            if val in settings['species_ignore']:
+                del_idx.append(idx)
+
+        keep_mask = ~np.isin(det_results['final_labels'], del_idx)
+
+        filt_scores = det_results['final_scores'][keep_mask]
+        filt_frames = det_results['final_frames'][keep_mask]
+
+        passes = filt_scores > settings['class_threshold']
+
+        n_frames_above = len(np.unique(filt_frames[passes]))
+        n_frames_scored = len(det_results['scored_frames'])
+
+        frac_frames_scored = n_frames_above / n_frames_scored
+        do_keep = frac_frames_scored > settings['frames_with_dets_threshold']
+
+
+        keep_reduced_mp4 = settings['keep_reduced_mp4']
+        if do_keep:
+            cpath = os.path.splitext(cset['.mp4'])[0] + '.has_objs'
+            with open(cpath,'w') as cc:
+                pass
+                did_purge = False
+        else:
+            did_purge = True
+
+            for ff, cf in cset.items():
+                if cf.endswith('npz') or cf.endswith('jpg'):
+                    pass
+                else:
+                    if not dry_run and os.path.exists(cf):
+                        if '_reduced' in cf:
+                            logger.info(f"Skipping reduced_mp4 deletion {pfm(cf)}")
+                            continue
+                            
+                        logger.info(f"Removing {pfm(cf)}")
+                    else:
+                        print(f'Skipping deleting {ff} because of dry run or path not exist')
+    return did_purge
+
+def get_score_file(f):
+    return f.key().decode().strip('"')
+
+from common_code.video_meta import FTPVideo
+
+
+def perform_kafka_loop():
+    try:
+#    if True:
+        result = queue.poll(timeout=0.05)
+        if result is None:
+            return
+        
+        queue.commit()
+        file_to_score = resolve_file_location(get_score_file(result))
+        if file_to_score is None:
+            logger.info(f"Not found {pfm(file_to_score)}")
+            return
+        logger.info(f"Executing file operations {pfm(file_to_score)}")
+        did_purge = exec_file_remove_logic(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;
+        
+        logger.info(f"Finished file operations {pfm(file_to_score)}")        
+    except Exception as e:
+        logger.error(f"Failed: {pfm(file_to_score)}, {str(e)}") 
+
+
+while True:
+    perform_kafka_loop()
+
+
+

vision_v3/20_delete_files_and_create_other_videos/create_other_videos.py

@@ -0,0 +1,38 @@
+from CommonCode import wq
+from CommonCode.video_meta import FTPVideo
+import os
+import sys
+import time
+
+
+sys.path.insert(0, "/home/thebears/Source/ml/sources_mltraining/scoring_scheduler")
+sys.path.insert(0, "/home/thebears/Source/ml/sources_mltraining/model_runner/")
+
+from trim_with_birds import execute_file_trimming
+from video_reduce_resolution import convert_file_with_check
+input_topic = wq.TOPICS.ml_vision_objdet_results_purge_skipped
+output_if_success = wq.TOPICS.ml_vision_videos_modify_success
+output_if_fail = wq.TOPICS.ml_vision_videos_modify_failed
+
+
+
+def perform_loop():
+    file_to_score = wq.consume(input_topic)
+    if file_to_score is None:
+        return
+
+    try:
+        did_trim = False#        did_trim = execute_file_trimming(files_to_check = file_to_score)
+        did_convert = convert_file_with_check(file_to_score)
+        FTPVideo(file_to_score).frames_info;
+        wq.publish(output_if_success, file_to_score)
+        print(f"File trimmed:{did_trim}, did_downsize:{did_convert}, {file_to_score}")
+
+    except Exception as e:
+        wq.publish(output_if_fail, {file_to_score: str(e)})
+        print(f"File fail to create {file_to_score}, {e}")
+
+# %%
+while True:
+    perform_loop()
+    time.sleep(1)

vision_v3/20_delete_files_and_create_other_videos/insert_in_db.py

@@ -0,0 +1,10 @@
+
+det_npz = '/srv/ftp/railing/2025/11/17/railing_00_20251117140039.detection.npz'
+mp4_path = '/srv/ftp/railing/2025/11/17/railing_00_20251117140039.mp4'
+import numpy as np
+ff = np.load(det_npz, allow_pickle = True)
+det_results = dict(ff)
+
+del_idx = list()
+from common_code.video_meta import FTPVideo
+FTPVideo(mp4_path)

vision_v3/20_delete_files_and_create_other_videos/purge_file_if_no_dets.py

@@ -0,0 +1,52 @@
+import os
+import sys
+import time
+
+
+from CommonCode import wq, kwq
+sys.path.insert(0, "/home/thebears/Source/ml/sources_mltraining/scoring_scheduler")
+sys.path.insert(0, "/home/thebears/Source/ml/sources_mltraining/model_runner/")
+from purge_without_birds import execute_file_purging
+
+from video_reduce_resolution import convert_file
+
+input_topic = wq.TOPICS.ml_vision_objdet_results_pg_success
+output_if_success = wq.TOPICS.ml_vision_objdet_results_purge_success
+output_if_fail = wq.TOPICS.ml_vision_objdet_results_purge_failed
+output_if_skipped = wq.TOPICS.ml_vision_objdet_results_purge_skipped
+
+
+topic_if_has_nuggets = kwq.TOPICS.exit_40_videos_with_nuggets
+topic_if_no_nuggets = kwq.TOPICS.exit_40_videos_without_nuggets
+
+
+
+def perform_loop():
+    file_to_score = wq.consume(input_topic)
+    if file_to_score is None:
+        return
+
+    try:
+        purged_result = execute_file_purging(files_to_check=file_to_score)
+        did_purge = False
+        if purged_result is not None and len(purged_result) > 0:
+            did_purge = purged_result[0]
+            
+        if did_purge:
+            wq.publish(output_if_success, file_to_score)
+            print(f"File purged {file_to_score}")
+            kwq.publish(topic_if_no_nuggets, file_to_score)
+        else:
+            wq.publish(output_if_skipped, file_to_score)
+            print(f"File not purged {file_to_score}")
+            kwq.publish(topic_if_has_nuggets, file_to_score)
+
+    except Exception as e:
+        wq.publish(output_if_fail, {file_to_score: str(e)})
+        print(f"File failed to purge {file_to_score}, {e}")
+
+
+# %%
+while True:
+    perform_loop()
+    time.sleep(1)

vision_v3/20_delete_files_and_create_other_videos/upload_from_folder.py

@@ -0,0 +1,75 @@
+from pymilvus import MilvusClient, DataType
+import numpy as np
+import time
+from pymilvus.client.types import LoadState
+client = MilvusClient(
+    uri="http://localhost:19530"
+)
+
+
+res = client.get_load_state(
+    collection_name="nuggets_so400m"
+)
+if res['state'] == LoadState.Loaded:
+    pass
+else:
+    client.load_collection(collection_name = 'nuggets_so400m')
+    for i in range(10):
+        time.sleep(1)
+        if res['state'] == LoadState.Loaded:
+            break
+
+
+def get_vec_path(vpath):
+    return os.path.splitext(vpath)[0]+'.oclip_embeds.npz'
+
+def get_db_embed_done_path(vpath):
+    return os.path.splitext(vpath)[0]+'.db_has_oclip_embeds'
+
+
+def upload_vector_file(vector_file_to_upload):
+    if os.path.exists(get_embed_done_path(vector_file_to_upload)):
+        print('Already exists in DB, skipping upload')
+        return
+
+    vector_file_to_upload = get_vec_path(vector_file_to_upload)
+    vf = np.load(vector_file_to_upload)
+
+    embeds = vf['embeds']
+    fr_nums = vf['frame_numbers']
+
+    fname_root = vector_file_to_upload.rsplit('/',1)[-1].split('.')[0]
+    fc = fname_root.split('_')[-1]
+
+    data = list()
+    filepath = vector_file_to_upload.replace('/srv/ftp/','').replace('/mergedfs/ftp','').split('.')[-0]
+
+    for embed, frame_num in zip(embeds, fr_nums):
+        fg = '{0:05g}'.format(frame_num)
+        id_num = int(fc+fg)
+        to_put = dict(primary_id= id_num, filepath=filepath, frame_number = int(frame_num), so400m=embed)
+        data.append(to_put)
+
+    client.insert(collection_name = 'nuggets_so400m', data = data)
+    print(f'Inserting into DB, {vector_file_to_upload}')
+
+    with open(get_embed_done_path(vector_file_to_upload),'w') as ff:
+        ff.write(str(time.time()))
+
+
+
+
+
+root_path = '/srv/ftp/railing/2026/'
+to_put = list()
+for root, dirs, files in os.walk(root_path):
+    for x in files:
+        if x.endswith('oclip_embeds.npz'):
+            to_put.append(os.path.join(root, x))
+
+
+for x in to_put:
+    upload_vector_file(x)
+
+
+