Implemented multi video pipeline, among other tweaks

Author: eduardotogpi

.gitignore

@@ -1,6 +1,9 @@
 *.wmv
 *.csv
 *.png
+*.mp4
+*.npz
+clear_data.py
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

LICENSE

@@ -1,3 +1,16 @@
 # pyCRTScripts
 
 User-friendly scripts designed to make pyCRT accessible to non-programmers.
+
+# TODO
+ * Add option to view the plots and CRT before saving;
+ * Try to implement interactive fromTime, toTime and algorithm selection;
+ * Implement PCA results as a channel;
+ * Develop algorithm to find appropriate fromTime and toTime;
+ * Add fromTime and toTime to avgIntensPlot;
+ * Add monochrome as a possible channel;
+ * Add tests for pyCRTScripts;
+ * Add 9010 as an algorithm;
+ * Add timestamp to csv;
+ * Add input "Test another video?";
+ * Add tests;

Plots/README.md

@@ -3,12 +3,13 @@
 plotPath = "Plots"
 csvPath = "results_sheet.csv"
 npzPath = "Npz"
-overwrite = true
+overwrite = false
+ignoreDuplicates = true
 
 [Video]
 
 rescaleFactor = 0.5
-livePlot = true
+livePlot = false  # Leave as false if using Windows
 displayVideo = true
 playbackSpeed = "fast"
 
@@ -17,12 +18,14 @@ playbackSpeed = "fast"
 sliceMethod = "from local max"
 fromTime = 0
 toTime = inf
-exclusionCriteria = 0.20
+exclusionCriteria = inf
 exclusionMethod = "first positive peak"
 channel = "g"
 initialGuesses = [1.0, -0.3, 0.0]
 roi = -1
 
 [General]
 
-logLevel = "INFO"
+logLevel = "DEBUG"
+showPlots = true
+askConfirmation = true

README.md

@@ -1,22 +1,30 @@
 from __future__ import annotations
 
 import logging
-import tkinter as tk
 from pathlib import Path
-from tkinter import filedialog
 from typing import Any
+import gui
 
 import numpy as np
 import pandas as pd
-import tomli
-from pyCRT.simpleUI import PCRT, RoiTuple
+
+try:
+    import tomllib as tomli  # Python 3.11+
+except ModuleNotFoundError:
+    import tomli  # fallback for older Python
+
+from pyCRT.simpleUI import DATETIME_FORMAT, PCRT, RoiTuple
 
 PLAYBACK_FPS_SPEED = {
     "fast": np.inf,
     "normal": 0,
     "slow": 25,
 }
 
+NORM_LEVEL = 25
+logging.addLevelName(NORM_LEVEL, "NORM")
+VIDEO_FORMATS = (".mp4", ".wmv", ".avi", ".mov", ".mkv")
+
 
 def getLogger(name: str):
     # {{{
@@ -35,21 +43,6 @@ def getLogger(name: str):
 LOGGER = getLogger("mauricio")
 
 
-def selectFile() -> Path:
-    # {{{
-    root = tk.Tk()
-    root.withdraw()  # hide main window
-    root.attributes("-topmost", True)
-    selection = filedialog.askopenfilename()
-    if not selection:
-        raise RuntimeError("No video selected. Aborting...")
-    filePath = Path(selection)
-    root.destroy()
-    return filePath
-
-
-# }}}
-
 
 def loadConfigFile(
     tomlPath, defaultsPath: Path | str = "defaults.toml"
@@ -221,7 +214,7 @@ def saveCSV(
     overwrite: bool = False,
 ) -> None:
     # {{{
-    colLabels = ["pCRT", "Unc", "RelUnc", "CT", "ExcCri", "ExcMet"]
+    colLabels = ["pCRT", "Unc", "RelUnc", "CT", "ExcCri", "ExcMet", "Time"]
 
     row = {
         "pCRT": round(float(pcrtObj.pCRT[0]), 3),
@@ -230,6 +223,7 @@ def saveCSV(
         "CT": round(float(pcrtObj.criticalTime), 3),
         "ExcCri": pcrtObj.exclusionCriteria,
         "ExcMet": pcrtObj.exclusionMethod,
+        "Time": pcrtObj.dateTime.strftime(DATETIME_FORMAT),
     }
 
     csvPath = Path(csvPath)
@@ -275,21 +269,11 @@ def saveNpz(
     npzFilePath = npzPath / f"{videoName}.npz"
     if not overwrite:
         npzFilePath = findUniquePath(npzFilePath)
-    np.savez(
-        str(npzFilePath),
-        fullTimeScdsArr=pcrtObj.fullTimeScdsArr,
-        channelsAvgIntensArr=pcrtObj.channelsAvgIntensArr,
-        channel=pcrtObj.channel,
-        fromTime=pcrtObj.fromTime,
-        toTime=pcrtObj.toTime,
-        expTuple=np.array(pcrtObj.expTuple),
-        polyTuple=np.array(pcrtObj.polyTuple),
-        pCRTTuple=np.array(pcrtObj.pCRTTuple),
-        criticalTime=pcrtObj.criticalTime,
-        exclusionMethod=pcrtObj.exclusionMethod,
-        exclusionCriteria=pcrtObj.exclusionCriteria,
-    )
+    pcrtObj.name = videoName
+    pcrtObj.save(npzFilePath)
     LOGGER.info(f"pCRT object saved in {npzFilePath}.")
+
+
 # }}}
 
 
@@ -340,16 +324,106 @@ def singleVideoPipeline(
 
     LOGGER.setLevel(configDict["General"]["logLevel"])
     LOGGER.info(f"Config loaded from {configPath}.")
-    crtVideoPath = selectFile()
+    crtVideoPath = gui.selectFile()
     videoName = crtVideoPath.stem
     pcrtObj, roi = measureCRTVideoFromConfig(crtVideoPath, configDict)
 
+    LOGGER.log(NORM_LEVEL, f"{pcrtObj.plotTitle}: {pcrtObj}")
+
     plotPath = configDict["Files"]["plotPath"]
     csvPath = configDict["Files"]["csvPath"]
     npzPath = configDict["Files"]["npzPath"]
     overwrite = configDict["Files"]["overwrite"]
 
+    showPlots = configDict["General"]["showPlots"]
+    if showPlots:
+        pcrtObj.showAvgIntensPlot()
+        pcrtObj.showPCRTPlot()
+
     savePlots(pcrtObj, videoName, plotPath, overwrite)
     saveCSV(pcrtObj, videoName, csvPath, overwrite)
     saveNpz(pcrtObj, videoName, npzPath, overwrite)
+
+
+# }}}
+
+
+def multiVideoPipeline(
+    configPath: Path | str,
+    defaultsPath: Path | str,
+) -> list[PCRT]:
+    # {{{
+    configDict = loadConfigFile(configPath, defaultsPath)
+
+    LOGGER.setLevel(configDict["General"]["logLevel"])
+    LOGGER.info(f"Config loaded from {configPath}.")
+
+    askConfirmation = configDict["General"]["askConfirmation"]
+    # Just in case the user decides to read the videos out of order
+    processedPaths = []
+
+    plotPath = configDict["Files"]["plotPath"]
+    csvPath = configDict["Files"]["csvPath"]
+    npzPath = configDict["Files"]["npzPath"]
+    overwrite = configDict["Files"]["overwrite"]
+    showPlots = configDict["General"]["showPlots"]
+
+    dirPath = gui.selectDirectory()
+    for candidatePath in dirPath.iterdir():
+        if not candidatePath.is_file():
+            LOGGER.debug(
+                f"Skipping {candidatePath.stem}.{candidatePath.suffix} "
+                "(not a file)..."
+            )
+            continue
+        if candidatePath.suffix not in VIDEO_FORMATS:
+            LOGGER.debug(
+                f"Skipping {candidatePath.stem}.{candidatePath.suffix} "
+                "(not a video)..."
+            )
+            continue
+        if candidatePath in processedPaths:
+            LOGGER.debug(
+                f"Skipping {candidatePath.stem}.{candidatePath.suffix} "
+                "(already processed)..."
+            )
+            continue
+        if askConfirmation:
+            answer = gui.askNextVideo(candidatePath)
+            if answer == "skip":
+                processedPaths.append(candidatePath)
+                continue
+            elif answer == "abort":
+                break
+            elif answer == "select":
+                actualPath = gui.selectFile()
+            else:
+                actualPath = candidatePath
+        else:
+            actualPath = candidatePath
+
+        LOGGER.info(f"Processing video {actualPath}.")
+        
+        videoName = actualPath.stem
+        try:
+            pcrtObj, _ = measureCRTVideoFromConfig(actualPath, configDict)
+            processedPaths.append(actualPath)
+        except (RuntimeError, TypeError, ValueError):
+            LOGGER.error("CRT calculation failed on {actualPath}")
+            processedPaths.append(actualPath)
+            continue
+
+        LOGGER.log(NORM_LEVEL, f"{pcrtObj.plotTitle}: {pcrtObj}")
+
+        if showPlots:
+            pcrtObj.showAvgIntensPlot()
+            pcrtObj.showPCRTPlot()
+        savePlots(pcrtObj, videoName, plotPath, overwrite)
+        saveCSV(pcrtObj, videoName, csvPath, overwrite)
+        saveNpz(pcrtObj, videoName, npzPath, overwrite)
+
+    LOGGER.info(f"Finished processing videos in {dirPath}")
+            
+
+
 # }}}

configuration.toml

@@ -0,0 +1,98 @@
+import tkinter as tk
+from pathlib import Path
+from tkinter import filedialog, ttk
+
+
+def askNextVideo(videoName: str) -> str:
+    # {{{
+    """
+    Open a dialog asking whether to continue to the next video. By chatGPT
+
+    Parameters
+    ----------
+    videoName : str
+        Name of the next video.
+
+    Returns
+    -------
+    str
+        One of: "yes", "skip", "select", "abort"
+    """
+    result: str | None = None
+
+    def set_result(value: str) -> None:
+        nonlocal result
+        result = value
+        root.destroy()
+
+    root = tk.Tk()
+    root.title("Continue?")
+    root.resizable(False, False)
+    root.attributes("-topmost", True)
+
+    frame = ttk.Frame(root, padding=12)
+    frame.pack(fill="both", expand=True)
+
+    label = ttk.Label(
+        frame,
+        text=f"Continue to next video:\n{videoName}?",
+        justify="center",
+    )
+    label.pack(pady=(0, 12))
+
+    btn_frame = ttk.Frame(frame)
+    btn_frame.pack()
+
+    ttk.Button(btn_frame, text="Yes", command=lambda: set_result("yes")).grid(
+        row=0, column=0, padx=5
+    )
+    ttk.Button(
+        btn_frame, text="Skip", command=lambda: set_result("skip")
+    ).grid(row=0, column=1, padx=5)
+    ttk.Button(
+        btn_frame, text="Select Video", command=lambda: set_result("select")
+    ).grid(row=0, column=2, padx=5)
+    ttk.Button(
+        btn_frame, text="Abort", command=lambda: set_result("abort")
+    ).grid(row=0, column=3, padx=5)
+
+    root.protocol("WM_DELETE_WINDOW", lambda: set_result("abort"))
+    root.mainloop()
+
+    assert result is not None
+    return result
+
+
+# }}}
+
+
+def selectFile() -> Path:
+    # {{{
+    root = tk.Tk()
+    root.withdraw()  # hide main window
+    root.attributes("-topmost", True)
+    selection = filedialog.askopenfilename()
+    if not selection:
+        raise RuntimeError("No video selected. Aborting...")
+    filePath = Path(selection)
+    root.destroy()
+    return filePath
+
+
+# }}}
+
+
+def selectDirectory() -> Path:
+    # {{{
+    root = tk.Tk()
+    root.withdraw()  # hide main window
+    root.attributes("-topmost", True)
+    selection = filedialog.askdirectory()
+    if not selection:
+        raise RuntimeError("No directory selected. Aborting...")
+    dirPath = Path(selection)
+    root.destroy()
+    return dirPath
+
+
+# }}}

defaults.toml

@@ -0,0 +1,4 @@
+#!/usr/bin/env python3
+from functions import multiVideoPipeline
+
+multiVideoPipeline("configuration.toml", "defaults.toml")

functions.py

@@ -1,4 +1,4 @@
 numpy>=2.0.0
 pandas
 pyCRT>=2.2.1
-tomli
+tomli; python_version < "3.11"