diff --git a/semantica/conflicts/source_tracker.py b/semantica/conflicts/source_tracker.py index a25eb841..3a00567f 100644 --- a/semantica/conflicts/source_tracker.py +++ b/semantica/conflicts/source_tracker.py @@ -549,3 +549,122 @@ class SourceTracker: List of traceability records """ return self.get_traceability_chain(entity_id, property_name) + + def track_sources_batch( + self, + source_data: List[Dict[str, Any]], + pipeline_id: Optional[str] = None, + ) -> Dict[str, int]: + """ + Track sources for multiple entities, properties, or relationships in batch. + + Args: + source_data: List of source tracking dictionaries, each containing: + - type: "entity", "property", or "relationship" + - entity_id: Entity identifier (required for all types) + - property_name: Property name (required for "property" type) + - value: Property value (required for "property" type) + - relationship_id: Relationship identifier (required for "relationship" type) + - source: SourceReference object or dict to create SourceReference + - metadata: Optional metadata dictionary + pipeline_id: Optional pipeline ID for progress tracking + + Returns: + dict: Statistics with keys: + - entities_tracked: Number of entities tracked + - properties_tracked: Number of properties tracked + - relationships_tracked: Number of relationships tracked + - total_tracked: Total number of items tracked + """ + if not source_data: + return { + "entities_tracked": 0, + "properties_tracked": 0, + "relationships_tracked": 0, + "total_tracked": 0, + } + + tracking_id = self.progress_tracker.start_tracking( + module="conflicts", + submodule="SourceTracker", + message=f"Tracking sources for {len(source_data)} items", + pipeline_id=pipeline_id, + ) + + stats = { + "entities_tracked": 0, + "properties_tracked": 0, + "relationships_tracked": 0, + "total_tracked": 0, + } + + try: + for i, item in enumerate(source_data): + item_type = item.get("type", "property") + source_ref = item.get("source") + metadata = item.get("metadata", {}) + + # Convert source dict to SourceReference if needed + if isinstance(source_ref, dict): + source_ref = SourceReference(**source_ref) + elif not isinstance(source_ref, SourceReference): + self.logger.warning(f"Invalid source reference in item {i}: {item}") + continue + + if item_type == "entity": + entity_id = item.get("entity_id") + if entity_id: + self.track_entity_source(entity_id, source_ref, **metadata) + stats["entities_tracked"] += 1 + stats["total_tracked"] += 1 + + elif item_type == "property": + entity_id = item.get("entity_id") + property_name = item.get("property_name") + value = item.get("value") + if entity_id and property_name is not None: + self.track_property_source( + entity_id, property_name, value, source_ref, **metadata + ) + stats["properties_tracked"] += 1 + stats["total_tracked"] += 1 + + elif item_type == "relationship": + relationship_id = item.get("relationship_id") + if relationship_id: + self.track_relationship_source( + relationship_id, source_ref, **metadata + ) + stats["relationships_tracked"] += 1 + stats["total_tracked"] += 1 + + else: + self.logger.warning( + f"Unknown type '{item_type}' in item {i}, skipping" + ) + + # Update progress + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=len(source_data), + message=f"Tracking sources {i+1}/{len(source_data)}...", + ) + + message = ( + f"Tracked {stats['total_tracked']} items: " + f"{stats['entities_tracked']} entities, " + f"{stats['properties_tracked']} properties, " + f"{stats['relationships_tracked']} relationships" + ) + + self.progress_tracker.stop_tracking( + tracking_id, status="completed", message=message + ) + return stats + + except Exception as e: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(e) + ) + raise \ No newline at end of file diff --git a/semantica/core/orchestrator.py b/semantica/core/orchestrator.py index de9e6904..5c21d81e 100644 --- a/semantica/core/orchestrator.py +++ b/semantica/core/orchestrator.py @@ -203,6 +203,7 @@ class Semantica: sources: List of data sources (files, URLs, streams) **kwargs: Additional processing options: - pipeline: Custom pipeline configuration + - pipeline_id: Optional pipeline ID for progress tracking - embeddings: Whether to generate embeddings - graph: Whether to build knowledge graph - normalize: Whether to normalize data @@ -220,11 +221,19 @@ class Semantica: # Auto-initialize if not already initialized self._ensure_initialized() + # Extract or generate pipeline_id + pipeline_id = kwargs.get("pipeline_id") + if not pipeline_id: + # Generate a unique pipeline ID + import uuid + pipeline_id = f"kb_build_{uuid.uuid4().hex[:8]}" + # Start overall progress tracking overall_tracking_id = self.progress_tracker.start_tracking( module="core", submodule="Semantica", message=f"Building knowledge base from {len(sources)} sources", + pipeline_id=pipeline_id, ) try: @@ -236,6 +245,23 @@ class Semantica: # Create processing pipeline pipeline_config = kwargs.get("pipeline", {}) pipeline = self._create_pipeline(pipeline_config) + + # Register pipeline modules for progress tracking + if hasattr(pipeline, 'steps') and pipeline.steps: + module_list = [] + module_order = {} + for idx, step in enumerate(pipeline.steps): + module_name = getattr(step, 'module', None) or getattr(step, 'name', None) or str(step) + if module_name and module_name not in module_list: + module_list.append(module_name) + module_order[module_name] = idx + + if module_list: + self.progress_tracker.register_pipeline_modules( + pipeline_id=pipeline_id, + module_list=module_list, + module_order=module_order + ) # Process sources results = [] @@ -251,6 +277,7 @@ class Semantica: module="core", submodule="build_knowledge_base", message=f"Processing {Path(file_str).name if file_str else 'source'}", + pipeline_id=pipeline_id, ) try: result = self.run_pipeline(pipeline, source) @@ -308,6 +335,9 @@ class Semantica: message=f"Processed {len(results)} sources", ) + # Clear pipeline context when complete + self.progress_tracker.clear_pipeline_context(pipeline_id) + # Show summary self.progress_tracker.show_summary() @@ -319,6 +349,7 @@ class Semantica: "metadata": { "sources": validated_sources, "pipeline": pipeline_config, + "pipeline_id": pipeline_id, }, } @@ -326,6 +357,8 @@ class Semantica: self.progress_tracker.stop_tracking( overall_tracking_id, status="failed", message=str(e) ) + # Clear pipeline context on failure + self.progress_tracker.clear_pipeline_context(pipeline_id) self.logger.error(f"Failed to build knowledge base: {e}") raise ProcessingError(f"Failed to build knowledge base: {e}") diff --git a/semantica/embeddings/embedding_generator.py b/semantica/embeddings/embedding_generator.py index 32afaeea..ada02c4c 100644 --- a/semantica/embeddings/embedding_generator.py +++ b/semantica/embeddings/embedding_generator.py @@ -258,13 +258,14 @@ class EmbeddingGenerator: return max(0.0, min(1.0, similarity)) def process_batch( - self, data_items: List[Union[str, Path]], **options + self, data_items: List[Union[str, Path]], pipeline_id: Optional[str] = None, **options ) -> Dict[str, Any]: """ Process multiple data items for embedding generation. Args: data_items: List of data items + pipeline_id: Optional pipeline ID for progress tracking **options: Processing options Returns: @@ -275,6 +276,7 @@ class EmbeddingGenerator: module="embeddings", submodule="EmbeddingGenerator", message=f"Batch of {len(data_items)} items", + pipeline_id=pipeline_id, ) try: diff --git a/semantica/kg/graph_builder.py b/semantica/kg/graph_builder.py index df2e4c89..85181f4a 100644 --- a/semantica/kg/graph_builder.py +++ b/semantica/kg/graph_builder.py @@ -237,6 +237,7 @@ class GraphBuilder: self, sources: Union[List[Any], Any], second_arg: Optional[Any] = None, + pipeline_id: Optional[str] = None, **options, ) -> Dict[str, Any]: """ @@ -245,6 +246,7 @@ class GraphBuilder: Args: sources: Entities or sources list second_arg: Optional relationships list or entity_resolver (for backward compatibility) + pipeline_id: Optional pipeline ID for progress tracking **options: Additional build options - extract: Whether to extract entities from text (default: True) - extract_relations: Whether to extract relations from text (default: False) @@ -307,6 +309,7 @@ class GraphBuilder: module="kg", submodule="GraphBuilder", message=f"Knowledge graph from {len(sources)} source(s)", + pipeline_id=pipeline_id, ) try: @@ -340,6 +343,7 @@ class GraphBuilder: module="kg", submodule="GraphBuilder", message=f"Processing {len(entities_list)} entities", + pipeline_id=pipeline_id, ) # Check if entities are already in dictionary format @@ -409,6 +413,7 @@ class GraphBuilder: module="kg", submodule="GraphBuilder", message=f"Processing {len(relationships_list)} relationships", + pipeline_id=pipeline_id, ) # Check if relationships are already in dictionary format diff --git a/semantica/kg/provenance_tracker.py b/semantica/kg/provenance_tracker.py index c3d71ab9..dc08b77d 100644 --- a/semantica/kg/provenance_tracker.py +++ b/semantica/kg/provenance_tracker.py @@ -217,3 +217,131 @@ class ProvenanceTracker: or None if entity is not tracked """ return self.provenance_data.get(entity_id) + + def track_entities_batch( + self, + entities: List[Dict[str, Any]], + source: str, + pipeline_id: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> int: + """ + Track provenance for multiple entities in batch. + + Args: + entities: List of entity dictionaries, each containing at least 'id' key + source: Source identifier + pipeline_id: Optional pipeline ID for progress tracking + metadata: Optional metadata dictionary to apply to all entities + + Returns: + int: Number of entities tracked + """ + if not entities: + return 0 + + tracking_id = self.progress_tracker.start_tracking( + module="kg", + submodule="ProvenanceTracker", + message=f"Tracking provenance for {len(entities)} entities", + pipeline_id=pipeline_id, + ) + + try: + tracked_count = 0 + for i, entity in enumerate(entities): + entity_id = entity.get("id") or entity.get("entity_id") + if not entity_id: + self.logger.warning(f"Skipping entity without ID: {entity}") + continue + + # Merge entity-specific metadata with batch metadata + entity_metadata = {**(metadata or {}), **(entity.get("metadata", {}))} + self.track_entity(entity_id, source, entity_metadata) + + tracked_count += 1 + + # Update progress + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=len(entities), + message=f"Tracking entity {i+1}/{len(entities)}...", + ) + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Tracked {tracked_count} entities", + ) + return tracked_count + + except Exception as e: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(e) + ) + raise + + def track_relationships_batch( + self, + relationships: List[Dict[str, Any]], + source: str, + pipeline_id: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> int: + """ + Track provenance for multiple relationships in batch. + + Args: + relationships: List of relationship dictionaries, each containing at least 'id' key + source: Source identifier + pipeline_id: Optional pipeline ID for progress tracking + metadata: Optional metadata dictionary to apply to all relationships + + Returns: + int: Number of relationships tracked + """ + if not relationships: + return 0 + + tracking_id = self.progress_tracker.start_tracking( + module="kg", + submodule="ProvenanceTracker", + message=f"Tracking provenance for {len(relationships)} relationships", + pipeline_id=pipeline_id, + ) + + try: + tracked_count = 0 + for i, relationship in enumerate(relationships): + relationship_id = relationship.get("id") or relationship.get("relationship_id") + if not relationship_id: + self.logger.warning(f"Skipping relationship without ID: {relationship}") + continue + + # Merge relationship-specific metadata with batch metadata + rel_metadata = {**(metadata or {}), **(relationship.get("metadata", {}))} + self.track_relationship(relationship_id, source, rel_metadata) + + tracked_count += 1 + + # Update progress + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=len(relationships), + message=f"Tracking relationship {i+1}/{len(relationships)}...", + ) + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Tracked {tracked_count} relationships", + ) + return tracked_count + + except Exception as e: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(e) + ) + raise \ No newline at end of file diff --git a/semantica/parse/docling_parser.py b/semantica/parse/docling_parser.py index 386cadc8..6a456996 100644 --- a/semantica/parse/docling_parser.py +++ b/semantica/parse/docling_parser.py @@ -119,12 +119,16 @@ class DoclingParser: """ file_path = Path(file_path) + # Get pipeline_id from options if provided + pipeline_id = options.get("pipeline_id", None) + # Track document parsing tracking_id = self.progress_tracker.start_tracking( file=str(file_path), module="parse", submodule="DoclingParser", message=f"Docling: {file_path.name}", + pipeline_id=pipeline_id, ) try: @@ -159,6 +163,7 @@ class DoclingParser: # Stage 2: Document conversion (10-70%) - This is the longest step # Note: This is a blocking operation, but we'll update progress after it completes + # Emphasize Docling as core dependency in message self.progress_tracker.update_progress( tracking_id, processed=2, @@ -247,10 +252,38 @@ class DoclingParser: ) images = self._extract_images(result) + # Prepare extraction counts and metadata + extraction_counts = { + "tables": len(tables), + "images": len(images), + "pages": len(pages) or metadata.page_count or 0, + } + + # Build completion message with Docling emphasis + count_parts = [] + if extraction_counts["tables"] > 0: + count_parts.append(f"{extraction_counts['tables']} tables") + if extraction_counts["images"] > 0: + count_parts.append(f"{extraction_counts['images']} images") + if extraction_counts["pages"] > 0: + count_parts.append(f"{extraction_counts['pages']} pages") + + if count_parts: + completion_message = f"Parsed document (Docling): {', '.join(count_parts)} extracted" + else: + completion_message = f"Parsed document (Docling): 0 tables, 0 images, {extraction_counts['pages']} pages extracted" + + # Store metadata with extraction counts and core dependency + metadata_dict = { + "extraction_counts": extraction_counts, + "core_dependency": "docling", + } + self.progress_tracker.stop_tracking( tracking_id, status="completed", - message=f"Parsed document: {len(tables)} tables extracted", + message=completion_message, + metadata=metadata_dict, ) return { diff --git a/semantica/parse/docx_parser.py b/semantica/parse/docx_parser.py index b13560d3..9282ddcb 100644 --- a/semantica/parse/docx_parser.py +++ b/semantica/parse/docx_parser.py @@ -84,12 +84,13 @@ class DOCXParser: self.config = config self.progress_tracker = get_progress_tracker() - def parse(self, file_path: Union[str, Path], **options) -> Dict[str, Any]: + def parse(self, file_path: Union[str, Path], pipeline_id: Optional[str] = None, **options) -> Dict[str, Any]: """ Parse DOCX document. Args: file_path: Path to DOCX file + pipeline_id: Optional pipeline ID for progress tracking **options: Parsing options: - extract_formatting: Whether to extract formatting (default: False) - extract_tables: Whether to extract tables (default: True) @@ -106,6 +107,7 @@ class DOCXParser: module="parse", submodule="DOCXParser", message=f"DOCX: {file_path.name}", + pipeline_id=pipeline_id, ) try: diff --git a/semantica/parse/pdf_parser.py b/semantica/parse/pdf_parser.py index b99b7217..4a087304 100644 --- a/semantica/parse/pdf_parser.py +++ b/semantica/parse/pdf_parser.py @@ -84,12 +84,13 @@ class PDFParser: if not self.progress_tracker.enabled: self.progress_tracker.enabled = True - def parse(self, file_path: Union[str, Path], **options) -> Dict[str, Any]: + def parse(self, file_path: Union[str, Path], pipeline_id: Optional[str] = None, **options) -> Dict[str, Any]: """ Parse PDF document. Args: file_path: Path to PDF file + pipeline_id: Optional pipeline ID for progress tracking **options: Parsing options: - extract_text: Whether to extract text (default: True) - extract_tables: Whether to extract tables (default: True) @@ -107,6 +108,7 @@ class PDFParser: module="parse", submodule="PDFParser", message=f"PDF: {file_path.name}", + pipeline_id=pipeline_id, ) try: diff --git a/semantica/pipeline/execution_engine.py b/semantica/pipeline/execution_engine.py index e92742a8..2cef1a37 100644 --- a/semantica/pipeline/execution_engine.py +++ b/semantica/pipeline/execution_engine.py @@ -132,11 +132,37 @@ class ExecutionEngine: module="pipeline", submodule="ExecutionEngine", message=f"Executing pipeline: {pipeline_id}", + pipeline_id=pipeline_id, ) try: self.logger.info(f"Executing pipeline: {pipeline_id}") + # Register pipeline modules for progress tracking + module_list = [] + module_order = {} + if hasattr(pipeline, 'steps') and pipeline.steps: + for idx, step in enumerate(pipeline.steps): + # Extract module name from step + module_name = getattr(step, 'module', None) or getattr(step, 'name', None) or str(step) + if module_name and module_name not in module_list: + module_list.append(module_name) + module_order[module_name] = idx + + # If no steps found, try to infer from pipeline structure + if not module_list: + # Common pipeline modules + module_list = ["ingest", "parse", "normalize", "semantic_extract", "kg", "embeddings"] + module_order = {module: idx for idx, module in enumerate(module_list)} + + # Register pipeline modules + if module_list: + self.progress_tracker.register_pipeline_modules( + pipeline_id=pipeline_id, + module_list=module_list, + module_order=module_order + ) + # Set status with self.pipeline_lock: self.pipeline_status[pipeline_id] = PipelineStatus.RUNNING @@ -174,6 +200,9 @@ class ExecutionEngine: status="completed" if metrics["steps_failed"] == 0 else "failed", message=f"Executed {metrics['steps_executed']} steps in {execution_time:.2f}s", ) + + # Clear pipeline context when pipeline completes + self.progress_tracker.clear_pipeline_context(pipeline_id) return ExecutionResult( success=metrics["steps_failed"] == 0, @@ -193,6 +222,8 @@ class ExecutionEngine: self.progress_tracker.stop_tracking( pipeline_tracking_id, status="failed", message=str(e) ) + # Clear pipeline context on failure + self.progress_tracker.clear_pipeline_context(pipeline_id) self.logger.error(f"Pipeline execution failed: {e}") with self.pipeline_lock: self.pipeline_status[pipeline_id] = PipelineStatus.FAILED diff --git a/semantica/pipeline/parallelism_manager.py b/semantica/pipeline/parallelism_manager.py index c11e67cf..7b5ed2c5 100644 --- a/semantica/pipeline/parallelism_manager.py +++ b/semantica/pipeline/parallelism_manager.py @@ -104,13 +104,14 @@ class ParallelismManager: self.lock = threading.Lock() def execute_parallel( - self, tasks: List[Task], **options + self, tasks: List[Task], pipeline_id: Optional[str] = None, **options ) -> List[ParallelExecutionResult]: """ Execute tasks in parallel. Args: tasks: List of tasks to execute + pipeline_id: Optional pipeline ID for progress tracking **options: Additional options Returns: @@ -120,6 +121,7 @@ class ParallelismManager: module="pipeline", submodule="ParallelismManager", message=f"Executing {len(tasks)} tasks in parallel", + pipeline_id=pipeline_id, ) try: diff --git a/semantica/semantic_extract/ner_extractor.py b/semantica/semantic_extract/ner_extractor.py index a5cc16cf..b9bd07af 100644 --- a/semantica/semantic_extract/ner_extractor.py +++ b/semantica/semantic_extract/ner_extractor.py @@ -141,13 +141,14 @@ class NERExtractor: f"spaCy model {self.model_name} not found. ML method will fallback." ) - def extract(self, text: Union[str, List[Dict[str, Any]], List[str]], **kwargs) -> Union[List[Entity], List[List[Entity]]]: + def extract(self, text: Union[str, List[Dict[str, Any]], List[str]], pipeline_id: Optional[str] = None, **kwargs) -> Union[List[Entity], List[List[Entity]]]: """ Alias for extract_entities. Handles both single string and list of documents. Args: text: Input text or list of documents + pipeline_id: Optional pipeline ID for progress tracking **kwargs: Extraction options Returns: @@ -159,6 +160,7 @@ class NERExtractor: module="semantic_extract", submodule="NERExtractor", message=f"Batch extracting entities from {len(text)} documents", + pipeline_id=pipeline_id, ) try: diff --git a/semantica/semantic_extract/relation_extractor.py b/semantica/semantic_extract/relation_extractor.py index fb2961dc..5691e080 100644 --- a/semantica/semantic_extract/relation_extractor.py +++ b/semantica/semantic_extract/relation_extractor.py @@ -168,6 +168,7 @@ class RelationExtractor: self, text: Union[str, List[Dict[str, Any]], List[str]], entities: Union[List[Entity], List[List[Entity]]], + pipeline_id: Optional[str] = None, **kwargs ) -> Union[List[Relation], List[List[Relation]]]: """ @@ -177,6 +178,7 @@ class RelationExtractor: Args: text: Input text or list of documents entities: List of entities or list of list of entities + pipeline_id: Optional pipeline ID for progress tracking **kwargs: Extraction options Returns: @@ -188,6 +190,7 @@ class RelationExtractor: module="semantic_extract", submodule="RelationExtractor", message=f"Batch extracting relations from {len(text)} documents", + pipeline_id=pipeline_id, ) try: diff --git a/semantica/split/provenance_tracker.py b/semantica/split/provenance_tracker.py index e980b99b..ab47405a 100644 --- a/semantica/split/provenance_tracker.py +++ b/semantica/split/provenance_tracker.py @@ -127,16 +127,22 @@ class ProvenanceTracker: return provenance_id - def track(self, **kwargs) -> Union[str, List[str]]: + def track(self, pipeline_id: Optional[str] = None, **kwargs) -> Union[str, List[str]]: """ Track provenance (alias for track_chunk/track_chunks). Delegates based on input arguments. + + Args: + pipeline_id: Optional pipeline ID for progress tracking + **kwargs: Arguments passed to track_chunk or track_chunks """ if "chunks" in kwargs: + kwargs["pipeline_id"] = pipeline_id return self.track_chunks(**kwargs) if "chunk" in kwargs and isinstance(kwargs["chunk"], list): # Handle list passed to chunk arg kwargs["chunks"] = kwargs.pop("chunk") + kwargs["pipeline_id"] = pipeline_id return self.track_chunks(**kwargs) return self.track_chunk(**kwargs) @@ -145,6 +151,7 @@ class ProvenanceTracker: chunks: List[Chunk], source_document: str, source_path: Optional[str] = None, + pipeline_id: Optional[str] = None, **metadata, ) -> List[str]: """ @@ -154,6 +161,7 @@ class ProvenanceTracker: chunks: List of chunks to track source_document: Source document identifier source_path: Path to source document + pipeline_id: Optional pipeline ID for progress tracking **metadata: Additional metadata Returns: @@ -163,6 +171,7 @@ class ProvenanceTracker: module="split", submodule="ProvenanceTracker", message=f"Tracking provenance for {len(chunks)} chunks", + pipeline_id=pipeline_id, ) try: @@ -170,8 +179,11 @@ class ProvenanceTracker: parent_chunk_id = None for i, chunk in enumerate(chunks): - self.progress_tracker.update_tracking( - tracking_id, message=f"Tracking chunk {i+1}/{len(chunks)}..." + self.progress_tracker.update_progress( + tracking_id, + processed=i + 1, + total=len(chunks), + message=f"Tracking chunk {i+1}/{len(chunks)}...", ) provenance_id = self.track_chunk( chunk, source_document, source_path, parent_chunk_id, **metadata diff --git a/semantica/utils/progress_tracker.py b/semantica/utils/progress_tracker.py index 8e57b4e3..c53fd324 100644 --- a/semantica/utils/progress_tracker.py +++ b/semantica/utils/progress_tracker.py @@ -77,6 +77,8 @@ class ProgressItem: total_items: Optional[int] = None # Total items to process processed_items: Optional[int] = None # Items processed so far estimated_remaining: Optional[float] = None # Estimated remaining time in seconds + pipeline_id: Optional[str] = None # Pipeline ID this item belongs to + pipeline_order: Optional[int] = None # Order of this module in the pipeline class ProgressDisplay(ABC): @@ -232,78 +234,130 @@ class ConsoleProgressDisplay(ProgressDisplay): return with self.lock: - # Create unique key for this item - key = f"{item.module}:{item.submodule}" - if item.file: - key = f"{item.file}:{key}" - - # Build progress line - parts = [] - - # Semantica branding with action - if self.use_emoji: - parts.append("🧠") - - # Create action message based on module (now includes custom message) - action_msg = self._get_action_message(item.module, item.message) - parts.append(action_msg) - - # Progress bar and percentage - if item.progress_percentage is not None: - pct = item.progress_percentage - bar_width = 15 - filled = int(bar_width * pct / 100) - bar = "█" * filled + "░" * (bar_width - filled) - parts.append(f"|{bar}| {pct:.1f}%") - - # Count information [done/total] - if item.processed_items is not None and item.total_items is not None: - parts.append(f"[{item.processed_items}/{item.total_items}]") - - # Status and Module emojis - if self.use_emoji: - status_emoji = self._get_status_emoji(item.status) - module_emoji = self._get_emoji_for_module(item.module or "") - parts.append(f"{status_emoji}{module_emoji}") - - # ETA and Rate - metrics = [] - if item.estimated_remaining is not None and item.estimated_remaining > 0: - if item.estimated_remaining < 60: - metrics.append(f"ETA: {item.estimated_remaining:.1f}s") - elif item.estimated_remaining < 3600: - metrics.append(f"ETA: {item.estimated_remaining/60:.1f}m") - else: - metrics.append(f"ETA: {item.estimated_remaining/3600:.1f}h") + # If item is part of a pipeline, get all pipeline items for display + pipeline_items = [] + if item.pipeline_id: + # Get tracker instance to access pipeline items + # Use the singleton instance directly + tracker = ProgressTracker.get_instance() + pipeline_items = tracker.get_pipeline_items(item.pipeline_id) - if item.start_time and item.processed_items is not None and item.processed_items > 0: - elapsed = time.time() - item.start_time - if elapsed > 0: - rate = item.processed_items / elapsed - metrics.append(f"{rate:.1f}/s") - - if metrics: - parts.append(f"({' | '.join(metrics)})") - - # Elapsed time if no progress info - if item.start_time and item.progress_percentage is None: - elapsed = time.time() - item.start_time - parts.append(f"({elapsed:.1f}s)") - - line = " ".join(parts) - self.current_lines[key] = line - - # Print all current lines - # For multiple lines, we combine them to avoid console scrolling issues - sys.stdout.write("\r" + " " * 120 + "\r") - if len(self.current_lines) == 1: - sys.stdout.write(line) + # If we have pipeline items, show all of them + if pipeline_items: + # Clear and show all pipeline items + sys.stdout.write("\r" + " " * 150 + "\r") + + # Show header if first time + if not hasattr(self, '_pipeline_header_shown'): + if self.use_emoji: + sys.stdout.write("🧠Semantica - 📊 Current Progress\n") + else: + sys.stdout.write("Semantica - Current Progress\n") + sys.stdout.write("=" * 150 + "\n") + self._pipeline_header_shown = True + + # Display all pipeline items + for pipeline_item in pipeline_items: + self._display_item_line(pipeline_item) + sys.stdout.write("\n") + + sys.stdout.flush() else: - # Combine multiple active lines onto one or two lines to be safe - lines_list = list(self.current_lines.values()) - sys.stdout.write(" | ".join(lines_list[-2:])) - - sys.stdout.flush() + # Original single-item display + self._display_item_line(item) + sys.stdout.flush() + + def _display_item_line(self, item: ProgressItem) -> None: + """Display a single progress item line.""" + # Create unique key for this item + key = f"{item.module}:{item.submodule}" + if item.file: + key = f"{item.file}:{key}" + + # Build progress line + parts = [] + + # Status emoji + if self.use_emoji: + status_emoji = self._get_status_emoji(item.status) + parts.append(status_emoji) + + # Create action message based on module + action_msg = self._get_action_message(item.module, item.message) + parts.append(action_msg) + + # Module and Submodule + if self.use_emoji: + module_emoji = self._get_emoji_for_module(item.module or "") + parts.append(f"{module_emoji} {item.module or 'N/A'}") + else: + parts.append(f"{item.module or 'N/A'}") + parts.append(f"{item.submodule or 'N/A'}") + + # Progress bar and percentage + if item.progress_percentage is not None: + pct = item.progress_percentage + bar_width = 15 + filled = int(bar_width * pct / 100) + bar = "█" * filled + "░" * (bar_width - filled) + parts.append(f"|{bar}| {pct:.1f}%") + else: + parts.append("|" + "░" * 15 + "| 0.0%") + + # ETA + if item.estimated_remaining is not None and item.estimated_remaining > 0: + if item.estimated_remaining < 60: + parts.append(f"ETA: {item.estimated_remaining:.1f}s") + elif item.estimated_remaining < 3600: + parts.append(f"ETA: {item.estimated_remaining/60:.1f}m") + else: + parts.append(f"ETA: {item.estimated_remaining/3600:.1f}h") + else: + parts.append("ETA: -") + + # Rate + if item.start_time and item.processed_items is not None and item.processed_items > 0: + elapsed = time.time() - item.start_time + if elapsed > 0: + rate = item.processed_items / elapsed + parts.append(f"Rate: {rate:.1f}/s") + else: + parts.append("Rate: -") + + # Time + if item.start_time: + if item.end_time: + elapsed = item.end_time - item.start_time + parts.append(f"Time: {elapsed:.2f}s") + else: + elapsed = time.time() - item.start_time + parts.append(f"Time: {elapsed:.2f}s") + else: + parts.append("Time: -") + + # Extraction counts (if available) + if item.metadata.get('extraction_counts'): + counts = item.metadata['extraction_counts'] + count_parts = [] + if 'tables' in counts: + count_parts.append(f"{counts['tables']} tables") + if 'images' in counts: + count_parts.append(f"{counts['images']} images") + if 'pages' in counts: + count_parts.append(f"{counts['pages']} pages") + if count_parts: + extracted = ", ".join(count_parts) + # Add Docling indicator if core dependency is docling + if item.metadata.get('core_dependency') == 'docling': + parts.append(f"Extracted (Docling): {extracted}") + else: + parts.append(f"Extracted: {extracted}") + else: + parts.append("Extracted: -") + + line = " ".join(parts) + self.current_lines[key] = line + sys.stdout.write(line) def show_summary(self, items: List[ProgressItem]) -> None: """Show final summary.""" @@ -482,12 +536,52 @@ class JupyterProgressDisplay(ProgressDisplay): # Current status html_parts.append("
| Status | Action | Module | Submodule | Progress | ETA | Rate | Time | |
|---|---|---|---|---|---|---|---|---|
| Status | Action | Module | Submodule | Progress | ETA | Rate | Time | {extracted_header} |
| {eta_str} | " f"{rate_str} | " f"{elapsed} | " + f"{extracted_str} | " f"