PicassoBatchProcess/main.py at main · HeilemannLab/PicassoBatchProcess · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import yaml
import os
from bg_filter import process_filtering
from linking import process_linking
from dbscan import process_dbscan_clustering
from mean_std_frame_filter import process_mean_std_filtering
from nena import process_nena

###############################################################################################
# Config loading and main function
###############################################################################################

def main():
    # Load configuration file
    with open('config.yaml', 'r') as f:
        config = yaml.safe_load(f)

    # Get input files and desired mode

    input_dirs = config['paths']['input_dirs']
    mode = config['paths']['mode']
    file_endings = config['paths']['file_endings']

    # Get pipeline steps configuration
    steps = config.get('pipeline_steps', {})

    for input_folder in input_dirs:

        # Step 1: Background signal Filtering
        if steps.get('bg_filtering', True):  # Default to True if not specified
            print("\n=== Step 1: Background signal Filtering ===")
            filter_params = config['bg_signal_filtering'].copy()
            input_extension = filter_params.pop('input_extension')
            output_extension = filter_params.pop('output_extension')

            process_filtering(
                input_folder=input_folder,
                filter_params=filter_params,
                input_extension=input_extension,
                output_extension=output_extension
            )

        # Step 2: NeNA values calculation
        nena_data = None
        if steps.get('nena', True):
            print("\n=== Step 2: NeNA Calculation ===")
            nena_data = process_nena(
                input_folder=input_folder,
                input_extension=config['nena']['input_extension'],
                pixelsize=config['pixelsize']['pixelsize']
            )

        # Step 3: Linking
        if steps.get('linking', True):
            if nena_data is None and steps.get('nena', True) is False:
                print("\nWarning: Linking step requires NeNA calculation. Running NeNA first...")
                nena_data = process_nena(
                    input_folder=input_folder,
                    input_extension=f"{input_extension}{output_extension}",
                    pixelsize=config['pixelsize']['pixelsize']
                )

            print("\n=== Step 3: Linking ===")
            process_linking(
                input_folder=input_folder,
                input_extension=config['linking']['input_extension'],
                output_extension=config['linking']['output_extension'],
                max_frame_gap=config['linking']['max_frame_gap'],
                nena_multiplier=config['linking']['nena_multiplier'],
                nena_data=nena_data
            )

        # Step 4: DBSCAN Clustering
        if steps.get('dbscan', True):
            if nena_data is None and steps.get('nena', True) is False:
                print("\nWarning: DBSCAN step requires NeNA calculation. Running NeNA first...")
                nena_data = process_nena(
                    input_folder=input_folder,
                    input_extension=f"{input_extension}{output_extension}",
                    pixelsize=config['pixelsize']['pixelsize']
                )

            print("\n=== Step 4: DBSCAN Clustering ===")
            process_dbscan_clustering(
                input_folder=input_folder,
                input_extension=config['dbscan']['input_extension'],
                output_clusters_extension=config['dbscan']['output_clusters_extension'],
                output_centers_extension=config['dbscan']['output_centers_extension'],
                min_samples=config['dbscan']['min_samples'],
                nena_multiplier=config['dbscan']['nena_multiplier'],
                pixelsize=config['pixelsize']['pixelsize'],
                nena_data=nena_data
            )

        # Step 5: Mean/Std Frame Filtering
        if steps.get('mean_std', True):
            print("\n=== Step 5: Mean/Std Frame Filtering ===")
            filter_params = config['mean_std_filtering'].copy()
            clusters_extension = filter_params.pop('clusters_extension')
            centers_extension = filter_params.pop('centers_extension')
            output_extension = filter_params.pop('output_extension')


            process_mean_std_filtering(
                input_folder=input_folder,
                filter_params=filter_params,
                clusters_extension=clusters_extension,
                centers_extension=centers_extension,
                output_extension=output_extension,
                mode=mode
            )

        print("\nSelected pipeline steps completed successfully!")

if __name__ == "__main__":
    main()