mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
KaifAhmad1
58707ff721
- Robust ID extraction in CentralityCalculator, CommunityDetector, and ConnectivityAnalyzer - Support for direct Entity objects and dictionaries as node identifiers - Improved Entity hashability in utils/types.py - Added integration test to verify fix and prevent regression
278 lines
11 KiB
Python
278 lines
11 KiB
Python
import os
|
|
import sys
|
|
import unittest
|
|
import numpy as np
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
import pytest
|
|
# Add project root to path
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
|
|
|
|
from semantica.visualization import (
|
|
KGVisualizer,
|
|
OntologyVisualizer,
|
|
EmbeddingVisualizer,
|
|
SemanticNetworkVisualizer,
|
|
AnalyticsVisualizer,
|
|
TemporalVisualizer
|
|
)
|
|
from semantica.kg import GraphBuilder, GraphAnalyzer, TemporalVersionManager
|
|
from semantica.ontology import OntologyGenerator
|
|
from semantica.embeddings import EmbeddingGenerator
|
|
|
|
pytestmark = pytest.mark.integration
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger("reproduce_notebooks")
|
|
|
|
def run_introduction_notebook():
|
|
logger.info("Running Introduction Notebook steps...")
|
|
|
|
# Step 1: Knowledge Graph Visualization
|
|
logger.info("Step 1: Knowledge Graph Visualization")
|
|
kg_visualizer = KGVisualizer()
|
|
builder = GraphBuilder()
|
|
|
|
entities = [
|
|
{"id": "e1", "type": "Organization", "name": "Apple Inc.", "properties": {}},
|
|
{"id": "e2", "type": "Person", "name": "Tim Cook", "properties": {}}
|
|
]
|
|
|
|
relationships = [
|
|
{"source": "e2", "target": "e1", "type": "CEO_of", "properties": {}}
|
|
]
|
|
|
|
kg = builder.build([{"entities": entities, "relationships": relationships}])
|
|
viz = kg_visualizer.visualize_network(kg, output="interactive")
|
|
assert viz is not None, "KG visualization failed"
|
|
logger.info("KG Visualization successful")
|
|
|
|
# Step 2: Ontology Visualization
|
|
logger.info("Step 2: Ontology Visualization")
|
|
ontology_visualizer = OntologyVisualizer()
|
|
generator = OntologyGenerator(min_occurrences=1)
|
|
|
|
ontology = generator.generate_ontology({"entities": entities, "relationships": relationships})
|
|
viz = ontology_visualizer.visualize_hierarchy(ontology, output="interactive")
|
|
# Note: verify if None is expected if ontology is simple or empty, but here it should be fine
|
|
if viz is None:
|
|
logger.warning("Ontology visualization returned None (might be due to empty hierarchy)")
|
|
else:
|
|
logger.info("Ontology Visualization successful")
|
|
|
|
# Step 3: Embedding Visualization
|
|
logger.info("Step 3: Embedding Visualization")
|
|
embedding_visualizer = EmbeddingVisualizer()
|
|
# Mocking EmbeddingGenerator to avoid heavy model loading if possible,
|
|
# but let's try to use the real one if it falls back gracefully.
|
|
# If it fails, we will catch and use random embeddings.
|
|
try:
|
|
emb_generator = EmbeddingGenerator()
|
|
texts = ["Apple Inc.", "Microsoft Corporation", "Amazon"]
|
|
embeddings = emb_generator.generate_embeddings(texts, data_type="text")
|
|
except Exception as e:
|
|
logger.warning(f"Embedding generation failed: {e}. Using random embeddings.")
|
|
embeddings = np.random.rand(3, 384)
|
|
|
|
labels = ["Apple", "Microsoft", "Amazon"]
|
|
|
|
# Need at least n_neighbors + 1 samples for UMAP usually, but with 3 samples it might warn.
|
|
# Let's use PCA or just catch potential UMAP errors if samples are too few.
|
|
try:
|
|
viz = embedding_visualizer.visualize_2d_projection(embeddings, labels, method="umap")
|
|
if viz is None:
|
|
# Fallback to pca if umap fails silently or returns None
|
|
viz = embedding_visualizer.visualize_2d_projection(embeddings, labels, method="pca")
|
|
except Exception as e:
|
|
logger.warning(f"UMAP visualization failed: {e}. Trying PCA.")
|
|
viz = embedding_visualizer.visualize_2d_projection(embeddings, labels, method="pca")
|
|
|
|
assert viz is not None, "Embedding visualization failed"
|
|
logger.info("Embedding Visualization successful")
|
|
|
|
# Step 4: Semantic Network Visualization
|
|
logger.info("Step 4: Semantic Network Visualization")
|
|
semantic_network = {
|
|
"nodes": [
|
|
{"id": "n1", "label": "Node 1", "type": "Entity"},
|
|
{"id": "n2", "label": "Node 2", "type": "Entity"}
|
|
],
|
|
"edges": [
|
|
{"source": "n1", "target": "n2", "label": "related_to"}
|
|
]
|
|
}
|
|
|
|
sem_viz = SemanticNetworkVisualizer()
|
|
viz1 = sem_viz.visualize_network(semantic_network, output="interactive")
|
|
viz2 = sem_viz.visualize_node_types(semantic_network, output="interactive")
|
|
viz3 = sem_viz.visualize_edge_types(semantic_network, output="interactive")
|
|
|
|
assert viz1 is not None, "Semantic Network visualization failed"
|
|
assert viz2 is not None, "Node Types visualization failed"
|
|
assert viz3 is not None, "Edge Types visualization failed"
|
|
logger.info("Semantic Network Visualization successful")
|
|
|
|
# Step 5: Advanced Embedding Visualization
|
|
logger.info("Step 5: Advanced Embedding Visualization")
|
|
text_emb = np.random.rand(50, 128)
|
|
image_emb = np.random.rand(50, 128)
|
|
audio_emb = np.random.rand(50, 128)
|
|
|
|
emb_viz = EmbeddingVisualizer()
|
|
viz1 = emb_viz.visualize_multimodal_comparison(text_emb, image_emb, audio_emb, output="interactive")
|
|
|
|
assert viz1 is not None, "Multimodal comparison failed"
|
|
logger.info("Advanced Embedding Visualization successful")
|
|
|
|
|
|
def run_advanced_notebook():
|
|
logger.info("Running Advanced Notebook steps...")
|
|
|
|
# Step 1: Create Sample Knowledge Graph
|
|
logger.info("Step 1: Create Sample Knowledge Graph")
|
|
builder = GraphBuilder()
|
|
|
|
entities = [
|
|
{"id": "e1", "type": "Person", "name": "Alice", "properties": {"age": 30}},
|
|
{"id": "e2", "type": "Person", "name": "Bob", "properties": {"age": 35}},
|
|
{"id": "e3", "type": "Organization", "name": "Tech Corp", "properties": {"founded": 2010}},
|
|
{"id": "e4", "type": "Location", "name": "San Francisco", "properties": {"country": "USA"}},
|
|
]
|
|
|
|
relationships = [
|
|
{"source": "e1", "target": "e2", "type": "knows", "properties": {"since": 2020}},
|
|
{"source": "e1", "target": "e3", "type": "works_for", "properties": {"role": "Engineer"}},
|
|
{"source": "e3", "target": "e4", "type": "located_in", "properties": {}},
|
|
]
|
|
|
|
knowledge_graph = builder.build([{"entities": entities, "relationships": relationships}])
|
|
|
|
# Step 2: Knowledge Graph Visualization
|
|
logger.info("Step 2: Knowledge Graph Visualization")
|
|
kg_visualizer = KGVisualizer(layout="force", color_scheme="vibrant")
|
|
viz = kg_visualizer.visualize_network(knowledge_graph, output="interactive")
|
|
assert viz is not None, "KG visualization failed"
|
|
logger.info("KG Visualization successful")
|
|
|
|
# Step 3: Generate Embeddings and Visualize
|
|
logger.info("Step 3: Generate Embeddings and Visualize")
|
|
# Use random embeddings to ensure stability
|
|
embeddings = np.random.rand(len(entities), 128)
|
|
labels = [entity.get("type", "Unknown") for entity in entities]
|
|
|
|
embedding_visualizer = EmbeddingVisualizer()
|
|
# t-SNE requires more samples typically, use PCA if it fails
|
|
try:
|
|
viz = embedding_visualizer.visualize_2d_projection(embeddings, labels, method="tsne", output="interactive", file_path=None)
|
|
except Exception as e:
|
|
logger.warning(f"t-SNE failed (likely too few samples): {e}. Using PCA.")
|
|
viz = embedding_visualizer.visualize_2d_projection(embeddings, labels, method="pca", output="interactive", file_path=None)
|
|
|
|
assert viz is not None, "Embedding visualization failed"
|
|
logger.info("Embedding Visualization successful")
|
|
|
|
# Step 5: Graph Analytics Visualization
|
|
logger.info("Step 5: Graph Analytics Visualization")
|
|
# Mocking GraphAnalyzer results
|
|
centrality_scores = {"e1": 0.5, "e2": 0.3, "e3": 0.8, "e4": 0.4}
|
|
# Wrap in expected format
|
|
centrality_data = {"centrality": centrality_scores}
|
|
|
|
community_dict = {"e1": 0, "e2": 0, "e3": 1, "e4": 1}
|
|
# Wrap in expected format
|
|
communities_data = {"node_assignments": community_dict}
|
|
|
|
analytics_visualizer = AnalyticsVisualizer()
|
|
viz1 = analytics_visualizer.visualize_centrality_rankings(centrality_data, title="Node Centrality Scores")
|
|
viz2 = analytics_visualizer.visualize_community_structure(
|
|
knowledge_graph,
|
|
communities_data,
|
|
title="Community Detection"
|
|
)
|
|
|
|
assert viz1 is not None, "Centrality visualization failed"
|
|
assert viz2 is not None, "Communities visualization failed"
|
|
logger.info("Analytics Visualization successful")
|
|
|
|
# Step 6: Temporal Data Visualization
|
|
logger.info("Step 6: Temporal Data Visualization")
|
|
temporal_kg = {
|
|
"entities": entities,
|
|
"relationships": relationships,
|
|
"timestamps": {
|
|
"e1": [2020, 2021, 2022],
|
|
"e2": [2020, 2021],
|
|
"e3": [2010, 2015, 2020, 2022],
|
|
}
|
|
}
|
|
|
|
# Generate events from timestamps
|
|
events = []
|
|
for entity_id, times in temporal_kg["timestamps"].items():
|
|
for t in times:
|
|
events.append({
|
|
"timestamp": t,
|
|
"type": "update",
|
|
"entity": entity_id,
|
|
"label": f"Update {entity_id}"
|
|
})
|
|
temporal_kg["events"] = events
|
|
|
|
entity_history = {
|
|
"e1": [
|
|
{"timestamp": 2020, "properties": {"age": 28}},
|
|
{"timestamp": 2021, "properties": {"age": 29}},
|
|
{"timestamp": 2022, "properties": {"age": 30}},
|
|
]
|
|
}
|
|
|
|
temporal_visualizer = TemporalVisualizer()
|
|
viz1 = temporal_visualizer.visualize_timeline(temporal_kg, output="interactive")
|
|
|
|
timestamps = [str(item["timestamp"]) for item in entity_history["e1"]]
|
|
age_values = [item["properties"]["age"] for item in entity_history["e1"]]
|
|
metrics_history = {"age": age_values}
|
|
viz2 = temporal_visualizer.visualize_metrics_evolution(metrics_history, timestamps, output="interactive")
|
|
|
|
assert viz1 is not None, "Timeline visualization failed"
|
|
assert viz2 is not None, "Metrics evolution visualization failed"
|
|
|
|
# Version Manager part
|
|
try:
|
|
version_manager = TemporalVersionManager()
|
|
v1 = version_manager.create_version(temporal_kg, timestamp="2020-01-01", version_label="v2020")
|
|
temporal_kg_v2 = {
|
|
"entities": temporal_kg.get("entities", []),
|
|
"relationships": temporal_kg.get("relationships", []) + [
|
|
{"source": "e1", "target": "e2", "type": "collaborated_with", "valid_from": "2023-01-01"}
|
|
]
|
|
}
|
|
v2 = version_manager.create_version(temporal_kg_v2, timestamp="2023-01-01", version_label="v2023")
|
|
snapshots = {v1["timestamp"]: v1, v2["timestamp"]: v2}
|
|
|
|
viz3 = temporal_visualizer.visualize_snapshot_comparison(snapshots, output="interactive")
|
|
|
|
version_history = [
|
|
{"version": v1.get("label"), "timestamp": v1.get("timestamp")},
|
|
{"version": v2.get("label"), "timestamp": v2.get("timestamp")}
|
|
]
|
|
viz4 = temporal_visualizer.visualize_version_history(version_history, output="interactive")
|
|
|
|
assert viz3 is not None, "Snapshot comparison failed"
|
|
assert viz4 is not None, "Version history visualization failed"
|
|
except Exception as e:
|
|
logger.warning(f"Temporal Version Manager part failed: {e}")
|
|
|
|
logger.info("Temporal Visualization successful")
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
run_introduction_notebook()
|
|
print("-" * 50)
|
|
run_advanced_notebook()
|
|
print("ALL NOTEBOOK REPRODUCTIONS SUCCESSFUL")
|
|
except Exception as e:
|
|
logger.error(f"Reproduction failed: {e}")
|
|
sys.exit(1)
|