Accessing variables

Accessing variables#

In the SDK, Variables, Aliases, RPV are presented to the client side through an interface called a watchable, e.g. something you can watch. Some watchables are available only when the server has loaded the SFD matching the device firmware (alias, var), others are available as soon as a device is connected (RPV)

Watchable types#

Type

Description

SFD required

Variable

A variable maps to a static or global variable declared in the embedded device firmware. The variable address, type, size and endianness is defined in the loaded SFD

Yes

Runtime Published Values (RPV)

Readable and writable elements identified by a numerical ID (16 bits) and declared by the device during the handshake phase with the server.

No

Alias

Abstract writable/readable entity that maps to either a variable or a RPV. Used to keep a consistent firmware interface with existing scripts using this SDK

Yes

Basics#

The first step to access a watchable, is to first tell the server that we want to subscribe to update event on that watchable. To do so, we use the watch method and specify the path of the watchable. The path depends on the firmware and must generally be known in advance. The path is dependent on the firmware and is typically known beforehand. It’s also possible to query the server for a list of available watchables, a feature utilized by the GUI.

For scripts based on the SDK, it is generally assumed that the elements to be accessed are predetermined and won’t necessitate user input for selection.

ScrutinyClient.watch(path)[source]#

Starts watching a watchable element identified by its display path (tree-like path)

Parameters:

path (str) – The path of the element to watch

Raises:

  • OperationFailure – If the watch request fails to complete

  • TypeError – Given parameter not of the expected type

Returns:

A handle that can read/write the watched element.

Return type:

WatchableHandle

Once an element is being watched, the server starts polling for the value of that element. Each time the value is updated, the server broadcast a value update to all subscribers, in this case, our client. Concurrently, a background thread is actively listening listen for these updates and accordingly modifies the value that the WatchableHandle refers to

class scrutiny.sdk.watchable_handle.WatchableHandle[source]#

A handle to a server watchable element (Variable / Alias / RuntimePublishedValue) that gets updated by the client thread.

__init__(client, display_path)[source]#
Parameters:
Return type:

None

has_enum()[source]#

Tells if the watchable has an enum associated with it

Return type:

bool

get_enum()[source]#

Returns the enum associated with this watchable

Raises:

BadEnumError – If the watchable has no enum assigned

Return type:

EmbeddedEnum

parse_enum_val(val)[source]#

Converts an enum value name (string) to the underlying integer value

Parameters:

val (str) – The enumerator name to convert

Raises:

  • BadEnumError – If the watchable has no enum assigned or the given value is not a valid enumerator

  • TypeError – Given parameter not of the expected type

Return type:

int

property display_path: str#

Contains the watchable full tree path

property name: str#

Contains the watchable name, e.g. the basename in the display_path

property type: WatchableType#

The watchable type. Variable, Alias or RuntimePublishedValue

property datatype: EmbeddedDataType#

The data type of the device element pointed by this watchable. (sint16, float32, etc.)

property value: int | float | bool#

The last value received for this watchable. Can be written

property value_bool: bool#

The value casted as bool

property value_int: int#

The value casted as int

property value_float: float#

The value casted as float

property value_enum: str#

The value converted to its first enum name (alphabetical order). Returns a string. Can be written with a string

property last_update_timestamp: datetime | None#

Time of the last value update. None if not updated at least once. Not reliable for change detection

property last_write_timestamp: datetime | None#

Time of the last successful write operation. None if never written

property update_counter: int#

Number of value update gotten since the creation of the handle. Can be safely used for change detection

class scrutiny.sdk.WatchableType[source]#

(Enum) Type of watchable available on the server

Variable = 1#

A variable found in the device firmware debug symbols

RuntimePublishedValue = 2#

A readable/writable element identified by a 16bits ID. Explicitly defined in the device firmware source code

Alias = 3#

A symbolic link watchable that can refers to a Variable or a RuntimePublishedValue

After getting a handle to the watchable, the value property and its derivative ( value_int, value_float, value_bool, value_enum) undergo automatic updates. These values are invalid until their initial update, meaning that after the call to watch, there is a period of time where accessing the value property will raise a InvalidValueError.

To await a single value update from the watchable, one can utilize the WatchableHandle.wait_update method. Alternatively, to wait for updates from all watched variables at once, the ScrutinyClient.wait_new_value_for_all method can be invoked.

import time
from scrutiny.sdk.client import ScrutinyClient

client = ScrutinyClient()
with client.connect('localhost', 8765):
    w1 = client.watch('/alias/my_alias1')
    w2 = client.watch('/rpv/x1234')
    w3 = client.watch('/var/main.cpp/some_func/some_var')
    client.wait_new_value_for_all() # Make sure all watchables have their first value available

    while w1.value_bool:            # Value updated by a background thread
        print(f"w2 = {w2.value}")   # Value updated by a background thread
        time.sleep(0.1)

    w3.value = 123  # Blocking write. This statement blocks until the device has confirmed that the variable is correctly written (or raise on failure).

Note

Reading and writing a watchable may raise an exception.

  • ReadingWhen value is unavailable. This will happen if

    1. The watchable has never been updated (small window of time after subscription)

    2. The server disconnects

    3. The device is disconnects

  • WritingWhen the value cannot be written. This will happen if

    1. The server disconnects

    2. The device is disconnects

    3. Writing is actively denied by the device. (Communication error or protected memory region)

    4. Timeout: The write confirmation takes more time than the client write_timeout

As demonstrated in the preceding example, device access is executed in a fully synchronized manner. Consequently, a script utilizing the Scrutiny Python SDK can be perceived as a thread operating on the embedded device with a slower memory access time.

Detecting a value change#

When developing a script that uses the SDK, it is common to have some back and forth between the device and the script. A good example would be the case of a test sequence, one could write a sequence that looks like this.

  1. Write a GPIO

  2. Wait for another GPIO to change its value

  3. Start an EEPROM clear sequence

  4. Wait for the sequence to finish

Each time the value is updated by the server, the WatchableHandle.update_counter gets incremented. Looking for this value is helpful to detect a change. Two methods can help the user to wait for remote events. WatchableHandle.wait_update and WatchableHandle.wait_value

The server periodically broadcasts value updates, typically at a rapid pace. The delay in updates is primarily dependent on the saturation level of the communication link with the device. Factors such as the number of watchable subscriptions and the available bandwidth will influence the update rate. The server polls the device for each watchable in a round-robin scheme. When value updates are available, they are aggregated and flushed to all clients. In most common scenarios, a value update can be expected within a few hundred milliseconds.

WatchableHandle.wait_update(timeout, previous_counter=None, sleep_interval=0.02)[source]#

Wait for the value to be updated by the server

Parameters:

  • timeout (float) – Amount of time to wait for a value update

  • previous_counter (int | None) – Optional update counter to use for change detection. Can be set to update_counter+N to wait for N updates

  • sleep_interval (float) – Value passed to time.sleep while waiting

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • InvalidValueError – If the watchable becomes invalid while waiting

  • TimeoutException – If no value update happens within the given timeout

Return type:

None

WatchableHandle.wait_value(value, timeout, sleep_interval=0.02)[source]#

Wait for the watchable to reach a given value. Raises an exception if it does not happen within a timeout value

Parameters:

  • value (int | float | bool | str) – The value that this watchable must have to exit the wait state

  • timeout (float) – Maximum amount of time to wait for the given value

  • sleep_interval (float) – Value passed to time.sleep while waiting

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • InvalidValueError – If the watchable becomes invalid while waiting

  • TimeoutException – If the watchable value never changes for the given value within the given timeout

Return type:

None

ScrutinyClient.wait_new_value_for_all(timeout=5)[source]#

Wait for all watched elements to be updated at least once after the call to this method

Parameters:

timeout (float) – Amount of time to wait for the update

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • TimeoutException – If not all watched elements gets updated in time

Return type:

None

Batch writing#

Writing multiples values in a row is inefficient due to the latency associated with device access. To optimize speed, one can consolidate multiple write operations into a single batched request using the ScrutinyClient.batch_write method.

In a batch write operation, multiple write requests are queued and dispatched to the server in a single API call. The server then executes all write operations in the correct order and confirms the completion of the entire batch.

It is permissible to perform multiple writes to the same watchable within the same batch. The server ensures that each write operation is completed and acknowledged by the device before initiating the subsequent operation.

ScrutinyClient.batch_write(timeout=None)[source]#

Starts a batch write. Write operations will be enqueued and committed together. Every write is guaranteed to be executed in the right order

Parameters:

timeout (float | None) – Amount of time to wait for the completion of the batch once committed. If None the default write timeout will be used.

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • OperationFailure – Failed to complete the batch write

Return type:

BatchWriteContext

Example#

w1 = client.watch('/alias/my_alias1')
w2 = client.watch('/rpv/x1234')
w3 = client.watch('/var/main.cpp/some_func/some_var')
try:
    with client.batch_write(timeout=3):
        w1.value = 1.234
        w2.value = 0x11223344
        w2.value = 0x55667788
        w3.value = 2.345
        w1.value = 3.456
        # Exiting the with block will block until the batch completion or failure (with an exception)

    print("Batch writing successfully completed")
except ScrutinySDKException as e:
    print(f"Failed to complete a batch write. {e}")

Accessing the raw memory#

In certain scenarios, it may be advantageous to directly access the device memory, bypassing the server’s interpretive layer that transforms the data into a meaningful value. Such scenarios could include:

  • Dumping a data buffer

  • Uploading a firmware

  • Pushing a ROM image

  • etc.

For those cases, one can use ScrutinyClient.read_memory and ScrutinyClient.write_memory to access the memory.

ScrutinyClient.read_memory(address, size, timeout=None)[source]#

Read the device memory synchronously.

Parameters:

  • address (int) – The start address of the region to read

  • size (int) – The size of the region to read, in bytes.

  • timeout (float | None) – Maximum amount of time to wait to get the data back. If None, the default timeout value will be used

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • OperationFailure – Failed to complete the reading

  • TimeoutException – If the read operation does not complete within the given timeout value

Return type:

bytes

ScrutinyClient.write_memory(address, data, timeout=None)[source]#

Write the device memory synchronously. This method will exit once the write is completed otherwise will throw an exception in case of failure

Parameters:

  • address (int) – The start address of the region to read

  • data (bytes) – The data to write

  • timeout (float | None) – Maximum amount of time to wait to get the write completion confirmation. If None, the default write timeout value will be used

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • OperationFailure – Failed to complete the reading

  • TimeoutException – If the read operation does not complete within the given timeout value

Return type:

None

Available watchables#

It is possible to query the server for the current number of available watchable items and also download their definition

This feature is typically not required for automation scripts; however, it can be ncessary for presenting users with selectable watchable items. It is currently utilized by the Scrutiny GUI to populate the Variable List widget

ScrutinyClient.get_watchable_count()[source]#

Request the server with the number of available watchable items, organized per type

Raises:

  • ValueError – Bad parameter value

  • TypeError – Given parameter not of the expected type

  • OperationFailure – If the command completion fails

Returns:

A dictionnary containing the number of watchables, classified by type

Return type:

Dict[WatchableType, int]

ScrutinyClient.download_watchable_list(types=None, max_per_response=500, name_patterns=[])[source]#

Request the server for the list of watchable items available in its datastore.

Parameters:

  • types (List[WatchableType] | None) – List of types to download. All of them if None

  • max_per_response (int) – Maximum number of watchable per datagram sent by the server.

  • name_patterns (List[str]) – List of name filters in the form of a glob string. Any watchable with a path that matches at least one name filter will be returned. All watchables are returned if None

Raises:

  • ValueError – Bad parameter value

  • TypeError – Given parameter not of the expected type

  • ConnectionError – If the connection to the server is broken

Returns:

A handle to the request object that can be used for synchronization (wait_for_completion) or cancel the request (cancel)

Return type:

WatchableListDownloadRequest

Following is the object returned download_watchable_list :

class scrutiny.sdk.client.WatchableListDownloadRequest[source]#

Represents a pending watchable download request. It can be used to wait for completion or cancel the request.

cancel()[source]#
Informs the client that this request can be canceled.

Subsequent server response will be ignored and this request will be marked as completed, but failed.

Raises:

TimeoutException – If the client fails to cancel the request. Should never happen

Return type:

None

get()[source]#

Returns the definition of all the watchables obtained through this request, classified by type.

Raises:

InvalidValueError – If the data is not available yet (if the request did not completed successfully)

Returns:

A dictionary of dictionary containing the difinition of each watchable entry that matched the filters. foo[type][display_path] = <definition>

Return type:

Dict[WatchableType, Dict[str, WatchableConfiguration]]

wait_for_completion(timeout=None)#

Wait for the request to complete

Params timeout:

Maximum wait time in seconds. Waits forever if None

Raises:

  • TimeoutException – If the request does not complete in less than the specified timeout value

  • OperationFailure – If an error happened that prevented the request to successfully complete

Parameters:

timeout (float | None) –

Return type:

None

property completed: bool#

Indicates whether the request has completed or not

property completion_datetime: datetime | None#

The time at which the request has been completed. None if not completed yet

property failure_reason: str#

When the request failed, this property contains the reason for the failure. Empty string if not completed or succeeded

property is_success: bool#

Indicates whether the request has successfully completed or not

class scrutiny.sdk.WatchableConfiguration[source]#

(Immutable struct) Represents a watchable available in the server datastore

server_id: str#

The unique ID assigned to that watchable item by the server

watchable_type: WatchableType#

The type of the item, either a Variable, an Alias or a Runtime Published Value

datatype: EmbeddedDataType#

The data type of the value in the embedded firmware that this watchable refers to

enum: EmbeddedEnum | None#

An optional enumeration associated with the possible values of the item

Contents