Spaces:
Paused
Paused
File size: 2,920 Bytes
c1d0c23 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | 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
}
|