maxwelllink.sockets package

class maxwelllink.sockets.AggregatedSocketHub

Bases: SocketHub

EM-side hub that aggregates multiple molecule requests into one bridge link.

This class keeps the same public methods used by MaxwellLink solvers (register_molecule_return_id, wait_until_bound, all_bound, step_barrier) while mapping many molecule IDs onto a smaller number of bridge connections.

Molecules are assigned to a bridge group through init_payload["aggregate_group"]. All molecules sharing the same group are sent together to one LocalSocketHubBridge.

Parameters:

• host (str or None, optional) – Interface to bind the upstream TCP server to. None, "", or "0.0.0.0" bind all interfaces; bridges then connect back over 127.0.0.1.

• port (int or None, default: 31415) – TCP port for the upstream server.

• timeout (float, default: 60000.0) – Default operation timeout (seconds) used for binding and stepping.

• latency (float, default: 0.01) – Polling interval (seconds) for the bind/step loops.

__init__(host=None, port=31415, timeout=60000.0, latency=0.01)

Initialize the socket hub.

Parameters:

• host (str or None, default: None) – Host address for AF_INET sockets. Ignored when using a UNIX socket.

• port (int or None, default: 31415) – TCP port for AF_INET sockets. Ignored for UNIX sockets.

• unixsocket (str or None, default: None) – Path (or name under /tmp/socketmxl_*) for a UNIX domain socket. When provided, host and port are ignored.

• timeout (float, default: 60000.0) – Socket timeout (seconds) for client operations.

• latency (float, default: 0.01) – Polling sleep (seconds) between hub sweeps; can be very small for local runs.

add_bridge(local_unixsocket)

Create, start, and return one hub-owned local UNIX-socket bridge.

This is the convenience entry point intended for minimal edits when migrating an existing single-layer SocketHub script to the new two-layer transport.

Parameters:

local_unixsocket (str) – Non-empty downstream UNIX-socket address local drivers connect to.

Returns:

A started handle wrapping the new node-local bridge.

Return type:

AggregatedBridge

Raises:

ValueError – If local_unixsocket is empty or already owned by another bridge on this hub.

all_bound(molecule_ids, require_init=True)

Check if all given molecule IDs are bound (and optionally initialized).

Parameters:

• molecule_ids (iterable of int) – Molecule IDs to check.

• require_init (bool, default: True) – Also require that clients completed INIT.

Returns:

True if all are bound (and initialized if requested), else False.

Return type:

bool

graceful_shutdown(reason=None, wait=2.0)

Politely ask all connected drivers to exit and wait briefly for BYE.

Parameters:

• reason (str or None, optional) – Optional reason to log for shutdown.

• wait (float, default: 2.0) – Seconds to wait for clean replies.

init_remote_bridges(molecules, *, molecules_per_bridge, unix_prefix='bridge_', save_file='aggregation.json')

Partition molecules across remote bridge groups and save a manifest.

This helper does not start any bridge threads locally. Instead it assigns molecule.init_payload["aggregate_group"] for each molecule and writes one JSON manifest that bridge-node scripts can consume via run_bridge_node().

Parameters:

• molecules (molecule or iterable of molecules) – Molecules to distribute across remote bridges.

• molecules_per_bridge (int) – Maximum number of molecules assigned to one bridge.

• unix_prefix (str, default: "bridge_") – Prefix used to generate downstream UNIX socket names f"{unix_prefix}{idx}".

• save_file (str, default: "aggregation.json") – Path where the bridge manifest should be written.

Returns:

The generated bridge specifications in order.

Return type:

list[RemoteBridgeSpec]

Raises:

ValueError – If no molecules are supplied or molecules_per_bridge is not a positive integer.

Notes

This method records the generated specs on self.remote_bridges and the full manifest on self.remote_bridge_info as a side effect, but does not start any bridge threads.

register_molecule(molecule_id)

Reserve a slot for a given molecule ID (client may connect later).

Parameters:

molecule_id (int) – Molecule ID to register.

Raises:

ValueError – If the molecule ID is already registered.

Return type:

None

register_molecule_return_id()

Reserve a slot for a molecule and return an auto-assigned ID.

Returns:

The assigned unique molecule ID.

Return type:

int

step_barrier(requests, timeout=None)

Dispatch all requested fields group-by-group and collect grouped replies.

Parameters:

• requests (dict[int, dict]) – Mapping from molecule ID to a request dict with keys: - "efield_au" : array-like (3,) field vector in a.u. - "init" : dict, optional INIT payload for a first bind.

• timeout (float, optional) – Maximum time (seconds) to wait for every group to reply. Defaults to the hub’s timeout setting.

Returns:

Mapping molid -> {"amp": ndarray(3,), "extra": bytes}, matching the SocketHub.step_barrier contract. Returns {} when paused, when a molecule is not yet bound, or on a mid-step disconnect or timeout.

Return type:

dict[int, dict]

Raises:

RuntimeError – If a bridge replies with the wrong set of molecule ids.

Notes

A single pending group is served by a direct blocking receive; multiple groups are awaited on the aggregate selector so whichever bridge becomes readable first is collected next.

stop()

Stop the aggregate hub and clean up bridge groups coherently.

The base SocketHub.stop() assumes one client per molecule, which is not true here. This override shuts down each bridge once and clears all molecule bindings associated with that bridge.

wait_until_bound(init_payloads, require_init=True, timeout=None)

Wait until all requested molecules are served by initialized bridges.

Molecules are grouped through init_payload["aggregate_group"] and each group must be backed by exactly one connected bridge.

Parameters:

• init_payloads (dict[int, dict]) – Mapping from molecule ID to INIT payload to use on bind.

• require_init (bool, default: True) – Also require that each backing bridge completed its AGGINIT handshake.

• timeout (float or None, optional) – Maximum time to wait (seconds). Uses the hub default if None.

Returns:

True if every requested molecule became bound (and, when require_init is set, initialized) within the time limit, else False.

Return type:

bool

class maxwelllink.sockets.LocalSocketHubBridge

Bases: object

Bridge process/thread that fans out aggregate requests to a local SocketHub.

Upstream:

one TCP connection to AggregatedSocketHub

Downstream:

one ordinary SocketHub using either TCP or UNIX sockets, connected to many existing MaxwellLink socket drivers.

Parameters:

• group_id (str) – Non-empty aggregate group identifier this bridge serves.

• upstream_host (str) – Host of the upstream AggregatedSocketHub.

• upstream_port (int) – TCP port of the upstream hub.

• timeout (float, default: 60.0) – Operation timeout (seconds) for both the upstream link and the downstream local hub.

• latency (float, default: 0.01) – Polling interval (seconds) propagated to the downstream local hub.

• local_host (str, default: "127.0.0.1") – Downstream bind host, used only when local_unixsocket is None.

• local_port (int or None, optional) – Downstream TCP port. Ignored when a UNIX socket is used.

• local_unixsocket (str or None, optional) – Downstream UNIX-socket address. When both this and local_port are None, a sanitized name derived from group_id is used.

Raises:

ValueError – If group_id is empty.

__init__(*, group_id, upstream_host, upstream_port, timeout=60.0, latency=0.01, local_host='127.0.0.1', local_port=None, local_unixsocket=None)

Parameters:

• group_id (str)

• upstream_host (str)

• upstream_port (int)

• timeout (float)

• latency (float)

• local_host (str)

• local_port (int | None)

• local_unixsocket (str | None)

property local_endpoint: dict

Return the downstream socket endpoint local drivers should connect to.

Returns:

{"unixsocket": <name>} when a UNIX socket is configured, otherwise {"host": <host>, "port": <port>}.

Return type:

dict

run()

Run the bridge loop until the hub sends STOP or disconnects.

Raises:

RuntimeError – If the upstream hub sends an unrecognized aggregate header.

Return type:

None

Notes

Connects upstream (with retry), sends HELLO, then services AGGINIT, AGGSTEP, and STOP frames in a loop. Upstream transport errors end the loop quietly; the downstream local hub is always stopped on exit.

start()

Start the bridge loop in a daemon thread and return the thread handle.

Returns:

The running daemon thread. If a thread is already alive it is returned unchanged rather than starting a second one.

Return type:

threading.Thread

stop(wait=2.0)

Stop the bridge loop and close the downstream local hub.

Parameters:

wait (float, default: 2.0) – Maximum time (seconds) to wait for the bridge thread to join after signalling it to stop.

Return type:

None

class maxwelllink.sockets.RemoteBridgeSpec

Bases: object

One remote aggregate bridge entry produced by init_remote_bridges.

idx

Zero-based bridge index used by run_bridge_node().

Type:

int

group_id

Aggregate group identifier transmitted upstream.

Type:

str

unixsocket

Downstream UNIX-socket address local drivers should connect to.

Type:

str

n_molecules

Number of molecules assigned to this bridge.

Type:

int

__init__(idx, group_id, unixsocket, n_molecules)

Parameters:

• idx (int)

• group_id (str)

• unixsocket (str)

• n_molecules (int)

Return type:

None

classmethod from_dict(payload)

Build one bridge specification from JSON-decoded manifest data.

Parameters:

payload (Mapping) – Mapping carrying idx, group_id, unixsocket, and n_molecules entries, as written by to_dict().

Returns:

The reconstructed, type-coerced specification.

Return type:

RemoteBridgeSpec

Raises:

KeyError – If a required field is missing from payload.

group_id: str

idx: int

n_molecules: int

to_dict()

Return a JSON-serializable bridge specification mapping.

Returns:

Mapping with the idx, group_id, unixsocket, and n_molecules fields coerced to plain JSON types.

Return type:

dict

unixsocket: str

class maxwelllink.sockets.SocketHub

Bases: object

Socket server coordinating multiple driver connections with an FDTD engine.

This server:

• Accepts and tracks many driver connections.

• Handles initialization handshakes, field dispatch, and result collection.

• Provides a barrier-style step to send fields and receive source amplitudes from all registered molecules.

__init__(host=None, port=31415, unixsocket=None, timeout=60000.0, latency=0.01)

Initialize the socket hub.

Parameters:

• host (str or None, default: None) – Host address for AF_INET sockets. Ignored when using a UNIX socket.

• port (int or None, default: 31415) – TCP port for AF_INET sockets. Ignored for UNIX sockets.

• unixsocket (str or None, default: None) – Path (or name under /tmp/socketmxl_*) for a UNIX domain socket. When provided, host and port are ignored.

• timeout (float, default: 60000.0) – Socket timeout (seconds) for client operations.

• latency (float, default: 0.01) – Polling sleep (seconds) between hub sweeps; can be very small for local runs.

all_bound(molecule_ids, require_init=True)

Check if all given molecule IDs are bound (and optionally initialized).

Parameters:

• molecule_ids (iterable of int) – Molecule IDs to check.

• require_init (bool, default: True) – Also require that clients completed INIT.

Returns:

True if all are bound (and initialized if requested), else False.

Return type:

bool

graceful_shutdown(reason=None, wait=2.0)

Politely ask all connected drivers to exit and wait briefly for BYE.

Parameters:

• reason (str or None, optional) – Optional reason to log for shutdown.

• wait (float, default: 2.0) – Seconds to wait for clean replies.

register_molecule(molecule_id)

Reserve a slot for a given molecule ID (client may connect later).

Parameters:

molecule_id (int) – Molecule ID to register.

Raises:

ValueError – If the molecule ID is already registered.

Return type:

None

register_molecule_return_id()

Reserve a slot for a molecule and return an auto-assigned ID.

Returns:

The assigned unique molecule ID.

Return type:

int

step_barrier(requests, timeout=None)

Barrier step: dispatch fields and collect source amplitudes from all clients.

Coordinates sending fields, waiting for results, and jointly committing the results once every requested molecule is ready. A frozen barrier is reused if a disconnect occurs mid-step.

Parameters:

• requests (dict[int, dict]) – Mapping from molecule ID to request dict with keys: - "efield_au" : array-like (3,) field vector in a.u. - "meta" : dict, optional metadata per send. - "init" : dict, optional INIT payload for first bind.

• timeout (float, optional) – Maximum time (seconds) to wait for the barrier to complete. Defaults to the hub’s timeout setting.

Returns:

Mapping molid -> {"amp": ndarray(3,), "extra": bytes}. Returns {} when paused, on abort, or if the barrier is incomplete.

Return type:

dict[int, dict]

stop()

Stop accepting new connections, request clients to exit, and close sockets.

Also removes the UNIX socket path if one was created.

wait_until_bound(init_payloads, require_init=True, timeout=None)

Block until all requested molecule IDs are bound (and optionally initialized).

Parameters:

• init_payloads (dict[int, dict]) – Mapping from molecule ID to INIT payload to use on bind.

• require_init (bool, default: True) – Also require that clients completed INIT.

• timeout (float or None, optional) – Maximum time to wait (seconds). Uses hub default if None.

Returns:

True if all requested IDs became bound within the time limit, else False.

Return type:

bool

class maxwelllink.sockets.SusceptibilitySocketHub

Bases: object

Process-backed hub for Meep MXLSocketSusceptibility connections.

The public object starts immediately during construction and exposes the endpoint fields consumed by mp.MXLSocketSusceptibility(hub=hub). The actual server runs in a child process because Meep’s long C-level time-step loop can hold Python’s GIL while waiting on the socket; an in-process Python thread would therefore deadlock at the first MXLINIT/MXLREADY handshake. The child server is _SusceptibilitySocketHubServer, which is the SocketHub-derived implementation that reuses MaxwellLink’s optimized driver binding and stepping path.

Under MPI only the master rank launches the child server; every rank then learns the resolved endpoint through mpi_bcast_from_master(), so all ranks agree on the host/port to connect to.

Parameters:

• host (str or None, optional) – Bind host for the server. None uses the server default.

• port (int or None, default: 31415) – Bind port. 0 requests an OS-chosen ephemeral port.

• timeout (float, default: 60000.0) – Socket timeout in seconds passed to the server.

• latency (float, default: 0.05) – Polling interval in seconds passed to the server.

• unixsocket (str or None, optional) – Reserved; must be falsy (TCP only).

host

Resolved bind host of the running server.

Type:

str

port

Resolved bind port of the running server.

Type:

int

address

Alias of host.

Type:

str

rank_stats

Latest per-Meep-rank statistics drained from the child process.

Type:

dict[int, dict]

Raises:

• ValueError – If unixsocket is given.

• RuntimeError – If the child server fails to start.

__init__(host=None, port=31415, timeout=60000.0, latency=0.05, unixsocket=None)

Parameters:

• host (str | None)

• port (int | None)

• timeout (float)

• latency (float)

• unixsocket (str | None)

lorentzian_conversion(frequency, sigma, resolution, *, gamma=0.0, dimensions=1, time_units_fs=0.1, mu0_au=187.0819866, orientation=0)

Convert a Meep Lorentzian susceptibility to SHO driver parameters.

Maps the standard Meep Lorentzian dielectric parameters (resonance frequency, oscillator strength sigma) onto the simple-harmonic oscillator (SHO) molecular driver: a resonance omega in atomic units, the fixed transition dipole mu0_au, and the rescaling_factor that the Meep MXLSocketSusceptibility applies so one socket molecule reproduces the per-cell Lorentzian response.

Parameters:

• frequency (float) – Lorentzian resonance frequency in Meep units (must be > 0).

• sigma (float) – Lorentzian oscillator strength (must be >= 0).

• resolution (float) – Meep grid resolution in pixels per unit length (must be > 0); sets the grid-cell measure (1 / resolution) ** dimensions.

• gamma (float, default: 0.0) – Lorentzian damping rate. Accepted for API symmetry but ignored: the current SHO driver is lossless (a warning is printed if nonzero).

• dimensions (int, default: 1) – Simulation dimensionality, used to form the cell measure (>= 1).

• time_units_fs (float, default: 0.1) – Meep time unit in femtoseconds; sets the Meep<->a.u. conversions.

• mu0_au (float, default: 187.0819866) – SHO transition dipole moment in atomic units (must be nonzero).

• orientation (int, default: 0) – Dipole axis: 0 (x), 1 (y), or 2 (z).

Returns:

Dictionary with two keys:

• "rescaling_factor" : float Value to pass to mp.MXLSocketSusceptibility(rescaling_factor=...).

• "driver_command" : str Ready-to-run shell command that launches one matching mxl_driver --model sho client against this hub.

Return type:

dict

Raises:

ValueError – If any argument is outside its documented valid range.

property rank_stats: dict[int, dict]

Latest per-Meep-rank statistics from the running server.

Returns:

Mapping from rank to its stats row (molecule_count, steps, requests, peer). Empty on non-master ranks.

Return type:

dict[int, dict]

stop()

Stop the hub and tear down the child server process.

Idempotent and safe on non-master ranks. Signals the child via the stop event, joins it, and falls back to terminate() if it does not exit; a final stats drain captures any closing counters.

Return type:

None

maxwelllink.sockets.am_master()

Return True if this process is the MPI master rank (rank 0), otherwise False.

Notes

Attempts to import mpi4py and query COMM_WORLD. If unavailable, returns True by treating the single process as rank 0.

maxwelllink.sockets.get_available_host_port(localhost=True, save_to_file=None)

Ask the OS for an available localhost TCP port.

Parameters:

• localhost (bool, default: True) – If True, bind to the localhost interface (“127.0.0.1”). If False, bind to all interfaces (“0.0.0.0”).

• save_to_file (str or None, default: None) – If provided, save the selected host and port to the given file with filename provided by save_to_file. The first line contains the host, and the second line contains the port.

Returns:

(host, port) pair, e.g., ("127.0.0.1", 34567).

Return type:

tuple

maxwelllink.sockets.mpi_bcast_from_master(value)

Broadcast a Python value from the master rank to all ranks via MPI.

Parameters:

value (any) – The value to broadcast.

Returns:

The broadcast value (unchanged when MPI is unavailable).

Return type:

any

maxwelllink.sockets.run_bridge_node(info='aggregation.json', *, idx=0)

Run one bridge node from a manifest written by init_remote_bridges.

Parameters:

• info (str or path-like, default: "aggregation.json") – JSON manifest written by AggregatedSocketHub.init_remote_bridges().

• idx (int, default: 0) – Zero-based bridge index identifying which bridge entry in info this node should start.

Raises:

IndexError – If no bridge entry in info has the requested idx.

Return type:

None

Notes

The call blocks until the bridge thread exits or a KeyboardInterrupt is received, after which the bridge is stopped on a best-effort basis.

Submodules

• maxwelllink.sockets.aggregated module
  • AggregatedBridge
    • AggregatedBridge.hub
    • AggregatedBridge.group_id
    • AggregatedBridge.__init__()
    • AggregatedBridge.address
    • AggregatedBridge.append()
    • AggregatedBridge.local_endpoint
    • AggregatedBridge.start()
    • AggregatedBridge.stop()
    • AggregatedBridge.unixsocket
    • AggregatedBridge.unixsocket_path
  • AggregatedSocketHub
    • AggregatedSocketHub.__init__()
    • AggregatedSocketHub.add_bridge()
    • AggregatedSocketHub.all_bound()
    • AggregatedSocketHub.graceful_shutdown()
    • AggregatedSocketHub.init_remote_bridges()
    • AggregatedSocketHub.register_molecule()
    • AggregatedSocketHub.register_molecule_return_id()
    • AggregatedSocketHub.remote_bridge_info
    • AggregatedSocketHub.remote_bridges
    • AggregatedSocketHub.step_barrier()
    • AggregatedSocketHub.stop()
    • AggregatedSocketHub.wait_until_bound()
  • LocalSocketHubBridge
    • LocalSocketHubBridge.__init__()
    • LocalSocketHubBridge.local_endpoint
    • LocalSocketHubBridge.run()
    • LocalSocketHubBridge.start()
    • LocalSocketHubBridge.stop()
  • RemoteBridgeSpec
    • RemoteBridgeSpec.idx
    • RemoteBridgeSpec.group_id
    • RemoteBridgeSpec.unixsocket
    • RemoteBridgeSpec.n_molecules
    • RemoteBridgeSpec.__init__()
    • RemoteBridgeSpec.from_dict()
    • RemoteBridgeSpec.group_id
    • RemoteBridgeSpec.idx
    • RemoteBridgeSpec.n_molecules
    • RemoteBridgeSpec.to_dict()
    • RemoteBridgeSpec.unixsocket
  • mxl_bridge_main()
  • run_bridge_node()
• maxwelllink.sockets.sockets module
  • SocketHub
    • SocketHub.__init__()
    • SocketHub.addrmap
    • SocketHub.all_bound()
    • SocketHub.bound
    • SocketHub.clients
    • SocketHub.expected
    • SocketHub.graceful_shutdown()
    • SocketHub.register_molecule()
    • SocketHub.register_molecule_return_id()
    • SocketHub.step_barrier()
    • SocketHub.stop()
    • SocketHub.wait_until_bound()
  • am_master()
  • get_available_host_port()
  • mpi_bcast_from_master()
• maxwelllink.sockets.susceptibility module
  • SusceptibilitySocketHub
    • SusceptibilitySocketHub.host
    • SusceptibilitySocketHub.port
    • SusceptibilitySocketHub.address
    • SusceptibilitySocketHub.rank_stats
    • SusceptibilitySocketHub.__init__()
    • SusceptibilitySocketHub.lorentzian_conversion()
    • SusceptibilitySocketHub.rank_stats
    • SusceptibilitySocketHub.stop()