Simulate the Next Generation of
Space Networks

from dsns.simulation import Actor
from dsns.events import Event
from dsns.message import MessageCreatedEvent, MessageReceivedEvent

class CustomActor(Actor):
    def __init__(self, params):
        super().__init__()
        # Custom logic

    def initialize(self) -> list[Event]:
        # Generate initial events
        events = [MessageCreatedEvent(...)]
        return events

    def handle_event(self, mobility, event):
        events = []
        if isinstance(event, MessageReceivedEvent):
            # Custom message handling logic
            events.append(MessageCreatedEvent(...))
        return events

import numpy as np
from dsns.constellation import Constellation, Satellite, Satellites, ISLHelper, OrbitalCenter
from dsns.helpers import IDHelper

class CustomConstellation(Constellation):
    def __init__(self, name: str, id_helper: IDHelper, isl_helper: ISLHelper, orbital_center: OrbitalCenter, num_planes: int, sats_per_plane: int):
        self.name = name
        self.isl_helper = isl_helper
        self.orbital_center = orbital_center
        self.num_planes = num_planes
        self.sats_per_plane = sats_per_plane
        self.num_sats = num_planes * sats_per_plane

        # Define initial satellite positions
        self.satellites = Satellites([])
        for plane in range(num_planes):
            for sat in range(sats_per_plane):
                sat_name = f"{name}_{plane}_{sat}"
                sat_id = id_helper.assign_id(sat_name)
                self.satellites.append(Satellite(sat_name, sat_id, self.name, orbital_center=orbital_center))

        self.satellite_positions = np.zeros((self.num_sats, 3))
        self.update_positions(0.0)

    def update_positions(self, time: float):
        for plane in range(self.num_planes):
            for sat in range(self.sats_per_plane):
                sat_index = (plane * self.sats_per_plane) + sat
                pos = ...

                self.satellites[sat_index].position = pos
                self.satellite_positions[sat_index] = pos

from dsns.constellation import Satellites, ISLHelper
from dsns.multiconstellation import ILLHelper
from dsns.helpers import SatID

# Implement a custom Inter-Satellite Link (ISL) helper
# to define connectivity rules between nodes.
class CustomISLHelper(ISLHelper):
    def get_isls(self, satellites: Satellites, positions: np.ndarray) -> list[tuple[SatID, SatID]]:
        # Custom link logic based on distance,
        # elevation, or other constraints
        ...

# Implement a custom Inter-Layer Link (ILL) helper
# to define connectivity rules between constellations or orbital shells.
class CustomILLHelper(ILLHelper):
    def get_ills(self, satellites: Satellites, positions: np.ndarray) -> list[tuple[SatID, SatID]]:
        # Custom link logic based on distance,
        # elevation, or other constraints
        ...

from dsns.simulation import RoutingDataProvider

# Build a custom routing strategy by extending
# the base RoutingDataProvider.

class CustomRoutingProvider(RoutingDataProvider):
    def _get_next_hop(self, source: SatID, destination: SatID) -> Optional[SatID]:
        # Custom next-hop selection logic
        ...

    def get_distance(self, source: SatID, destination: SatID) -> Optional[float]:
        # Latency between a given source and destination
        ...

    def get_neighbors(self, source: SatID, max_distance: Optional[float] = None) -> list[SatID]:
        # Get the reachable neighbors for a given source, up to a maximum time difference
        ...

import os, pickle
from dsns.presets import EarthMoonMarsMultiConstellation
from dsns.simulation import Simulation
from dsns.transmission import LinkTransmissionActor, MessageLocationTracker
from dsns.traffic_sim import MultiPointToPointTrafficActor
from dsns.message_actors import MessageRoutingActor, LookaheadRoutingDataProvider
from dsns.logging import PreprocessedLoggingActor, BandwidthLoggingActor, LTPTransmissionLoggingActor
from dsns.message import LossConfig

constellation = EarthMoonMarsMultiConstellation(moon=True, mars=False, moon_mars_link=False)
tracker = MessageLocationTracker()
routing_data_provider = LookaheadRoutingDataProvider(resolution=15.0, num_steps=600)

actors = [
    LinkTransmissionActor(default_bandwidth=100e6/8, buffer_if_link_busy=True, reroute_on_link_down=True, message_location_tracker=tracker),
    MultiPointToPointTrafficActor([("EarthToMoon", 0, 345, 1e6, 15.0)], update_interval=300, reliable_messages=False, cutoff=28800),
]

actors.append(MessageRoutingActor(routing_data_provider, store_and_forward=True, model_bandwidth=True, loss_config=LossConfig()))

logging_actor_pre = PreprocessedLoggingActor(log_other=False)
logging_actor_bw = BandwidthLoggingActor()
logging_actor_ltp = LTPTransmissionLoggingActor()

sim = Simulation(constellation, actors=actors, logging_actors=[logging_actor_pre, logging_actor_bw, logging_actor_ltp], data_providers=[routing_data_provider], timestep=15.0)
sim.initialize(0)
sim.run(43200, progress=True)

os.makedirs("results", exist_ok=True)
with open("results/pre.pickle", "wb") as f: pickle.dump((pre.direct_messages, pre.broadcast_messages, pre.other_events), f)
with open("results/bw.pickle", "wb") as f: pickle.dump(bw.aggregate(1.0, default_bandwidth=100e6/8), f)
with open("results/ltp.pickle", "wb") as f: pickle.dump(ltp.aggregate(1.0), f)