feat: add ffprobe side_data parsing for HDR/SDR detection

data/tracker.csv

@@ -1,8 +1,4 @@
-ff,2
-simple_ffprobe_script,1
-show_path,2
-append,15
-deappend,17
-rename_to_first_n,3
-count_files,4
-remove_fname_last_n,2
+git-honney333-work-cred,794
+ff,53
+deappend,8
+rename_to_first_n,1

ff

@@ -3,6 +3,7 @@ import sys
 import os
 import subprocess
 import json
+import argparse
 from check_video import check_video_ext, normalize_language, normalize_lang_code
 
 color = True
@@ -124,7 +125,7 @@ def short_codec_name(codec):
     return fixed_width(new_codec.upper(), 5)
 
 class video_lines:
-    def __init__(self, stream):
+    def __init__(self, stream, hdr_type):
         if stream.get("index"):
             self.id = stream.get("index")
         else:
@@ -185,6 +186,11 @@ class video_lines:
         else:
             self.field_order = ""
         
+        if hdr_type:
+            self.hdr_type = hdr_type
+        else:
+            self.hdr_type = ""
+
     def __str__(self):
         string = fixed_width("Video", 6)
         if self.id != None:
@@ -197,7 +203,10 @@ class video_lines:
         if self.resolution != "x":
             string += f" ({self.resolution}"
             if self.framerate != "x":
-                string += f"@{self.framerate})"
+                string += f"@{self.framerate}"
+            string += f" {self.hdr_type})"
+        elif self.hdr_type != "Unknown HDR":
+            string += f" ({self.hdr_type})"
         if self.aspect_ratio:
             string += f" [{self.aspect_ratio}]"
         if self.pix_fmt and self.color_space:
@@ -255,6 +264,12 @@ class audio_lines:
         
         if self.language:
             string += f" [{self.language}]"
+        if len(self.language) >= 14:
+            pass
+        elif len(self.language) >= 8:
+            string += "\t"
+        else:
+            string += "\t\t"
             
         if self.duration:
             string += f" {self.duration}s"
@@ -309,6 +324,13 @@ class subtitles:
         if self.language:
             parts.append(f"[{self.language}]")
 
+        if len(self.language) >= 14:
+            pass
+        elif len(self.language) >= 7:
+            parts.append("\t")
+        else:
+            parts.append("\t\t")
+
         if self.duration:
             parts.append(f"{self.duration}s")
             
@@ -375,6 +397,212 @@ class subtitles:
 
 #     return duration, video_streams, audio_streams, subtitle_streams
 
+def _side_data_hints(side_data_list):
+    """Return set of hints from ffprobe side_data_list (case-insensitive)."""
+    hints = set()
+    for sd in side_data_list or []:
+        try:
+            t = str(sd.get("side_data_type", "")).lower()
+        except Exception:
+            continue
+        if "dovi" in t or "dolby" in t or "rpu" in t:
+            hints.add("DV")
+        if "2094-40" in t or "hdr dynamic" in t or "hdr10+" in t:
+            hints.add("HDR10+")
+        if "mastering" in t or "content light" in t or "content_light" in t or "cll" in t:
+            hints.add("HDR-static")
+    return hints
+
+
+def _classify_fast(video):
+    """Tier 0: pure-python, zero extra subprocess.
+
+    Returns (label, needs_deep). Final labels need no further probing.
+    UNKNOWN with needs_deep=True means Tier 1/2 should run.
+    Uses SDR? for highly-probable-but-untagged SDR, HDR? for probable HDR.
+    """
+    transfer = str(video.get("color_transfer") or "").lower()
+    primaries = str(video.get("color_primaries") or "").lower()
+    space = str(video.get("color_space") or "").lower()
+    pix_fmt = str(video.get("pix_fmt") or "").lower()
+    codec = str(video.get("codec_name") or "").lower()
+    profile = str(video.get("profile") or "").lower()
+    bits = video.get("bits_per_raw_sample")
+    try:
+        bits = int(bits) if bits is not None else 0
+    except Exception:
+        bits = 0
+
+    hints = _side_data_hints(video.get("side_data_list"))
+
+    # Cover art / thumbnails are not real video tracks — never deep-probe them.
+    dispo = video.get("disposition") or {}
+    try:
+        if dispo.get("attached_pic") == 1 or codec in ("mjpeg", "png", "gif", "bmp", "tiff"):
+            return ("SDR", False)
+    except Exception:
+        pass
+
+    # 1. Definitive transfer function
+    if transfer == "smpte2084":
+        if "DV" in hints:
+            return ("DV", False)
+        if "HDR10+" in hints:
+            return ("HDR10+", False)
+        return ("HDR10", False)
+    if transfer == "arib-std-b67":
+        return ("HLG", False)
+    sdr_transfers = {
+        "bt709", "bt470m", "bt470bg", "smpte170m", "smpte240m",
+        "gamma22", "gamma28", "iec61966-2-1", "iec61966-2-4", "srgb",
+    }
+    if transfer in sdr_transfers:
+        return ("SDR", False)
+
+    # 2. Primaries / matrix when transfer was stripped
+    is_bt2020 = primaries == "bt2020" or space in ("bt2020nc", "bt2020c", "bt2020_ncl", "bt2020_cl")
+    is_bt709 = primaries == "bt709" or space == "bt709"
+    if is_bt2020:
+        if "DV" in hints:
+            return ("DV", False)
+        if "HDR10+" in hints:
+            return ("HDR10+", False)
+        # Transfer missing so HDR10 vs HLG unknown — probable HDR.
+        return ("HDR?", False)
+    if is_bt709:
+        return ("SDR", False)
+    if "HDR-static" in hints or "DV" in hints or "HDR10+" in hints:
+        if "DV" in hints:
+            return ("DV", False)
+        if "HDR10+" in hints:
+            return ("HDR10+", False)
+        return ("HDR?", False)
+
+    # 3. All color tags empty — fall back to bit-depth / codec heuristics.
+    is_10bit_plus = ("p10" in pix_fmt or "p12" in pix_fmt or "p14" in pix_fmt
+                     or "p16" in pix_fmt or bits >= 10 or "10" in profile
+                     or "12" in profile)
+    if not transfer and not primaries and not space:
+        if is_10bit_plus:
+            # 10-bit SDR (anime etc.) and stripped HDR look identical here.
+            return ("UNKNOWN", True)
+        # 8-bit + common SDR codecs = highly probable untagged SDR.
+        return ("SDR?", False)
+
+    # Partial tags, unknown values — probable SDR, no expensive probe.
+    if not is_10bit_plus:
+        return ("SDR?", False)
+    return ("UNKNOWN", True)
+
+
+def _classify_colors(transfer, primaries, space, hints):
+    """Shared helper for Tier 1/2 dicts. Returns label or None."""
+    transfer = str(transfer or "").lower()
+    primaries = str(primaries or "").lower()
+    space = str(space or "").lower()
+    if transfer == "smpte2084":
+        if "DV" in hints:
+            return "DV"
+        if "HDR10+" in hints:
+            return "HDR10+"
+        return "HDR10"
+    if transfer == "arib-std-b67":
+        return "HLG"
+    if transfer in ("bt709", "smpte170m", "smpte240m", "gamma22", "gamma28",
+                    "iec61966-2-1", "iec61966-2-4", "srgb", "bt470m", "bt470bg"):
+        return "SDR"
+    if primaries == "bt2020" or space in ("bt2020nc", "bt2020c", "bt2020_ncl", "bt2020_cl"):
+        if "DV" in hints:
+            return "DV"
+        if "HDR10+" in hints:
+            return "HDR10+"
+        return "HDR?"
+    if primaries == "bt709" or space == "bt709":
+        return "SDR"
+    if "DV" in hints:
+        return "DV"
+    if "HDR10+" in hints:
+        return "HDR10+"
+    if "HDR-static" in hints:
+        return "HDR?"
+    return None
+
+
+def _probe_frame_tier1(path, v_idx):
+    """Tier 1 (cheap): 1-frame + full stream dump for one v stream. No decode."""
+    try:
+        cmd = [
+            "ffprobe", "-v", "error",
+            "-select_streams", f"v:{v_idx}",
+            "-show_streams",
+            "-show_frames", "-read_intervals", "%+#1",
+            "-show_entries", "frame=color_space,color_transfer,color_primaries,pix_fmt,side_data_list",
+            "-of", "json",
+            path,
+        ]
+        r = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+        if r.returncode != 0:
+            return None
+        info = json.loads(r.stdout or "{}")
+        streams = info.get("streams") or []
+        frames = info.get("frames") or []
+        s0 = streams[0] if streams else {}
+        f0 = frames[0] if frames else {}
+        hints = _side_data_hints(s0.get("side_data_list")) | _side_data_hints(f0.get("side_data_list"))
+        # Prefer stream tags, fall back to frame tags per field.
+        label = _classify_colors(
+            s0.get("color_transfer") or f0.get("color_transfer"),
+            s0.get("color_primaries") or f0.get("color_primaries"),
+            s0.get("color_space") or f0.get("color_space"),
+            hints,
+        )
+        return label
+    except Exception:
+        return None
+
+
+def _probe_trace_tier2(path):
+    """Tier 2 (expensive): bitstream VUI/SEI via trace_headers, 1 frame, no decode."""
+    try:
+        import shutil
+        if shutil.which("ffmpeg") is None:
+            return None
+        cmd = [
+            "ffmpeg", "-hide_banner", "-v", "info",
+            "-i", path,
+            "-c:v", "copy", "-bsf:v", "trace_headers",
+            "-frames:v", "1", "-f", "null", "-",
+        ]
+        r = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+        out = ((r.stderr or "") + "\n" + (r.stdout or "")).lower()
+        if not out.strip():
+            return None
+        has_dv = "dovi" in out or "rpu" in out
+        has_hdr10plus = "2094-40" in out or "hdr10+" in out or "dhdr10" in out
+        has_static = "mastering" in out or "content_light" in out or "max_cll" in out
+        # VUI transfer_characteristics: 16=PQ(ST2084), 18=HLG; primaries 9=BT.2020
+        import re
+        m = re.search(r"transfer_characteristics\s*[:=]\s*(\d+)", out)
+        trc = m.group(1) if m else ""
+        m2 = re.search(r"colour_primaries\s*[:=]\s*(\d+)|color_primaries\s*[:=]\s*(\d+)", out)
+        prim = (m2.group(1) or m2.group(2)) if m2 else ""
+        if has_dv:
+            return "DV"
+        if has_hdr10plus:
+            return "HDR10+"
+        if trc == "16":
+            return "HDR10"
+        if trc == "18":
+            return "HLG"
+        if prim == "9" or has_static:
+            return "HDR?"
+        if trc in ("1", "6", "7"):
+            return "SDR"
+        return None
+    except Exception:
+        return None
+
+
 def get_media_info(file):
     cmd = [
         "ffprobe",
@@ -382,12 +610,12 @@ def get_media_info(file):
         "-show_entries",
         (
             "format=duration:format_tags=title:"
-            "stream=index,codec_type,codec_name,"
+            "stream=index,codec_type,codec_name,profile,"
             "width,height,r_frame_rate,bit_rate,duration,nb_frames,"
             "pix_fmt,field_order,time_base,display_aspect_ratio,"
             "color_space,color_transfer,color_primaries,bits_per_raw_sample,"
             "sample_rate,channels,bits_per_sample,"
-            "stream_disposition=forced,default:"
+            "stream_disposition=forced,default,attached_pic:"
             "stream_tags=language,title"
         ),
         "-of", "json",
@@ -399,7 +627,7 @@ def get_media_info(file):
         info = json.loads(result.stdout)
     except subprocess.CalledProcessError as e:
         print(f"Error running ffprobe: {e.stderr}")
-        return None, [], [], [], None
+        return None, [], [], [], None, []
 
     # Duration
     duration = None
@@ -424,12 +652,42 @@ def get_media_info(file):
         elif stream_type == "subtitle":
             subtitle_streams.append(stream)
 
+    # Tiered HDR/SDR detection: Tier 0 fast path first, expensive only on UNKNOWN.
+    hdr_types = []
+
+    if video_streams:
+        trace_cache = {}  # file-level Tier 2 result, run at most once per file
+        for v_pos, video in enumerate(video_streams):
+            label, needs_deep = _classify_fast(video)
+            if not needs_deep:
+                hdr_types.append(label)
+                continue
+            # Tier 1: 1-frame + stream dump (cheap, no decode)
+            deep = _probe_frame_tier1(file, v_pos)
+            if deep:
+                hdr_types.append(deep)
+                continue
+            # Tier 2: bitstream trace_headers (expensive, no decode)
+            if "trace" not in trace_cache:
+                trace_cache["trace"] = _probe_trace_tier2(file)
+            if trace_cache["trace"]:
+                hdr_types.append(trace_cache["trace"])
+                continue
+            hdr_types.append("UNKNOWN")
+
     if duration is not None:
         duration = float(duration)
     else:
         duration = float('nan')
 
-    return duration, video_streams, audio_streams, subtitle_streams, title
+    return (
+        duration,
+        video_streams,
+        audio_streams,
+        subtitle_streams,
+        title,
+        hdr_types
+    )
 
 def seconds_to_hms(seconds):
     if type(seconds) is float:
@@ -453,9 +711,9 @@ class video_file:
         self.name = os.path.basename(path) # 25.mkv
         self.size = os.path.getsize(path)
 
-        self.duration, videos, audios, subs, title = get_media_info(path)
+        self.duration, videos, audios, subs, title, hdr_types = get_media_info(path)
         
-        self.sort_video_info(videos)
+        self.sort_video_info(videos, hdr_types)
         self.sort_audio_info(audios)
         self.sort_subs_info(subs)
 
@@ -464,11 +722,11 @@ class video_file:
         self.duration = seconds_to_hms(self.duration)
         self.title = title
 
-    def sort_video_info(self, videos):
+    def sort_video_info(self, videos, hdr_types):
         self.videos = []
         if videos:
-            for vl in videos:
-                video_line = video_lines(vl)
+            for vl, hdr_t in zip(videos, hdr_types):
+                video_line = video_lines(vl, hdr_t)
                 self.videos.append(video_line)
 
     def sort_audio_info(self, audios):
@@ -485,6 +743,7 @@ class video_file:
                 self.subtitles.append(subtitle)
 
     def print(self):
+        if not mini:
             if self.base_tab == "\t":
                 np(f"{self.base_tab}{self.name} ({self.size}, {self.duration}, {self.bitrate} MB/s):", NORMAL_STYLE)
             else:
@@ -492,13 +751,29 @@ class video_file:
                 np(f"\t{self.name} ({self.size}, {self.duration}, {self.bitrate} MB/s):", NORMAL_STYLE)
             if self.title:
                 np(f"\t\tTitle: \"{self.title}\"", NORMAL_STYLE)
+            if self.videos:
+                np(f"\t    Videos:", INFO_STYLE)
             for video in self.videos:
                 np(f"\t\t{video}", NORMAL_STYLE)
+            if self.audios:
+                np(f"\t    Audios:", INFO_STYLE)
             for audio in self.audios:
                 np(f"\t\t{audio}", NORMAL_STYLE)
+            if self.subtitles:
+                np(f"\t    Subtitles:", INFO_STYLE)
             for subtitle in self.subtitles:
                 np(f"\t\t{subtitle}", NORMAL_STYLE)
             print()
+        else:
+            if self.base_tab == "\t":
+                np(f"{self.base_tab}{self.name} ({self.size}, {self.duration}, {self.bitrate} MB/s):", NORMAL_STYLE)
+            else:
+                np(f"{os.path.dirname(self.path)}/", INFO_STYLE)
+                np(f"\t{self.name} ({self.size}, {self.duration}, {self.bitrate} MB/s):", NORMAL_STYLE)
+            if self.title:
+                np(f"\t\tTitle: \"{self.title}\"", NORMAL_STYLE)
+            print()
+
         
 
 def get_folder_info(files):
@@ -561,32 +836,37 @@ def handle_folders(dirs, all_files):
                 all_files.append(dir_files)
 
 if __name__ == "__main__":
-    # print(sys.argv)
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-m", "--mini", action="store_true", help="Enable mini mode")
+    parser.add_argument("paths", nargs="*", help="Files or directories to process")
+    args = parser.parse_args()
+    mini = args.mini
+    # print(args)
     subprocess.run(["python", "/home/honney/.bin/tracker.py", "add", "ff"])
     file_dir_array = []
-    if len(sys.argv) == 0:
-        print("Something went horribly wrong!")
-    if len(sys.argv) == 1:
-        # current_dir = os.path.dirname(os.path.realpath(__file__))
+    # No paths were provided: use the current directory
+    if not args.paths:
         current_dir = os.getcwd()
         handle_folders([os.path.abspath(current_dir)], file_dir_array)
     else:
         files = []
         dirs = []
-        for argv in sys.argv[1:]:
+        for argv in args.paths:
             if os.path.isfile(argv):
                 if check_video_ext(os.path.splitext(argv)[1]):
                     files.append(os.path.abspath(argv))
                 # else:
-                #     np(f"{os.path.abspath(argv)} is not a compatabile Video file", WARN_STYLE)
+                #     np(
+                #         f"{os.path.abspath(argv)} is not a compatible Video file",
+                #         WARN_STYLE
+                #     )
             elif os.path.isdir(argv):
                 dirs.append(os.path.abspath(argv))
             else:
-                np(f"This is not a file or directory: {argv}\nNow canceling!", ERROR_STYLE)
+                np(f"This is not a file or directory: {argv}\nNow canceling!",ERROR_STYLE)
                 sys.exit()
         handle_folders(dirs, file_dir_array)
         handle_files(files, file_dir_array)
-
     for element in file_dir_array:
         if type(element) == list:
             get_folder_info(element)

recount_files

@@ -18,32 +18,54 @@ def main():
     else:
         print("Usage: recount_files.py <number> (<directory>)")
         sys.exit(1)
+
     import subprocess
     subprocess.run(["python", "/home/honney/.bin/tracker.py", "add", "recount_files"])
 
     number = int(sys.argv[1])
 
-    pattern = re.compile(r'^\d+')
+    # Pattern matches numbers (int or float) at the start (Group 1) 
+    # and captures the rest of the filename (Group 2)
+    pattern = re.compile(r'^(\d+(?:\.\d+)?)(.*)')
 
-    files = sorted(
-        [f for f in dir.iterdir() if f.is_file() and pattern.match(f.name)],
-        key=lambda f: float(f.stem)
-    )
+    files_data = []
+    for f in dir.iterdir():
+        if f.is_file():
+            match = pattern.match(f.stem)
+            if match:
+                # Store a tuple: (Path object, numeric_string, rest_of_filename)
+                files_data.append((f, match.group(1), match.group(2)))
 
-    min_num = int(files[0].stem)
-    max_len = max(len(files[-1].stem), 2)
+    if not files_data:
+        print("No matching files found.")
+        sys.exit(0)
+
+    # Sort files based on the numeric part cast to a float
+    files_data = sorted(files_data, key=lambda x: float(x[1]))
+
+    min_num_float = float(files_data[0][1])
+    
+    # Calculate padding based on the longest numeric string
+    max_len = max(len(files_data[-1][1]), 2)
+
+    failed = []
+
+    for f, num_str, rest in files_data:
+        # Determine if the current prefix is a float or an int
+        if '.' not in num_str:
+            delta = int(num_str) - int(min_num_float)
+            new_num_str = f"{delta+number:0{max_len}d}"
+        else:
+            delta = float(num_str) - min_num_float
+            new_num_str = f"{delta+number:0{max_len+2}.1f}"
+            
+        # Recombine: New Number + Rest of Name + Extension
+        new_file_name = Path(f"{f.parent}/{new_num_str}{rest}{f.suffix}")
         
-    for f in files:
-        try:
-            delta = int(f.stem)-min_num
-            new_file_name = Path(f"{f.parent}/{delta+number:0{max_len}d}{f.suffix}")
-        except:
-            delta = float(f.stem)-min_num
-            new_file_name = Path(f"{f.parent}/{delta+number:0{max_len+2}.1f}{f.suffix}")
         if not new_file_name.is_file():
             print(
                 colored("Renamed: ", "green") +
-                colored(str(f.parent), "cyan") +
+                colored(str(f.parent)+"/", "cyan") +
                 colored("[", "white") +
                 colored(f.name, "red") +
                 colored(" -> ", "white") +
@@ -52,8 +74,32 @@ def main():
             )
             f.rename(new_file_name)
         else:
+            failed.append([f, new_file_name])
+            print(colored(f"{new_file_name.name} already exists. Retrying later", "red"))
+
+    failed2 = []
+    if failed:
+        failed.reverse()
+        for f, new_file_name in failed:
+            if not new_file_name.is_file():
+                print(
+                    colored("Renamed: ", "green") +
+                    colored(str(f.parent)+"/", "cyan") +
+                    colored("[", "white") +
+                    colored(f.name, "red") +
+                    colored(" -> ", "white") +
+                    colored(new_file_name.name, "green") +
+                    colored("]", "white")
+                )
+                f.rename(new_file_name)
+            else:
+                failed2.append([f, new_file_name])
                 print(colored(f"{new_file_name.name} already exists. Aborting", "red"))
 
+    if failed2:
+        print(colored("Failed to rename the following files:", "red"))
+        for f, new_file_name in failed2:
+            print(colored(f"{new_file_name.name} already exists. Aborting", "red"))
 
 if __name__ == "__main__":
     main()

run_in_all_folders

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+if [ "$#" -lt 1 ]; then
+  echo "Usage: run_in_all_folders command [args...]"
+  exit 1
+fi
+
+for dir in */; do
+  [ -d "$dir" ] || continue
+
+  echo "==> Running in $dir"
+
+  (
+    cd "$dir" || exit
+    "$@"
+  )
+done