Source Base

class gpype.backend.sources.base.source.Source

Bases: ONode

Base class for data source nodes in a pipeline.

Provides foundation for all data source nodes that generate or acquire data. Sources have only output ports and serve as pipeline entry points. Handles validation of output ports, channel counts, and frame sizes.

Public as gp.Source: subclass it for a source of your own, and the pipeline builds the same chain around it as around a built-in one. In a realtime pipeline a source drives its own cycles, from a thread or from the call that has something to emit.

TIME_BASE = 'wall_clock'

What this source’s sample positions advance against. Live sources keep the default; a reader that replays stored data overrides it, which is what lets a pipeline reject the two being mixed before anything runs. See Constants.TimeBase.

EXECUTION_MODE = 'realtime'

How a pipeline containing this source is driven. Live sources keep the default; a reader asked for one monolithic block overrides it per instance. Declared here rather than derived from TIME_BASE, because a recording paced to the clock is an ordinary realtime pipeline – the two properties are independent.

DERIVE_FRAME_SIZE = True

Whether this kind of source derives a frame size from its rate when the author names none. False for sources where a frame is not a slice of a live stream: an event source emits one row when something happens, and a batch source’s frame is the recording.

FRAME_RATE_DIVISOR = 62.5

Samples per second a frame should nominally span, as a divisor. 62.5 puts one frame at 16 ms and reproduces the frame sizes the hardware family already uses: 250 Hz -> 4, 500 and 512 -> 8, 1200 -> 16, 2400 -> 32.

static frame_size_for(sampling_rate)

Derived frame size, or the fallback if the rate is unusable.

Never raises, and that is the point. A rate of -1 is the rate’s problem, and the source’s own validation says so precisely; deriving first and raising here would answer a question about the sampling rate with a message about frame sizes, which is exactly the kind of misdirection that costs a user their afternoon.

A rate may also arrive per port, or be inherited rather than declared. Neither is something to derive from.

Parameters:

sampling_rate – Rate in Hz, a per-port list of them, Constants.INHERITED, or None.

Returns:

Frame size, at least 1.

Return type:

int

static default_frame_size(sampling_rate)

Frame size a source should use at this rate, if none is given.

The power of two nearest sampling_rate / 62.5, which holds one frame at roughly 13 to 20 ms across the whole supported range while keeping the size a power of two.

Applied by _default_frame_size() whenever a source is built without one and knows its rate. Event sources and batch sources opt out through DERIVE_FRAME_SIZE.

Two details are deliberate. The clamp: below about 44.2 Hz the unclamped expression returns a fraction (0.5 at 32 Hz), which Source.__init__ rejects with a message about integers rather than about rates, and this repository really does use 1, 10 and 50 Hz. And floor(x + 0.5) rather than round: Python rounds halves to even, so a tie would resolve inconsistently between one rate and the next.

Parameters:

sampling_rate (float) – Rate in Hz. Must be positive.

Returns:

Frame size, at least 1.

Return type:

int

Raises:

ValueError – If sampling_rate is not positive.

attach_timeline(timeline)

Bind this source to the pipeline’s timeline.

Parameters:

timeline – Timeline manager owned by the pipeline.

Return type:

None

attach_provenance(record)

Learn this run’s provenance record.

Called by Pipeline.start() on every node exposing this method (the same hook a sink uses to learn the run’s entitlement), before any node’s setup() runs. setup() merges this source’s own input identity into a copy of it and publishes the result under Constants.Keys.PROVENANCE.

Parameters:

record (dict) – The run’s provenance record, from common._private.provenance.build.

Return type:

None

property is_exhausted

Whether this source has emitted everything it will.

False for a live source, because a clock does not end. A reader overrides it, which is what lets a batch driver know when to stop asking – without it, the driver would have to read a reader’s private state or guess a cycle count.

Returns:

True when nothing further will be emitted.

master_candidacy()

Return this source’s claim to own the master timeline.

Declared at construction, so the pipeline can settle the election before any data flows. That timing is the whole point: a source that only claims when its first block arrives loses to whichever source starts producing soonest, and a synthetic source always beats real hardware that has to open a device first. Measured against a g.HIamp: the amplifier lost every time.

Returns:

(sampling_rate, priority), or None for a source that cannot serve as master. Sparse sources return None: they have no rate of their own and are placed onto the master’s timeline rather than defining one.

static scalar(value)

Return a per-port configuration value as a single number.

Values like channel_count and frame_size are stored per output port, so they are normalised to a list on the way into the configuration. A source rebuilt from that configuration receives the list form back, and constructors that expect a number would otherwise fail or, worse, pass the list on unnoticed.

Parameters:

value – A number, a one-element sequence of numbers, or None.

Returns:

The number, or None if there was none.

serialize()

Serialise the node, writing back what it was given.

A stand-in only lets the configuration validate where the node is never run. Written into a document it would reach the process that does run it as an instruction, so the author’s own value – None, or no key at all – is written instead.

An edge_id left unset is written as no key at all, so a document that assigns no edge is what it was before edge ids existed. See document.omit_unassigned_edge.

Return type:

dict

Returns:

The serialised node.

class Configuration

Bases: Configuration

Configuration class for Source parameters.

class Keys

Bases: Keys

Configuration keys for source-specific settings.

CHANNEL_COUNT = 'channel_count'

Configuration key for number of channels per port

FRAME_SIZE = 'frame_size'

Configuration key for samples per frame

__init__(output_ports=None, channel_count=None, frame_size=None, edge_id=None, **kwargs)

Initialize source with output port configuration.

Parameters:

• output_ports (Optional[list]) – List of output port configurations. Required.

• channel_count (Union[list, int, None]) – Number of channels per port. Can be int (all ports) or list (per port). Defaults to 1. Must be >= 1 or INHERITED.

• frame_size (Union[list, int, None]) – Samples per frame. Can be int (all ports) or list (per port). Defaults to 1. Must be >= 1 or INHERITED.

• edge_id (Optional[str]) – Which edge runs this node, matched against the edge process’s –edge-id. None, the default, is every edge; ignored when the pipeline is not distributed.

• **kwargs – Additional arguments for parent ONode class.

Raises:

• ValueError – If validation fails or input_ports specified, or if edge_id is empty or padded with whitespace.

• TypeError – If edge_id is not a string.

setup(data, port_context_in)

Setup output port contexts with channel and frame information.

Parameters:

• data (dict[str, ndarray]) – Input data arrays (empty for source nodes).

• port_context_in (dict[str, dict]) – Input port contexts (empty for source nodes).

Return type:

dict[str, dict]

Returns:

Dictionary of output port contexts with channel_count and frame_size information for each output port.