0rbix / src /graph_analysis.py
nicolasleiva's picture
Initial commit: Add complete Orbix project
3dc2617
Raw History Blame Contribute Delete
1.82 kB
import pandas as pd
import logging
import matplotlib.pyplot as plt
import networkx as nx
from stellargraph import StellarGraph
logger = logging.getLogger("Orbix")
def build_satellite_graph(satellite_data: pd.DataFrame, debris_data: pd.DataFrame) -> StellarGraph:
"""
Construye un grafo heterog茅neo a partir de datos de sat茅lites y desechos.
Se espera que `satellite_data` y `debris_data` contengan las columnas 'satellite_id' y 'debris_id', respectivamente.
"""
if satellite_data.empty or debris_data.empty:
logger.warning("Datos insuficientes para construir el grafo.")
return None
satellite_nodes = satellite_data.set_index("satellite_id")
debris_nodes = debris_data.set_index("debris_id")
# Crear DataFrame de aristas ficticio basado en interacciones
edges_df = pd.DataFrame({
"source": satellite_data['satellite_id'].tolist(),
"target": debris_data['debris_id'].tolist(),
"weight": [1] * min(len(satellite_data), len(debris_data))
})
graph = StellarGraph(
{"Satellite": satellite_nodes, "Debris": debris_nodes},
edges={"interaction": edges_df}
)
return graph
def analyze_graph(graph: StellarGraph):
"""
Analiza y visualiza el grafo satelital.
"""
if graph is None:
logger.error("No se proporcion贸 un grafo para analizar.")
return
logger.info("N煤mero de nodos: %d", graph.number_of_nodes())
logger.info("N煤mero de aristas: %d", graph.number_of_edges())
# Convertir a NetworkX para visualizaci贸n
nx_graph = graph.to_networkx()
pos = nx.spring_layout(nx_graph)
plt.figure(figsize=(8,6))
nx.draw(nx_graph, pos, with_labels=True, node_size=500, font_size=10)
plt.title("Visualizaci贸n del grafo satelital")
plt.show()