ControlPort Tools

Tools for advanced runtime control via ControlPort (Thrift protocol).

`connect_controlport`

Connect to a GNU Radio ControlPort/Thrift endpoint.

ControlPort provides richer functionality than XML-RPC:

Native type support (complex numbers, vectors)
Performance counters (throughput, timing, buffer utilization)
Knob metadata (units, min/max, descriptions)
PMT message injection
Regex-based knob queries

Parameters

Name	Type	Default	Description
`host`	`str`	`"127.0.0.1"`	Hostname or IP address
`port`	`int`	`9090`	ControlPort Thrift port

Returns

Type: ThriftConnectionInfoModel

Connection details.

Example

connect_controlport(host="127.0.0.1", port=9090)
# Returns: ThriftConnectionInfoModel(
#   host="127.0.0.1",
#   port=9090,
#   container_name=None
# )

`connect_to_container_controlport`

Connect to a flowgraph’s ControlPort by container name.

Parameters

Name	Type	Default	Description
`name`	`str`	-	Container name

Returns

Type: ThriftConnectionInfoModel

Connection details.

Example

connect_to_container_controlport(name="fm-profiled")

`disconnect_controlport`

Disconnect from the current ControlPort endpoint.

Returns

Type: bool

True if successful.

Example

disconnect_controlport()
# Returns: True

`get_knobs`

Get ControlPort knobs, optionally filtered by regex pattern.

Knobs are named using the pattern: block_alias::varname

Parameters

Name	Type	Default	Description
`pattern`	`str`	`""`	Regex pattern for filtering (empty = all)

Returns

Type: list[KnobModel]

List of knobs with name, value, and type.

Example

# Get all knobs
knobs = get_knobs(pattern="")

# Filter by pattern
knobs = get_knobs(pattern=".*frequency.*")
# Returns: [KnobModel(name="sig_source_0::frequency", value=1000.0, ...)]

# Get knobs for a specific block
knobs = get_knobs(pattern="osmosdr_source_0::.*")

`set_knobs`

Set multiple ControlPort knobs atomically.

Parameters

Name	Type	Default	Description
`knobs`	`dict[str, Any]`	-	Dict mapping knob names to new values

Returns

Type: bool

True if successful.

Example

set_knobs({
    "sig_source_0::frequency": 1500.0,
    "sig_source_0::amplitude": 0.8
})
# Returns: True

`get_knob_properties`

Get metadata (units, min/max, description) for specified knobs.

Parameters

Name	Type	Default	Description
`names`	`list[str]`	-	Knob names to query (empty = all)

Returns

Type: list[KnobPropertiesModel]

Knob metadata.

Example

props = get_knob_properties(names=["sig_source_0::frequency"])
# Returns: [KnobPropertiesModel(
#   name="sig_source_0::frequency",
#   units="Hz",
#   min=0.0,
#   max=1e9,
#   description="Output signal frequency"
# )]

`get_performance_counters`

Get performance metrics for blocks via ControlPort.

Requires the flowgraph to be launched with enable_controlport=True and enable_perf_counters=True (default).

Parameters

Name	Type	Default	Description
`block`	`str \| None`	`None`	Block alias to filter (None = all blocks)

Returns

Type: list[PerfCounterModel]

Performance metrics.

Example

counters = get_performance_counters()
# Returns: [PerfCounterModel(
#   block="low_pass_filter_0",
#   nproduced=1048576,
#   nconsumed=10485760,
#   avg_work_time_us=45.3,
#   avg_throughput=23.2e6,
#   pc_input_buffers_full=0.85,
#   pc_output_buffers_full=0.12
# ), ...]

# Filter by block
counters = get_performance_counters(block="low_pass_filter_0")

Metrics

Metric	Description
`nproduced`	Items produced by block
`nconsumed`	Items consumed by block
`avg_work_time_us`	Average time in `work()` function
`avg_throughput`	Items per second
`pc_input_buffers_full`	Input buffer utilization (0-1)
`pc_output_buffers_full`	Output buffer utilization (0-1)

`post_message`

Send a PMT message to a block’s message port via ControlPort.

Parameters

Name	Type	Default	Description
`block`	`str`	-	Block alias (e.g., “msg_sink0”)
`port`	`str`	-	Message port name (e.g., “in”)
`message`	`Any`	-	Message to send (converted to PMT if needed)

Returns

Type: bool

True if successful.

Example

# Send a string message
post_message(block="pdu_sink_0", port="pdus", message="hello")

# Send a dict (converted to PMT dict)
post_message(block="command_handler_0", port="command", message={"freq": 1e6})