1+ """Meta-analysis functionality for Autonima."""
2+
3+ import argparse
4+ import os
5+ from pathlib import Path
6+ import json
7+ import importlib
8+ from nimare .correct import FDRCorrector , FWECorrector
9+ from nimare .workflows import CBMAWorkflow
10+ from nimare .meta .cbma import MKDADensity , ALE , KDA
11+ from nimare .nimads import Studyset , Annotation
12+ from nimare .reports .base import run_reports
13+
14+
15+ def create_estimator (estimator_name , estimator_args ):
16+ """Create an estimator instance based on the name and arguments."""
17+ # Map estimator names to classes
18+ estimator_map = {
19+ "ale" : ALE ,
20+ "mkdadensity" : MKDADensity ,
21+ "kda" : KDA ,
22+ }
23+
24+ if estimator_name not in estimator_map :
25+ raise ValueError (f"Unsupported estimator: { estimator_name } " )
26+
27+ estimator_class = estimator_map [estimator_name ]
28+ return estimator_class (** estimator_args )
29+
30+
31+ def create_corrector (corrector_name , corrector_args ):
32+ """Create a corrector instance based on the name and arguments."""
33+ # Map corrector names to classes
34+ if corrector_name == "fdr" :
35+ return FDRCorrector (** corrector_args )
36+ elif corrector_name == "montecarlo" :
37+ return FWECorrector (method = "montecarlo" , ** corrector_args )
38+ elif corrector_name == "bonferroni" :
39+ return FWECorrector (method = "bonferroni" , ** corrector_args )
40+ else :
41+ raise ValueError (f"Unsupported corrector: { corrector_name } " )
42+
43+
44+ def find_nimads_files (output_folder ):
45+ """Find NiMADS StudySet and Annotation JSON files in the output folder."""
46+ output_path = Path (output_folder )
47+ studyset_file = output_path / "nimads_studyset.json"
48+ annotation_file = output_path / "nimads_annotation.json"
49+
50+ if not studyset_file .exists ():
51+ raise FileNotFoundError (f"StudySet file not found: { studyset_file } " )
52+
53+ if not annotation_file .exists ():
54+ raise FileNotFoundError (f"Annotation file not found: { annotation_file } " )
55+
56+ return str (studyset_file ), str (annotation_file )
57+
58+
59+ def run_meta_analysis_for_column (studyset , annotation , annotation_data , column , output_dir ,
60+ estimator_name = "mkdadensity" , estimator_args = None ,
61+ corrector_name = "fdr" , corrector_args = None ):
62+ """Run meta-analysis for a specific annotation column."""
63+ print (f"Running meta-analysis for column: { column } " )
64+
65+ # Get column type from the original annotation data
66+ column_type = annotation_data ["note_keys" ].get (column )
67+ if column_type is None :
68+ print (f"Column { column } not found in annotation. Skipping." )
69+ return
70+
71+ # Only process boolean columns
72+ if column_type != "boolean" :
73+ print (f"Column { column } is not boolean. Skipping." )
74+ return
75+
76+ # Get analysis ids for the studies to include
77+ analysis_ids = [
78+ n ["analysis" ] for n in annotation_data ["notes" ] if n ["note" ].get (column )
79+ ]
80+
81+ if not analysis_ids :
82+ print (f"No studies found for column { column } . Skipping." )
83+ return
84+
85+ # Slice the studyset to include only selected studies
86+ first_studyset = studyset .slice (analyses = analysis_ids )
87+
88+ # Switch studyset.name with studyset.id to ensure uniqueness
89+ for study in first_studyset .studies :
90+ study .name = study .id
91+
92+ # Switch analysis names to IDs to ensure uniqueness
93+ for analysis in study .analyses :
94+ analysis .name = analysis .id
95+
96+
97+ # Convert to dataset
98+ first_dataset = first_studyset .to_dataset ()
99+
100+ # Set up estimator and corrector
101+ estimator = create_estimator (estimator_name , estimator_args or {})
102+ corrector = create_corrector (corrector_name , corrector_args or {})
103+
104+ # Run meta-analysis
105+ workflow = CBMAWorkflow (
106+ estimator = estimator ,
107+ corrector = corrector ,
108+ diagnostics = "focuscounter" ,
109+ output_dir = output_dir ,
110+ )
111+
112+ meta_results = workflow .fit (first_dataset )
113+
114+ run_reports (meta_results , output_dir )
115+
116+ return meta_results
117+
118+
119+ def run_meta_analyses (output_folder , estimator_name = "mkdadensity" , estimator_args = None ,
120+ corrector_name = "fdr" , corrector_args = None ):
121+ """Run meta-analyses on all boolean annotation columns in the NiMADS files."""
122+ # Find the NiMADS files
123+ studyset_file , annotation_file = find_nimads_files (output_folder )
124+
125+ # Create output directory
126+ output_dir = Path (output_folder ) / "meta_analysis_results"
127+ output_dir .mkdir (exist_ok = True )
128+
129+ # Load the JSON data
130+ print ("Loading studyset JSON..." )
131+ with open (studyset_file , 'r' ) as f :
132+ studyset_data = json .load (f )
133+
134+ print ("Loading annotation JSON..." )
135+ with open (annotation_file , 'r' ) as f :
136+ annotation_data = json .load (f )
137+
138+ # Process the files using NiMARE classes
139+ print ("Creating studyset..." )
140+ studyset = Studyset (studyset_data )
141+
142+ print ("Creating annotation..." )
143+ annotation = Annotation (annotation_data , studyset )
144+
145+ # Get all boolean columns from the original annotation data
146+ boolean_columns = [
147+ col for col , col_type in annotation_data ["note_keys" ].items ()
148+ if col_type == "boolean"
149+ ]
150+
151+ print (f"Found { len (boolean_columns )} boolean columns: { boolean_columns } " )
152+
153+ # Run meta-analysis for each boolean column
154+ results = {}
155+ for column in boolean_columns :
156+ column_output_dir = output_dir / column
157+ column_output_dir .mkdir (exist_ok = True )
158+ try :
159+ meta_results = run_meta_analysis_for_column (
160+ studyset , annotation , annotation_data , column , str (column_output_dir ),
161+ estimator_name , estimator_args , corrector_name , corrector_args
162+ )
163+ results [column ] = meta_results
164+ except Exception as e :
165+ print (f"Error running meta-analysis for column { column } : { e } " )
166+ import traceback
167+ traceback .print_exc ()
168+
169+ return results
170+
171+
172+ def main ():
173+ """Main function to run meta-analyses from command line."""
174+ parser = argparse .ArgumentParser (description = "Run meta-analyses on autonima output" )
175+ parser .add_argument (
176+ "output_folder" ,
177+ help = "Path to the autonima output folder containing NiMADS files"
178+ )
179+
180+ # Add arguments for estimator
181+ parser .add_argument (
182+ "--estimator" ,
183+ choices = ["ale" , "mkdadensity" , "kda" ],
184+ default = "mkdadensity" ,
185+ help = "CBMA estimator to use (default: mkdadensity)"
186+ )
187+
188+ parser .add_argument (
189+ "--estimator-args" ,
190+ type = str ,
191+ default = "{}" ,
192+ help = "JSON string of arguments for the estimator (default: {})"
193+ )
194+
195+ # Add arguments for corrector
196+ parser .add_argument (
197+ "--corrector" ,
198+ choices = ["fdr" , "montecarlo" , "bonferroni" ],
199+ default = "fdr" ,
200+ help = "Corrector to use (default: fdr)"
201+ )
202+
203+ parser .add_argument (
204+ "--corrector-args" ,
205+ type = str ,
206+ default = "{}" ,
207+ help = "JSON string of arguments for the corrector (default: {})"
208+ )
209+
210+ args = parser .parse_args ()
211+
212+ # Parse estimator and corrector arguments
213+ try :
214+ estimator_args = json .loads (args .estimator_args )
215+ corrector_args = json .loads (args .corrector_args )
216+ except json .JSONDecodeError as e :
217+ print (f"Error parsing JSON arguments: { e } " )
218+ return
219+
220+ # Check if output folder exists
221+ if not os .path .exists (args .output_folder ):
222+ print (f"Error: Output folder { args .output_folder } does not exist" )
223+ return
224+
225+ results = run_meta_analyses (
226+ args .output_folder ,
227+ estimator_name = args .estimator ,
228+ estimator_args = estimator_args ,
229+ corrector_name = args .corrector ,
230+ corrector_args = corrector_args
231+ )
232+ print (f"Completed meta-analyses for { len (results )} columns" )
233+
234+
235+ if __name__ == "__main__" :
236+ main ()
0 commit comments