Add all ids option to config

Author: adelavega

autonima/config.py

@@ -134,8 +134,14 @@ def _validate_config(self, config: PipelineConfig) -> None:
             ConfigurationError: If configuration is invalid
         """
         # Validate search configuration
-        if not config.search.query.strip():
-            raise ConfigurationError("Search query cannot be empty")
+        if config.search.pmids_file or config.search.pmids_list:
+            # PMID-based search
+            if config.search.query.strip():
+                raise ConfigurationError("Cannot specify both search query and PMIDs list/file")
+        else:
+            # Query-based search
+            if not config.search.query.strip():
+                raise ConfigurationError("Search query cannot be empty when not using PMIDs list/file")
 
         if config.search.max_results <= 0:
             raise ConfigurationError("max_results must be positive")

autonima/coordinates/processor.py

@@ -6,74 +6,25 @@
 from typing import List
 
 from .openai_client import CoordinateParsingClient
-from .schema import Analysis
 
 logger = logging.getLogger(__name__)
 
 
 class CoordinateProcessor:
     """Processor for parsing coordinates from activation tables."""
 
-    def __init__(self, model: str = "gpt-4o-mini"):
+    def __init__(self, model: str = "gpt-4o-mini", 
+                 path_preference: List[str] = ['table_raw_path', 'table_data_path']):
         """
         Initialize the coordinate processor.
         
         Args:
             model: The model to use for parsing
         """
         self.model = model
+        self.path_preference = path_preference
         self.client = CoordinateParsingClient()
-    
-    def process_study(self, study):
-        """
-        Process all activation tables for a study and extract analyses.
-        
-        Args:
-            study: The study to process
-            
-        Returns:
-            List of analyses extracted from the study's tables
-        """
-        # Import locally to avoid circular imports
-        from autonima.models.types import Study, ActivationTable
-        
-        if not study.activation_tables:
-            return []
-        
-        all_analyses = []
-        
-        for table in study.activation_tables:
-            try:
-                # Load the table data
-                table_path = Path(table.table_path)
-                if not table_path.exists():
-                    logger.warning(f"Table file not found: {table_path}")
-                    continue
-                
-                # Read the table as text
-                with open(table_path, "r", encoding="utf-8") as f:
-                    reader = csv.reader(f)
-                    rows = list(reader)
-                    table_text = "\n".join([",".join(r) for r in rows])
-                
-                # Create a prompt for the table
-                prompt = self._create_table_prompt(
-                    table_text,
-                    table_caption=table.table_caption or "",
-                    table_foot=table.table_foot or ""
-                )
-                
-                # Parse the table
-                result = self.client.parse_analyses(prompt, model=self.model)
-                
-                # Add the analyses to our list
-                all_analyses.extend(result.analyses)
-                
-            except Exception as e:
-                logger.warning(f"Error processing table {table.table_path}: {e}")
-                continue
-        
-        return all_analyses
+
 
     def process_single_table(self, table):
         """
@@ -87,7 +38,15 @@ def process_single_table(self, table):
         """
         try:
             # Load the table data
-            table_path = Path(table.table_path)
+            for path_attr in self.path_preference:
+                table_path_value = getattr(table, path_attr, None)
+                if table_path_value:
+                    table_path = Path(table_path_value)
+                    break
+            else:
+                logger.warning(f"No valid table path found for table: {table.table_id}")
+                return []
+            table_path = Path(table_path)
             if not table_path.exists():
                 logger.warning(f"Table file not found: {table_path}")
                 return []
@@ -107,7 +66,6 @@ def process_single_table(self, table):
 
             # Parse the table
             result = self.client.parse_analyses(prompt, model=self.model)
-            
             return result.analyses
 
         except Exception as e:

autonima/models/types.py

@@ -25,10 +25,10 @@ class ActivationTable:
     """Represents a table containing activation coordinates from a study."""
     table_id: str  # New identifier for the table
     table_label: str  # Label or identifier for the table
-    table_path: str  # Path to the raw table file (HTML/CSV/etc)
     table_caption: Optional[str] = None  # Caption of the table
     table_foot: Optional[str] = None  # Footer of the table
     table_data_path: Optional[str] = None  # Path to processed table data file
+    table_raw_path: Optional[str] = None  # Path to raw table data file
 
 
 @dataclass
@@ -86,10 +86,10 @@ def to_dict(self) -> Dict[str, Any]:
                 {
                     "table_id": table.table_id,  # Added table_id
                     "table_label": table.table_label,
-                    "table_path": table.table_path,
                     "table_caption": table.table_caption,
                     "table_foot": table.table_foot,
-                    "table_data_path": table.table_data_path
+                    "table_data_path": table.table_data_path,
+                    "table_raw_path": table.table_raw_path
                 } for table in self.activation_tables
             ],
             "analyses": [
@@ -145,6 +145,8 @@ class SearchConfig:
     date_from: Optional[str] = None
     date_to: Optional[str] = None
     email: Optional[str] = None  # Required for NCBI API
+    pmids_file: Optional[str] = None  # Path to file with PMIDs (one per line)
+    pmids_list: Optional[List[str]] = None  # Direct list of PMIDs
 
 
 @dataclass
@@ -201,7 +203,6 @@ class OutputConfig:
     nimads: bool = False
 
 
-
 @dataclass
 class PipelineConfig:
     """Main configuration for the Autonima pipeline."""
@@ -222,6 +223,8 @@ def to_dict(self) -> Dict[str, Any]:
                 "date_from": self.search.date_from,
                 "date_to": self.search.date_to,
                 "email": self.search.email,
+                "pmids_file": self.search.pmids_file,
+                "pmids_list": self.search.pmids_list,
             },
             "screening": {
                 "abstract": self.screening.abstract,