Spaces:
Running
Running
| from typing import Dict, List, Optional, Any, Set, Tuple | |
| import numpy as np | |
| from datetime import datetime | |
| from tinytroupe.agent import TinyPerson | |
| from tinytroupe.agent.social_types import ConnectionEdge | |
| class Community: | |
| """Represents a cluster of closely connected personas""" | |
| def __init__(self, community_id: str, members: List[str]): | |
| self.community_id = community_id | |
| self.members = members | |
| self.density: float = 0.0 | |
| self.central_personas: List[str] = [] | |
| self.shared_interests: List[str] = [] | |
| self.avg_engagement_rate: float = 0.0 | |
| class NetworkTopology: | |
| """Represents the entire social network structure""" | |
| def __init__(self): | |
| self.nodes: Dict[str, TinyPerson] = {} # persona_id -> persona | |
| self.edges: List[ConnectionEdge] = [] | |
| self.adjacency_matrix: Optional[np.ndarray] = None | |
| self.influence_matrix: Optional[np.ndarray] = None | |
| self.communities: List[Community] = [] | |
| def add_persona(self, persona: TinyPerson) -> None: | |
| self.nodes[persona.name] = persona | |
| # Update adjacency matrix if necessary | |
| def add_connection(self, source_id: str, target_id: str, **kwargs) -> ConnectionEdge: | |
| connection = ConnectionEdge(connection_id=f"{source_id}_{target_id}", **kwargs) | |
| self.edges.append(connection) | |
| # Also update the persona's internal social_connections | |
| if source_id in self.nodes: | |
| self.nodes[source_id].social_connections[target_id] = connection | |
| return connection | |
| def remove_connection(self, source_id: str, target_id: str) -> None: | |
| self.edges = [e for e in self.edges if not (e.connection_id == f"{source_id}_{target_id}")] | |
| if source_id in self.nodes: | |
| self.nodes[source_id].social_connections.pop(target_id, None) | |
| def get_neighbors(self, persona_id: str, depth: int = 1) -> List[TinyPerson]: | |
| if depth <= 0: return [] | |
| neighbors = [] | |
| if persona_id in self.nodes: | |
| neighbor_ids = list(self.nodes[persona_id].social_connections.keys()) | |
| neighbors = [self.nodes[nid] for nid in neighbor_ids if nid in self.nodes] | |
| if depth > 1: | |
| for nid in neighbor_ids: | |
| neighbors.extend(self.get_neighbors(nid, depth - 1)) | |
| return list(set(neighbors)) | |
| def calculate_centrality_metrics(self) -> Dict[str, float]: | |
| # Placeholder for centrality calculation | |
| return {name: 0.0 for name in self.nodes} | |
| def detect_communities(self) -> List[Community]: | |
| # Placeholder for community detection | |
| return self.communities | |
| def get_metrics(self) -> Dict[str, Any]: | |
| return { | |
| "num_nodes": len(self.nodes), | |
| "num_edges": len(self.edges), | |
| "density": len(self.edges) / (len(self.nodes) * (len(self.nodes) - 1)) if len(self.nodes) > 1 else 0 | |
| } | |