Generics

class Processor

Base class for processing modules.

This class provides a basic structure for processing modules, including a forward method that defines how input data should be processed. Derived classes should implement the forward method to define their specific processing logic.

Signal Model

The generic model for a processor is :

\[\mathbf{Y} = \mathbf{f}(\mathbf{X};\boldsymbol \theta)\]

• \(\mathbf{X}\) corresponds to the input data,

• \(\mathbf{Y}\) corresponds to the output data,

• \(\boldsymbol \theta\) corresponds to the processor parameters.

• \(\mathbf{f}(.)\) is a multidimensional nonlinear function.

When a processor is called with input data, it automatically computes the output data by calling its forward method.

Note

A processor is not necessarly fully deterministic. Some processor can also contain a stochastic part.

forward(x: ndarray, /) → ndarray

Process the input data

set_debug(debug: bool | None = None) → None

Change the debugging mode

prepare(X: ndarray, /) → None

Prepare the object before calling the forward method

class ChainGraph

Renderer-independent model returned by Sequential.graph().

class Sequential(module_list: ~typing.List[~comnumpy.core.generics.Processor], debug: bool = False, name: str = 'sequential', callbacks: ~typing.Dict[str | int, ~typing.Callable[[~typing.Any], None]] | None = <factory>, *, observations: ~typing.List[str] | None = None, wiring: ~typing.Dict[str, str] | None = None)

A sequential container for processing modules.

This class allows to create complex chain by stacking \(L\) different processor modules. These processors are executed in the order they are added.

Signal Model

• Initialisation:

\[\mathbf{Y}_0 = \mathbf{X}\]

• Iterations. For \(l=0, \cdots, L-1\), perform the following assignement

  \[\mathbf{Y}_{l+1} = \mathbf{f}_l(\mathbf{Y}_{l};\boldsymbol \theta_l)\]

  where \(\mathbf{f}_l()\) corresponds to the multidimensional function of the \(l^{th}\) processor.

The forward method returns the last output \(\mathbf{Y}_{L}\)

Callbacks

You can optionally provide a dictionary of callbacks via the callbacks attribute. These functions will be called after each processor executes, receiving that processor’s output as input. Keys of the dictionary correspond to processor names (processor.name) in the chain.

module_list

Ordered list of processing modules to be executed sequentially.

Type:

list

debug

Enables debug mode if True (default is False).

Type:

bool, optional

name

Name of the sequential processor (default is ‘sequential’).

Type:

str, optional

callbacks

Dictionary of callback functions called after each processor. Keys are processor names (str) or indices (int), values are callables accepting the processor output.

Type:

dict, optional

observations

Block ids (see block_ids()) whose output should be retained during forward. An observation is simply the output of a named mechanism, kept for experimental analysis: it performs no computation and does not alter the signal. Observations are chain metadata, so the module list describes the communication system and nothing else, and the cost is one dictionary store per observed block (a reference is kept, no copy – blocks allocate fresh outputs, so the reference stays valid). Retrieve with observation().

Type:

list of str, optional, keyword-only

elapsed_

Wall time of the last pass, in seconds (data-dependent, hence the trailing underscore, decision D23). Every call records it, so “how long does this chain take” needs no stopwatch around the call site; profile_execution_time() breaks the same number down block by block.

Type:

float

wiring

Extra data edges, as {"block_id.param": "source_block_id"}. Before a target block runs, the chain assigns it the signal produced by the source block in the same pass – this is how a data-aided estimator receives a reference generated upstream ({"phase_comp.reference": "tx"}). The source is observed automatically and must come earlier in the chain, so the value is never stale. Like observations, the edge is chain metadata: blocks stay declarative and never hold a reference to another block.

Type:

dict, optional, keyword-only

Examples

>>> from comnumpy.core.generators import SymbolGenerator
>>> from comnumpy.core.channels import AWGN
>>> chain = Sequential([SymbolGenerator(4), AWGN(snr_dB=10)],
...                    observations=["generator"])
>>> y = chain.seed(1)(5)
>>> print(chain.observation("generator"))
[0 0 3 1 3]
>>> chain.elapsed_ > 0
True

block_ids() → List[str]

Return the addressable identifier of each block (decision D31/D34).

The identifier is the block’s name field when it has one (lower snake case), otherwise the class name in snake case; a numeric suffix disambiguates duplicates (awgn, awgn_2).

get_module_by_id(block_id: str) → Processor

Retrieve a module by its block_ids() identifier.

seed(seed: int) → Sequential

Seed every stochastic block deterministically (decision D6).

A numpy.random.SeedSequence spawned from seed gives each block that declares a seed field its own independent child seed, then re-runs its parametric initialization so the RNG is rebuilt. A nested Sequential – in the module list or held by a wrapper block – is seeded recursively from its own child seed. Same chain + same seed = same signal, whatever the number, order or nesting of stochastic blocks.

Examples

>>> from comnumpy.core.generators import SymbolGenerator
>>> chain = Sequential([SymbolGenerator(4)])
>>> _ = chain.seed(42)
>>> x1 = chain(5)
>>> _ = chain.seed(42)
>>> print(all(chain(5) == x1))
True

set_params(**params: Any) → Sequential

Reconfigure blocks after construction (decision D34).

Parameters are addressed by block and field, written either "<block_id>.<field>" or <block_id>__<field> – the second is the same address as a plain keyword argument, so it needs no **{...} around it. <block_id> comes from block_ids(). After all assignments, the parametric precomputation of each touched block (__post_init__) is re-run, so the block state stays consistent.

Examples

>>> from comnumpy.core.channels import AWGN
>>> chain = Sequential([AWGN(sigma2=0.1)])
>>> _ = chain.set_params(awgn__sigma2=0.01)
>>> chain[0].sigma2
0.01

The dotted form is the one to use when the address is computed:

>>> _ = chain.set_params(**{"awgn.sigma2": 0.02})
>>> chain[0].sigma2
0.02

summary(X: Any, print_out: bool = True) → List[Tuple[int, str, str, Tuple[int, ...] | None, str, float]]

Run the chain on X and tabulate each block’s output shape, dtype and execution time (decision D33b). Returns the rows.

The pass is the ordinary one: observations are retained and the data edges declared in wiring are fed, so the chain summarized is the chain that runs.

graph() → ChainGraph

Structural view of the chain: nodes, signal edges, data edges.

The renderer-independent model behind to_mermaid(). Signal edges are the implicit linear path; data edges are the ones declared in wiring (decision D42c), which a picture of the chain must show or it is not telling the truth.

Returns:

dict – nodes (list of {"id", "type"}), signal_edges (list of (source, target)), data_edges (list of (source, target, param)) and observations (observed ids).

Examples

>>> from comnumpy.core.generators import SymbolGenerator
>>> from comnumpy.core.channels import AWGN
>>> g = Sequential([SymbolGenerator(4), AWGN(snr_dB=10)]).graph()
>>> print(g["signal_edges"])
[('generator', 'awgn')]

to_mermaid() → str

Mermaid flowchart of the chain (decision D33c), renderable by sphinxcontrib-mermaid, GitHub, or any mermaid viewer.

Signal flow is drawn with solid arrows, declared data edges (wiring, D42c) with dashed arrows labelled by the parameter they feed, and observed blocks are outlined – so the picture shows every edge the chain actually has.

Examples

>>> from comnumpy.core.generators import SymbolGenerator
>>> from comnumpy.core.channels import AWGN
>>> chain = Sequential([SymbolGenerator(4), AWGN(snr_dB=10)],
...                    observations=["generator"])
>>> print(chain.to_mermaid())
flowchart LR
    generator["SymbolGenerator"]
    awgn["AWGN"]
    generator --> awgn
    classDef observed stroke-dasharray: 4 2
    class generator observed

set_debug(debug: bool | None = None) → None

Change the debugging mode

profile_execution_time(X: ndarray) → Dict[str, float]

Time each block over one pass, keyed by block id.

The pass is the ordinary one: observations are retained and the data edges declared in wiring are fed, so a chain that profiles is the chain that runs. The keys are the ids of block_ids(), so two blocks sharing a name still get one entry each.

Parameters:

X (np.ndarray) – Input signal, or the requested size when the chain starts with a source block.

Returns:

dict – Seconds spent in each block, in chain order.

Examples

>>> from comnumpy.core.generators import SymbolGenerator
>>> from comnumpy.core.channels import AWGN
>>> chain = Sequential([SymbolGenerator(4), AWGN(snr_dB=10)])
>>> print(list(chain.profile_execution_time(8)))
['generator', 'awgn']

forward(X: ndarray) → ndarray

Process the input data through all modules in the sequence.

observation(block_id: str) → Any

Return the output retained at a declared observation during the last run. The signal comes back exactly as the block produced it – same shape, same dtype, no computation.

get_module_by_index(index: int) → Processor

Retrieve a module from the module list by its index.

set_module_by_index(module: Processor, index: int)

Set module by index

get_module_by_name(module_name: str) → Processor

Retrieve a module from the module list by its name.