Accessing variables

Contents

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 (Aliases and Variable), 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’s address, type, size and endianness are 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 watchable’s path. 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, update_rate=None)[source]#

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

Parameters:

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

  • update_rate (float | None) – The update rate at which the server should update this value. The actual rate may be faster if another client requires a higher update frequency A value of None indicates that updates should occur as quickly as possible. This rate applies to the server/device communication while set_server_throttling controls the server/client communication

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 begins polling for its value. Each time the value is updated, the server broadcasts a value update to all subscribers, in this case, our client. Concurrently, a background thread actively listens for these updates and modifies the value that the WatchableHandle refers to accordingly.

Calling watch multiple times on the same element will always return the same handle. It is possible to query whether a handle already exists for a given element using try_get_existing_watch_handle and try_get_existing_watch_handle_by_server_id. A handle will exist if a previous call to watch has been done.

ScrutinyClient.try_get_existing_watch_handle(path)[source]#

Retrieve an existing watchable handle created after a call to watch() if it exists. This methods makes no request to the server and is therefore non-blocking.

Parameters:

path (str) – The path of the element being watched

Raises:

TypeError – Given parameter not of the expected type

Returns:

A handle that can read/write the watched element or None if the element is not being watched.

Return type:

WatchableHandle | None

ScrutinyClient.try_get_existing_watch_handle_by_server_id(server_id)[source]#

Retrieve an existing watchable handle created after a call to watch() if it exists, identified by its unique server_id. This methods makes no request to the server and is therefore non-blocking.

Parameters:

server_id (str) – The server_id assigned to the handle returned by watch()

Raises:

TypeError – Given parameter not of the expected type

Returns:

A handle that can read/write the watched element or None if the element is not being watched.

Return type:

WatchableHandle | None

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

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

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 server_path: str#

Returns the watchable full tree path given by the server

property name: str#

Returns the watchable name, e.g. the basename in the server_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 server_id: str#

The unique ID assigned by the server for this watchable

property value: int | float | bool#

The value without cast.

  • When reading, returns a int, float or bool.

  • When writing, accepts int, float, bool or a str.

If a string is assigned, the value is sent “as is” to the server which will then try to parse it. The server will accepts “true”, “false” or a mathematical expression supporting arithmetic operators (+, -, *, /, ^), base prefix (0x, 0b), scientific notation (1.5e-2), constants (such as pi) and common math functions. including: abs, exp, pow, sqrt, mod, ceil, floor, log, ln, log10, hypot, degrees, radians, cos, cosh, acos, sin, sinh, asin, tan, tanh, atan, atan2

Raises:

  • InvalidValueError – When reading, if the value has never been set or the handle is no longer valid.

  • OperationFailure – When writing, if the server fails to complete the write.

property value_bool: bool#

The value cast as bool

property value_int: int#

The value cast as int

property value_float: float#

The value cast 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.

Raises:

  • BadEnumError – When reading, if the watchable has no enum defined. When writing, if val is not a valid enumerator name.

  • ValueError – When writing, if the assigned value is not a str.

  • OperationFailure – When writing, if the server fails to complete the write.

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

property is_dead: bool#

Flag indicating if this handle is dead, meaning it will never be updated in the future. Once a handle is dead, it can be disposed of. Unwatching a handle will mark it “dead”.

property var_details: DetailedVarWatchableConfiguration#

Returns the variable-specific metadata.

Raises:

BadTypeError – If the watchable type is not Variable.

property alias_details: DetailedAliasWatchableConfiguration#

Returns the alias-specific metadata.

Raises:

BadTypeError – If the watchable type is not Alias.

property rpv_details: DetailedRPVWatchableConfiguration#

Returns the RPV-specific metadata.

Raises:

BadTypeError – If the watchable type is not RuntimePublishedValue.

class scrutiny.sdk.WatchableType[source]#

(Enum) Type of watchable available on the server

Variable = 'var'#

A variable found in the device firmware debug symbols

RuntimePublishedValue = 'rpv'#

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

Alias = 'alias'#

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

classmethod all()[source]#

Return the list of valid Watchable types. Mainly for unit testing

Return type:

List[WatchableType]

__new__(value)#

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).
    w3.value = 'floor(1.23e5*cos(radians(5^2)))' # The expression will be parsed by the server. The value written will be 111475

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 disconnected

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

    1. The server disconnects

    2. The device is disconnected

    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.

Watchable metadata#

It is possible to query the server for the watchable metadata, without actually subscribing for value updates. The metadata content varies according to the watchable type.

ScrutinyClient.get_watchable_info(paths)[source]#

Request the server for details about a list of watchables. The information returned is the same as one would have gotten after a call to watch, without actually subscribing to the server for updates.

Parameters:

paths (List[str]) – The server path of every watchable to query

Returns:

A dictionary mapping server path and detailed info structure

Raises:

  • ValueError – If paths is not valid

  • TypeError – Given parameter not of the expected type

  • OperationFailure – If the request fails in any way

Return type:

Dict[str, DetailedVarWatchableConfiguration | DetailedAliasWatchableConfiguration | DetailedRPVWatchableConfiguration]

For convenience, 3 specialization of get_watchable_info are available for the case only one watchable of a given type is of interest.

ScrutinyClient.get_var_watchable_info(path)[source]#

Performs a call to get_watchable_info for a single watchable of type Variable.

Parameters:

path (str) – Server path to the watchable

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • OperationFailure – If the request fails in any way

  • BadTypeError – If the requested watchable is not an variable

Return type:

DetailedVarWatchableConfiguration

ScrutinyClient.get_alias_watchable_info(path)[source]#

Performs a call to get_watchable_info for a single watchable of type Alias.

Parameters:

path (str) – Server path to the watchable

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • OperationFailure – If the request fails in any way

  • BadTypeError – If the requested watchable is not an alias

Return type:

DetailedAliasWatchableConfiguration

ScrutinyClient.get_rpv_watchable_info(path)[source]#

Performs a call to get_watchable_info for a single watchable of type RPV.

Parameters:

path (str) – Server path to the watchable

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • OperationFailure – If the request fails in any way

  • BadTypeError – If the requested watchable is not a Runtime Published Value

Return type:

DetailedRPVWatchableConfiguration

The metadata structures are as follow

class scrutiny.sdk.DetailedVarWatchableConfiguration[source]#

Structure containing the metadata of a watchable of type Variable

get_enum()#

Returns the enum associated with this watchable

Raises:

BadEnumError – If the watchable has no enum assigned

Return type:

EmbeddedEnum

has_enum()#

Tells if the watchable has an enum associated with it

Return type:

bool

parse_enum_val(val)#

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

address: int | None#

Absolute memory address of the variable in the firmware. This field will contain a value for static and global variables. The address will be None if and only if the watchable element is a pointed variable and the pointer element is either zero or unavailable (not read yet or part of a forbidden region)

bitoffset: int | None#

For bitfield, the startbit of the value. None if not a bitfield

bitsize: int | None#

For bitfield, the size in bits of the value. None if not a bitfield

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

property name: str#

Returns the watchable name, e.g. the basename in the server_path. If the server_path is /aaa/bbb/ccc, then this returns “ccc”

server_id: str#

The unique ID assigned to that watchable item by the server

server_path: str#

The path assigned by the server to this watchable

watchable_type: WatchableType#

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

class scrutiny.sdk.DetailedAliasWatchableConfiguration[source]#

Structure containing the metadata of a watchable of type Alias

get_enum()#

Returns the enum associated with this watchable

Raises:

BadEnumError – If the watchable has no enum assigned

Return type:

EmbeddedEnum

has_enum()#

Tells if the watchable has an enum associated with it

Return type:

bool

parse_enum_val(val)#

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

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

gain: float | None#

An optional gain. Scale the value when read or writen. Does not apply when None

max: float | None#

An optional upper bound that applies only when written. Does not apply when None

min: float | None#

An optional lower bound that applies only when written. Does not apply when None

property name: str#

Returns the watchable name, e.g. the basename in the server_path. If the server_path is /aaa/bbb/ccc, then this returns “ccc”

offset: float | None#

An optional offset. Add a bias to the value when read or writen. Does not apply when None

server_id: str#

The unique ID assigned to that watchable item by the server

server_path: str#

The path assigned by the server to this watchable

target: str#

The server path to the reference watchable (variable or alias)

target_type: WatchableType#

The type of watchable referenced by this alias

watchable_type: WatchableType#

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

class scrutiny.sdk.DetailedRPVWatchableConfiguration[source]#

Structure containing the metadata of a watchable of type RuntimePublishedValue

get_enum()#

Returns the enum associated with this watchable

Raises:

BadEnumError – If the watchable has no enum assigned

Return type:

EmbeddedEnum

has_enum()#

Tells if the watchable has an enum associated with it

Return type:

bool

parse_enum_val(val)#

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

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

property name: str#

Returns the watchable name, e.g. the basename in the server_path. If the server_path is /aaa/bbb/ccc, then this returns “ccc”

rpvid: int#

The numerical 16bits ID of this RPV

server_id: str#

The unique ID assigned to that watchable item by the server

server_path: str#

The path assigned by the server to this watchable

watchable_type: WatchableType#

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

Example#

my_var = client.get_var_watchable_info('/var/static/main.cpp/main()/my_var')
print(my_var.address)

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

  • BadEnumError – If value is a string and no enumerator value matches it

  • 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

  • InvalidValueError – If a watched element becomes invalid while waiting

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

Return type:

None

Batch writing#

Writing multiple 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 successful 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 proceeding to the next 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 to download their definition

This feature is typically not required for automation scripts; however, it can be necessary for presenting users with selectable watchable items. It is currently used 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 dictionary containing the number of watchables, classified by type

Return type:

Dict[ServerDatastoreContentType, int]

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

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

The returned data includes the path to the watchable and the properties that are common to all types of watchable (data type and enum) More information might be downlaoded from a watchable by either calling watch or get_watchable_info

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

  • partial_reception_callback (Callable[[WatchableListContentPart, bool], None] | None) – A callback to be called by the client whenever new data is received by the server. Data might be segmented in several parts. Expected signature : my_callback(data, last_segment) where data is a dictionary of the form data.<type>[path] = watchable and last_segment indicate if that data segment was the last one. If None is given, the received data will be stored inside the request object and can be fetched once the request has completed by calling get() Note This callback is called from an internal thread.

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.

Parameters:

  • client – A reference to the client object

  • request_id – The request ID of the initiating request

  • new_data_callback – A callback to process the segmented data as it comes in. Called from the internal thread.

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:
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 definition of each watchable entry that matched the filters. foo[type][server_path] = <definition>

Return type:

WatchableListContentPart

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.ServerDatastoreContentType[source]#

(Enum) Type of items that can be downloaded from the server. This enum contains all watchable types + some special items.

Variable = 'var'#

A variable found in the device firmware debug symbols.

RuntimePublishedValue = 'rpv'#

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

Alias = 'alias'#

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

VariableFactory = 'var_factory'#

A data structure containing a pattern that can generate variables. Mainly used for array instantiation

classmethod all()[source]#

Return the list of valid items. Mainly for unit testing

Return type:

List[ServerDatastoreContentType]

class scrutiny.sdk.BriefWatchableConfiguration[source]#

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

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

class scrutiny.sdk.WatchableListContentPart[source]#

A partial dataset received from the server when downloading the watchable list

var: Dict[str, BriefWatchableConfiguration]#

The variables in the server datastore

alias: Dict[str, BriefWatchableConfiguration]#

The aliases in the server datastore

rpv: Dict[str, BriefWatchableConfiguration]#

The runtime Published values in the server datastore

var_factory: Dict[str, VariableFactoryInterface]#

The Variable factories (to instantiate variables from arrays) in the server datastore

class scrutiny.sdk.VariableFactoryInterface[source]#

(Immutable struct) Represent a VariableFactory in the server. Can be used to generate locally all the variables that the server can instantiate from this factory. Mainly used to create every array elements from a set of dimensions

access_path: str#

The basic path of the factory with encoded array information

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

array_dims: Dict[str, Tuple[int, ...]]#

The dimensions of each subpath that are arrays element

count_possible_paths()[source]#

Returns how many path this factory can generate

Return type:

int

iterate_possible_paths()[source]#

Return all the variable display path that can be generated by this factory

Return type:

Generator[Tuple[str, BriefWatchableConfiguration], None, None]

Controlling the update rate#

The server is can be instructed to poll a watchable at a reduced rate to avoid unnecessary bandwidth usage on the device link. This can be usefull if a comunication link is fast and the value does not ened to eb refreshed often (displayed on a screen for example).

When calling ScrutinyClient.watch(), one can specify an optional update_rate value. This update rate can also be set or changed after the initial call to ScrutinyClient.watch(), by calling WatchableHandle.change_update_rate()

WatchableHandle.change_update_rate(update_rate)[source]#

Request the server to change the target update rate for this watchable (optionally set when calling watch()). When there are multiple clients watching the same watchable, the server applies the fastest required rate.

Parameters:

update_rate (float | None) – The new polling rate. A value of None indicates that updates should happen as fast as possible. Must be None or greater or equal to 1

Returns:

The effective update rate at the moment of change. May be higher or change later if another client requires it.

Raises:

  • TypeError – Given parameter not of the expected type

  • ValueError – Given parameter has an invalid value

  • OperationFailure – If the request fails to complete

Return type:

float | None

Example

from scrutiny.sdk.client import ScrutinyClient

client = ScrutinyClient()
with client.connect('localhost', 8765):
    w1 = client.watch('/global/my_var', update_rate=30) # 30 update/sec maximum
    w1.change_update_rate(None) # Go as fast as possible.
Contents