diff --git a/vision_v3/20_delete_files_and_create_other_videos/20_add_into_queue.py b/vision_v3/20_delete_files_and_create_other_videos/20_add_into_queue.py new file mode 100644 index 0000000..6a61fa5 --- /dev/null +++ b/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) + +# %% + 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 new file mode 100644 index 0000000..b0cc4e1 --- /dev/null +++ b/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() + + + diff --git a/vision_v3/20_delete_files_and_create_other_videos/create_other_videos.py b/vision_v3/20_delete_files_and_create_other_videos/create_other_videos.py new file mode 100644 index 0000000..bec548e --- /dev/null +++ b/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) diff --git a/vision_v3/20_delete_files_and_create_other_videos/insert_in_db.py b/vision_v3/20_delete_files_and_create_other_videos/insert_in_db.py new file mode 100644 index 0000000..362fc10 --- /dev/null +++ b/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) diff --git a/vision_v3/20_delete_files_and_create_other_videos/purge_file_if_no_dets.py b/vision_v3/20_delete_files_and_create_other_videos/purge_file_if_no_dets.py new file mode 100644 index 0000000..7f76212 --- /dev/null +++ b/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) diff --git a/vision_v3/20_delete_files_and_create_other_videos/upload_from_folder.py b/vision_v3/20_delete_files_and_create_other_videos/upload_from_folder.py new file mode 100644 index 0000000..414aa84 --- /dev/null +++ b/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) + + +