dspk.ops.layout

Channels in a new order on the device, and channel maps.

A channel map says which input channel becomes which output channel, as pairs (like a hardware channel mapper). base is 0, or 1 for 1-based numbers (TDT, MATLAB):

TOML
JSON

[remap]
base = 0
pairs = [
  { from = 4, to = 0, name = "A1" },
  { from = 2, to = 1, name = "A2" },
]

{
  "remap": {
    "base": 0,
    "pairs": [
      {"from": 4, "to": 0, "name": "A1"},
      {"from": 2, "to": 1, "name": "A2"}
    ]
  }
}

Every output channel 0..n gets exactly one input; inputs left out are dropped; the same input twice is refused; names are optional. A map is a dict {"base": 0, "pairs": [...]} (or the file’s {"remap": {...}}), a .toml / .json path, or a list of pairs. In a pipeline spec the same map is a {"op": "remap", "pairs": [...]} (or "path": "map.toml") node.


`as_map`	Any form of channel map as one checked dict.
`gather_channels`	Rows of `data` in a new order, on the device.
`map_order`	The input channel and the name of every output channel of a map.
`read_map`	Reads and checks a channel map file.
`remap`	`data` through a channel map, on the device.
`write_map`	Writes a channel map file, TOML or JSON by the suffix of `path`.

Functions

`as_map`

as_map(map: MapLike, *, base: int = DEFAULT_BASE) -> ChannelMap

Any form of channel map as one checked dict.

Parameters

map : dict, list of dict, str or os.PathLike
A channel map: a {"base", "pairs"} dict (or a file’s {"remap": {...}}), a list of {"from", "to", "name"} pairs, or the path of a .toml / .json map file.
base : int, default DEFAULT_BASE
Channel numbering of a list of pairs: 0, or 1 for 1-based numbers (TDT, MATLAB).

Returns

dict — {"base": ..., "pairs": [...]}.

Raises

ValueError: When the map is invalid: an output without an input, an input used twice, …

`gather_channels`

gather_channels(
    data,
    order: Sequence[int],
    *,
    runtime: str | None = None,
) -> np.ndarray

Rows of data in a new order, on the device: row i is channel order[i].

Channels may be left out or repeated.

Parameters

data : array_like
Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
order : sequence of int
Input channel (0-based) of each output row.
runtime : str, default None
A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

np.ndarray — (len(order), samples), float32.

Raises

ValueError: When order names a channel data does not have.
RuntimeError: When the runtime has no device.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> dspk.ops.layout.gather_channels(x, [3, 0]).shape
(2, 30000)

`map_order`

map_order(
    map: MapLike,
    channels: int | None = None,
) -> tuple[list[int], list[str | None]]

The input channel and the name of every output channel of a map.

Parameters

map : dict, list of dict, str or os.PathLike
A channel map: a {"base", "pairs"} dict (or a file’s {"remap": {...}}), a list of {"from", "to", "name"} pairs, or the path of a .toml / .json map file.
channels : int, default None
Number of input channels, to check the map against.

Returns

list of int — The input channel (0-based) of each output channel.
list of str or None — Each output channel’s name (None when unnamed).

Raises

ValueError: When the map is invalid or reads a channel past channels.

Examples

>>> import dspk
>>> dspk.ops.layout.map_order([{"from": 2, "to": 0}, {"from": 0, "to": 1}])
([2, 0], [None, None])

`read_map`

read_map(path: str | PathLike) -> ChannelMap

Reads and checks a channel map file.

Parameters

path : str or os.PathLike
A .toml or .json file with the pairs under remap.

Returns

dict — The map, checked: {"base": ..., "pairs": [...]}.

Raises

ValueError: For another suffix or an invalid map.
OSError: When the file cannot be read.

`remap`

remap(
    data,
    map: MapLike,
    *,
    runtime: str | None = None,
) -> np.ndarray

data through a channel map, on the device.

Inputs the map leaves out are dropped; map_order gives the outputs’ names.

Parameters

data : array_like
Signal (channels, samples). float32, int8, int16, uint16 and int32 arrays go to the device as they are and are converted there; other dtypes and lists become float32 first.
map : dict, list of dict, str or os.PathLike
A channel map: a {"base", "pairs"} dict (or a file’s {"remap": {...}}), a list of {"from", "to", "name"} pairs, or the path of a .toml / .json map file.
runtime : str, default None
A runtime from runtime.available ("cuda", "hip", "wgpu", "cpu") for this call. By default: the one chosen with runtime.set, else DSPK_RUNTIME, else the first available (GPUs first).

Returns

np.ndarray — (outputs, samples), float32.

Raises

ValueError: When the map is invalid or reads a channel data does not have.
RuntimeError: When the runtime has no device.

Examples

>>> import numpy as np
>>> import dspk
>>> x = np.random.default_rng(0).standard_normal((4, 30_000)).astype(np.float32)
>>> dspk.ops.layout.remap(x, [{"from": 3, "to": 0}, {"from": 1, "to": 1}]).shape
(2, 30000)

`write_map`

write_map(map: MapLike, path: str | PathLike) -> Path

Writes a channel map file, TOML or JSON by the suffix of path.

Parameters

map : dict, list of dict, str or os.PathLike
A channel map: a {"base", "pairs"} dict (or a file’s {"remap": {...}}), a list of {"from", "to", "name"} pairs, or the path of a .toml / .json map file.
path : str or os.PathLike
The .toml or .json file to write.

Returns

Path — The path written.

Raises

ValueError: For another suffix or an invalid map.

Examples

>>> import dspk
>>> pairs = [{"from": 1, "to": 0, "name": "A1"}, {"from": 0, "to": 1, "name": "A2"}]
>>> dspk.ops.layout.write_map(pairs, "map.toml")

Constants

Name	Type	Value
`DEFAULT_BASE`	`int`	`0`
`ChannelMap`		`dict[str, Any]`
`MapLike`		`ChannelMap \| Sequence[dict[str, Any]] \| str \| PathLike`