`magnelio.io`#

An open Project is both the store reader and a scattering result: it satisfies ScatteringResult, so S, db, phase, plot_s and the Touchstone / scikit-rf exports work on it exactly as on the in-RAM result.

Project-store components. open_project lives in the core magnelio namespace; this component holds the reader/writer classes and geometry file I/O. The one-shot save_project/load_project (io/hdf5.py) was removed; the store supersedes it.

class magnelio.io.LoadedGeometry(shapes, background)#

Read-only geometry recovered from a project’s BREP + material list.

Mirrors the iteration surface of GeometryModel (shapes, background, iteration, len) so it drops into the same consumers.

Parameters:: shapes (list)

plot(**kwargs)#: Interactive 3D view — same wrapper as GeometryModel.plot.

plot_cross_section(normal, position, **kwargs)#

2D cross-section — same wrapper as GeometryModel.plot_cross_section.

Parameters:

normal (str)
position (float)

class magnelio.io.Project(path)#

Read-only view over a project directory.

Lazily loads and caches the mesh, geometry and setup metadata on first access. Implements the same scattering-result contract as the in-RAM ScatteringTDResult (see magnelio.analysis.result_interface), so a separate post-processing script and the return value of run() are one and the same reader.

Parameters:: path (str | Path)

S(out_port, in_port, *, mode_out=0, mode_in=0, f_axis=None)#: S-parameter column, derived on read (optionally on a custom f_axis).

a(port, mode=0, *, excited=None, f_ref=None, destagger=True)#: Incident power-wave time series a(t) (see ScatteringTDResult.a).

b(port, mode=0, *, excited=None, f_ref=None, destagger=True)#: Outgoing power-wave time series b(t) (see ScatteringTDResult.b).

checkpoint_state(excited=None)#

Load a run’s resume checkpoint (state_dict), or None.

Returns the nested-dict solver state persisted at the last periodic / final / graceful-abort checkpoint, or None if the run wrote no checkpoint (streaming without resume, or aborted before the first interval). resume() feeds this straight into FITTimeDomainSolver.load_state_dict; the n_completed entry is the step the checkpoint corresponds to.

Parameters:: excited (str | tuple[str, int] | None)
Return type:: dict | None

db(out_port, in_port, *, mode_out=0, mode_in=0, floor_db=-200.0, f_axis=None)#

S-parameter magnitude in dB, derived on read (see S()).

floor_db clamps the result from below so an exact zero stays plottable instead of becoming -inf.

energy_trace(excited=None)#

Stored (step, time, energy) trace of a run [structured array].

Parameters:: excited (str | tuple[str, int] | None)

export_paraview(excited=None, *, glyph_percentile=98.0, bake_state=True)#

(Re-)generate the ready-to-open ParaView session for one run.

The run close already generates this automatically; call this to regenerate with different options, or after the automatic export was skipped (paraview=False, missing pvpython). Writes paraview_open.py (open with paraview --script=…), per-monitor data descriptors under paraview/, and — when pvpython is available and bake_state — the double-clickable paraview.pvsm, all in the run directory.

Parameters:

excited (str or tuple, optional) – Selects the run: a port name, or a (name, mode) pair. May be omitted when the project holds one run.
glyph_percentile (float, default 98.0) – Percentile of the field-vector magnitude used as the glyph clip cap (edge singularities would otherwise dictate the arrow scaling).
bake_state (bool, default True) – Bake paraview.pvsm via pvpython when available.

Returns:

Written artefact paths (script, state or None, monitors); empty when there is nothing to visualise.

Return type:

dict

export_paraview_eigenmodes(*, glyph_percentile=98.0, bake_state=True)#

(Re-)generate the ParaView session for the stored eigenmodes.

The eigenmode counterpart of export_paraview(); writing the eigenmode result already generates this automatically. Eigenmodes have no excitation and no time axis, so they belong to the project rather than to a run: the artefacts land in the project directory itself (paraview_open.py, paraview.pvsm, paraview/eigenmodes.pvd and one .vtr per mode).

Stepping the ParaView time axis steps through the modes — degenerate pairs share an eigenfrequency exactly, so the frequency cannot serve as that axis; it travels as field data instead. Fields are peak-normalised per mode, since an eigenvector carries no absolute amplitude, and the divisors are written alongside so the scaling stays reversible.

Parameters:

glyph_percentile (float, default 98.0) – Percentile of |E| used as the glyph magnitude clip cap (a field peak on a conductor edge would otherwise dictate the arrow scaling).
bake_state (bool, default True) – Bake paraview.pvsm via pvpython when available.

Returns:

Written artefact paths (script, state or None, monitors); empty when the project stores no eigenmodes.

Return type:

dict

Examples

>>> project = mio.AnalysisEigenmode(mesh=mesh, n_modes=4,
...                                 project="cavity").run()
>>> project.export_paraview_eigenmodes()["script"]

monitors_for(excited=None)#

Lazy monitor readers for one run, keyed by monitor name.

Resolves each monitor kind by name into its matching lazy reader — _LoadedFieldMonitor (MonitorFieldTime snapshots) and _LoadedFluxMonitor (MonitorFluxTime scalar series) from the run’s results.h5, plus _LoadedFreqMonitor (MonitorFieldFrequency DFT) from fields_freq.h5 and _LoadedMonitorWallLoss (per-tag dissipated fractions) from wall_loss.h5. The user only knows the name, not the kind or the file. excited selects the run; it may be omitted when the project holds one run.

Parameters:: excited (str | tuple[str, int] | None)
Return type:: dict

phase(out_port, in_port, *, mode_out=0, mode_in=0, deg=True, unwrap=True, f_axis=None)#

Phase of one S-parameter over the frequency axis.

Parameters:

out_port (str) – Observed / excited port names (modes via mode_out / mode_in), as in S().
in_port (str) – Observed / excited port names (modes via mode_out / mode_in), as in S().
deg (bool, default True) – Return degrees; False returns radians.
unwrap (bool, default True) – Unwrap 2π discontinuities along the frequency axis.
f_axis (array-like, optional) – Custom frequency axis; recomputed from the recorded signals like S().
mode_out (int)
mode_in (int)

Returns:

Phase per frequency point.

Return type:

np.ndarray

plot_s(*pairs, db=True, floor_db=-200.0, ax=None)#

Plot S-parameter magnitudes over frequency.

Parameters:

*pairs (tuple) – Channels to plot, each (out_port, in_port) or (out_port, in_port, mode_out, mode_in). Without arguments every recorded channel of every excitation is plotted.
db (bool, default True) – Magnitude in dB (with floor_db) instead of linear.
floor_db (float, default -200.0) – Clip floor for the dB display.
ax (matplotlib.axes.Axes, optional) – Axes to draw into; a new figure is created otherwise.

Returns:

fig (matplotlib.figure.Figure)
ax (matplotlib.axes.Axes)

refresh()#

Re-read the metadata and drop cached run / S-parameter data.

Call this on a live project to pick up newly appended steps, a newly added run, or a status change written by a concurrent solver. The immutable model (mesh, geometry) is kept. Returns self for chaining (project.refresh().s_params).

Return type:: Project

to_skrf(name='magnelio')#

Return the complete S-matrix as a skrf.Network.

Delegates to magnelio.post.SParameterResult.to_skrf(); requires scikit-rf and a complete square matrix.

Parameters:: name (str)

to_touchstone(path)#

Write the complete S-matrix as a Touchstone .sNp file.

Delegates to magnelio.post.SParameterResult.to_touchstone(); requires every channel to have been excited.

Return type:: None

property channels: tuple#: Observed (port_name, mode_idx) pairs, in S-matrix order.

property eigenmodes#: The stored EigenmodeResult, or None if absent (lazy).

property excitations: tuple#: Excited (port_name, mode_idx) pairs — the S-matrix columns stored.

property f_axis: ndarray#: Frequency axis of the stored S-matrix [Hz], ascending.

property geometry#

The reconstructed geometry (LoadedGeometry) or None.

Requires OCC. Returns None if the project carries no geometry (geometry.brep absent).

property grid#: The mesh grid (shortcut for project.mesh.grid).

property mesh#: The reconstructed Mesh (lazy).

property meta: dict#: The parsed project.json contents.

property monitors: dict#

Monitor readers of the project’s sole run.

Convenience for the common single-excitation case (project.monitors["Ez_plane"].plot(t=…)) — resolves time, flux, frequency and wall-loss monitors by name (see monitors_for()). With more than one run the monitor name alone is ambiguous — use monitors_for() with the excited pair.

property params: dict#

Free-form user parameters stored with the project.

Whatever dict the analysis was given as params= — design variables, sweep coordinates, notes. Empty when none were stored.

property reference_signal#: Excitation waveform of the longest run (see ScatteringTDResult).

property runs: dict#

The run index from project.json (name → excited/state/…).

Run states: pending (planned by the analysis, not started — no run directory on disk yet) → running → done / aborted. Live watchers iterating this index should skip pending entries; per-run readers raise on them.

property s_params#: The full S-matrix, derived on read from the stored signals.

property settings: RunSettings#

Settings the stored run was produced with (result contract).

Filled from the stored recipe and run data; entries the store does not (yet) record are None.

property setup: dict#: Analysis setup metadata stored at creation time.

property signals: dict#: {excited_key: {(port, mode): (V, I)}} across all runs.

property status: str#

created → running → done.

done means every planned run is done: the analysis pre-registers its runs as pending, so the status holds at running in the gaps between sequential runs — a live watcher may poll project.refresh().status without racing the writer.

Type:: Project status

class magnelio.io.ProjectStore(path)#

Write-once model store for a project directory.

Persists the static model — geometry (BREP + VTM), mesh, and setup metadata. Streamed time-domain results and per-run resume checkpoints are added by later work packages; ProjectStore holds the directory handle they will attach to.

Create a store with create(); read one back with open_project().

Parameters:: path (str | Path)

classmethod create(path, mesh, *, geometry=None, setup=None, paraview=True, exist_ok=True)#

Create a project directory and write the static model.

Parameters:

path (str or Path) – Project directory (created if absent).
mesh (Mesh) – The simulation mesh (grid, materials, conformal sub-cell data) — written to mesh.h5.
geometry (GeometryModel or list, optional) – Source geometry. When given, an exact geometry.brep and (if paraview) a per-solid geometry.vtm are written, plus a geometry.json carrying per-shape names and materials in compound order.
setup (dict, optional) – JSON-serialisable analysis metadata (dt, frequency plan, port/BC/excitation descriptions). Stored under setup in project.json; consumed by later work packages.
paraview (bool, default True) – Also write the tessellated geometry.vtm for ParaView, and generate the per-run ParaView session at run close.
exist_ok (bool, default True) – Reuse an existing directory (raise if False and it exists).

Return type:

ProjectStore

open_scattering_run(run_name, *, excited, dt, f_axis, channels, port_modes, port_normal_dx, port_line_params, waveform_fn, recorder, port_model='modal', energy_stop_db=None, port_signal_stop_db=None, total_time_steps=None, taper_signals=False, monitors=None, grid=None)#

Open a live streaming sink for one scattering excitation.

Declares the run’s resizable results.h5 streams (HDF5-SWMR), registers the run in project.json as running (replacing a pending pre-registration, see register_planned_runs()), and returns a solver-attachable _ScatteringRunSink. The solver appends the V/I and energy tails during run(); call _ScatteringRunSink.close() when the run finishes to flip its state to done. run_name is sanitised for the filesystem.

energy_stop_db / total_time_steps are recorded in the run index so resume() can default to the run’s original stop criterion — e.g. finish an aborted run to the target it was launched with, without the caller repeating it. monitors + grid declare the field-monitor write-through streams; the sink drains each MonitorFieldTime to disk as the run proceeds.

Parameters:

run_name (str)
excited (tuple[str, int])
dt (float)
port_modes (dict)
port_normal_dx (dict)
port_line_params (dict)
port_model (str)
energy_stop_db (float | None)
port_signal_stop_db (float | None)
total_time_steps (int | None)
taper_signals (bool)

Return type:

_ScatteringRunSink

register_planned_runs(planned)#

Pre-register planned runs as pending in the run index.

planned is an iterable of (run_name, excited) pairs, one per excitation the caller is about to stream. Registering them up front closes the status gap between sequential runs: without it, finishing run k while run k+1 is not yet in the index made _finalize_run() report status = "done" for a project that was still mid-analysis. A pending entry counts as not-done, so the project status stays "running" until the last planned run finishes.

Existing entries are left untouched (fill-in: a second analysis adding excitations must not clobber done/aborted runs). pending entries carry no run directory on disk; they are replaced wholesale when open_scattering_run() starts the run.

Return type:: None

reopen_scattering_run(run_name, *, recorder, waveform_fn, dt, n_keep, step_offset, monitors=None, grid=None, monitor_keep=None, flux_keep=None)#

Reopen a run’s streams to append a resumed tail.

Truncates results.h5 back to n_keep (the checkpoint’s committed step count), marks the run running again, and returns a sink whose reference sampling is offset to global step step_offset so the appended tail is phase-aligned with the pre-resume stream. The caller attaches it to the resuming solver exactly like a fresh sink. monitor_keep truncates each field monitor’s stream to its checkpointed snapshot count so the resumed run appends onward consistently.

Parameters:

run_name (str)
dt (float)
n_keep (int)
step_offset (int)
monitor_keep (dict | None)
flux_keep (dict | None)

Return type:

_ScatteringRunSink

write_eigenmodes(result)#

Persist an EigenmodeResult to eigenmodes.h5.

Eigenmode analysis has no time-marching state, so it produces a one-shot result rather than a streamable/resumable run.

Return type:: None

magnelio.io.read_brep(path)#

Read a BREP compound file back into an ordered list of shapes.

Parameters:: path (str or Path) – A .brep file written by write_brep().
Returns:: The compound’s direct children, in write order.
Return type:: list of TopoDS_Shape

magnelio.io.write_brep(shapes, path)#

Write an ordered list of geometry shapes to a BREP compound file.

Parameters:

shapes (list) – Geometry shape objects (each exposing _occ_shape()) or raw TopoDS_Shape instances. The write order is preserved and recovered verbatim by read_brep().
path (str or Path) – Output .brep file.

Return type:

None

magnelio.io#

`magnelio.io`#