comnum.Chain#

class comnum.Chain(*blocks: Any, **named: Any)#

An ordered, named sequence of blocks; a chain is itself a block.

Parameters:

*blocks – unnamed blocks, named as the Rust runner names the blocks of a link: by their type (source; a Python block by its class or function, gain), a repeated name followed by the position of its block (awgn_2, awgn_3).
**named – named blocks, in order, after the unnamed ones.

settings#

the campaign settings of the specification the chain comes from (name, seed, grid, metrics, stop, fiber…), the defaults of to_spec() and run().

Type:: dict

base_dir#

the directory of the specification file the chain comes from, against which the relative paths of its blocks (the file of a capture) are resolved, as the comnum command does; None: the working directory.

Type:: pathlib.Path | None

Example

>>> import comnum
>>> from comnum import blocks as B
>>> link = comnum.Chain(tx=B.Source(n_symbols=64), norm=B.Normalize())
>>> link.names, len(link)
(['tx', 'norm'], 2)
>>> comnum.Chain(B.Source(), B.Awgn(snr_db=3.0), B.Awgn(snr_db=6.0)).names
['source', 'awgn_2', 'awgn_3']

classmethod from_spec(spec: str | Path | Mapping[str, Any]) → Chain#

The chain of a specification: the blocks of its link, named by their labels (or types; a repeated name takes the position of its block, as the Rust runner numbers them), and its other fields as settings.

The specification is read by the Rust parser (comnum.spec.load()); the directory of a file is kept as base_dir, so that the relative paths of the blocks are those of the comnum command.

Parameters:

spec (str | os.PathLike | dict) – a mapping, a YAML or JSON text, or a file.

Returns:

the chain.

Return type:

Chain

Raises:

FileNotFoundError – for a string that is neither a file nor a specification.
ValueError – for an invalid specification.

The blocks of a condition as one callable (see Block.bind()).

Parameters:

condition (dict | None) – the values of the grid variables.
fiber (dict | None) – the fiber of the specification (the one of settings by default).
name (str | None) – the name of the campaign, for the blocks that write files.
label (str | None) – unused (the blocks are named by the chain).
profile (bool) – record the time of every block as <name>.time_ms and of the link as link.time_ms.
base_dir (str | os.PathLike | None) – the directory the relative paths of the native blocks are resolved against (base_dir by default).

Returns:

apply(frame, rng).

Return type:

Callable[[Frame, Rng], None]

Example

>>> import comnum
>>> from comnum import blocks as B
>>> link = comnum.Chain(tx=B.Source(n_symbols=8), ch=B.Awgn(snr_db=comnum.var("snr")))
>>> apply = link.bind({"snr": 10.0})
>>> f = comnum.Frame()
>>> apply(f, comnum.Rng(1))
>>> f.samples.size
8

flatten() → list[tuple[str, Block]]#

The blocks of the chain and of its sub-chains, with dotted names.

Returns:: the names and the blocks, in order.
Return type:: list[tuple[str, Block]]

insert_after(after: str, **blocks: Any) → Chain#

A copy of the chain with blocks inserted after a block.

Parameters:

after (str) – the name of the block.
**blocks – the new blocks, by name, in order.

Returns:

the copy.

Return type:

Chain

probe(after: str, metrics: Any = ('snr_db',), *, label: str | None = None, **params: Any) → Chain#

A copy of the chain with a probe block after a block: the metrics measured at that point of every trial of a campaign, reported as <label>.<metric>. The metrics of the specifications need symbols aligned at one sample per symbol; power_dbm and papr_db are measured anywhere.

Parameters:

after (str) – the name of the block.
metrics (list[str]) – the metrics.
label (str | None) – the prefix of the metrics (after by default).
**params – the other parameters of the block (skip_start, skip_end, scale).

Returns:

the copy.

Return type:

Chain

probes(points: Mapping[str, Any], **params: Any) → Chain#

A copy of the chain with a probe after every block named in points.

Parameters:

points (dict) – {block: metrics}.
**params – the parameters of every probe.

Returns:

the copy.

Return type:

Chain

profile(**run: Any) → list[dict[str, Any]]#

Where the time goes: run the chain with profile=True and tabulate the mean time per trial of every block (see comnum.Results.profile()).

Parameters:: **run – the arguments of run() (by default one condition and 8 trials).
Returns:: one row per block, block, time_ms, share, then the whole link.
Return type:: list[dict]

Example

>>> import comnum
>>> from comnum import blocks as B
>>> link = comnum.Chain(tx=B.Source(n_symbols=256), ch=B.Awgn(snr_db=10.0))
>>> [r["block"] for r in link.profile(stop={"max_trials": 2})]
['tx', 'ch', 'link']

replace(**blocks: Any) → Chain#

A copy of the chain with some blocks replaced.

Parameters:: **blocks – the new blocks, by name.
Returns:: the copy.
Return type:: Chain
Raises:: KeyError – for a name the chain does not have.

A Monte Carlo campaign over the grid, the settings updated by the arguments given (as to_spec(); the metrics default to ["ber"]).

Both kinds of chain run on the Rust runner, trial t of condition i drawing from Rng.stream(seed, name, i, t): a native chain on all cores (comnum.run_spec()), a chain with Python blocks on one thread; the stopping rule (a mapping, or a comnum.sim.Stop: a native chain runs it on all cores when a specification can write it, comnum.sim.Stop.to_spec(), on one thread otherwise), the statistics and the outputs are the same, so a native chain gives the same results on both.

Parameters:

grid (dict | None) – the grid.
metrics (list[str] | dict | None) – the metrics.
stop (dict | Stop | None) – the stopping rule.
seed (int | None) – the base seed.
name (str | None) – the campaign name.
threads (int | None) – worker threads (0: all cores); a chain with Python blocks runs on one thread unless this, or the threads of the settings, asks for more (Python blocks are not thread-safe).
progress (bool) – print one line per condition on the standard error.
**extra – other fields of the specification (batch, profile, fiber, output…).

Returns:

one dict per condition, as comnum.run_spec(), with tabular access, plots and persistence.

Return type:

Results

Raises:

ValueError – for threads other than 1 on the one-thread path (Python blocks, or a comnum.sim.Stop that a specification cannot write), which would ignore it.

The declarative specification of the chain (native blocks only): the settings, updated by the arguments given.

Parameters:

grid (dict | None) – the grid, {variable: values}.
name (str | None) – the campaign name (it enters the seeds; default "chain").
metrics (list[str] | dict | None) – the reported metrics (["ber", "snr_db"]) or the full metrics mapping (report, skip_start…).
stop (dict | Stop | None) – the stopping rule (min_errors, min_trials, max_trials, max_time_s, precision), or a comnum.sim.Stop (written by comnum.sim.Stop.to_spec()).
seed (int | None) – the base seed.
**extra – other fields (batch, fiber, confidence…).

Returns:

the specification.

Return type:

dict

Raises:

TypeError – for a chain with Python blocks.
ValueError – for a comnum.sim.Stop that a specification cannot write.

trace(condition: Mapping[str, Any] | None, rng: Rng, *, name: str | None = None) → Trace#

Run the chain once and keep a copy of the frame after every block.

Parameters:

condition (dict | None) – the values of the grid variables.
rng (Rng) – the generator.
name (str | None) – the name of the campaign, for the blocks that write files (as bind()).

Returns:

{name: Frame} in the order of the chain (trace.plot(), trace.save(path)).

Return type:

Trace

property is_native: bool#

Whether every block belongs to the registry (the chain then runs on all cores).

Type:: bool

property names: list[str]#

The names of the blocks, in order.

Type:: list[str]