malcolm.core

This is the core of Malcolm

exception malcolm.core.AbortedError

The operation was aborted

class malcolm.core.Alarm(severity: Anno(name='AAlarmSeverity', typ=<enum 'AlarmSeverity'>, description='The alarm severity') = <AlarmSeverity.NO_ALARM: 0>, status: Anno(name='AAlarmStatus', typ=<enum 'AlarmStatus'>, description='The alarm status') = <AlarmStatus.NO_STATUS: 0>, message: Anno(name='AMessage', typ=<class 'str'>, description='A descriptive alarm message') = '')

Model representing a alarm state with severity, status and message

Parameters:

• severity (AlarmSeverity) – The alarm severity
• status (AlarmStatus) – The alarm status
• message (str) – A descriptive alarm message

ok = Alarm(severity=<AlarmSeverity.NO_ALARM: 0>, status=<AlarmStatus.NO_STATUS: 0>, message='')

class malcolm.core.AlarmSeverity

An alarm severity

NO_ALARM = 0

MINOR_ALARM = 1

MAJOR_ALARM = 2

INVALID_ALARM = 3

UNDEFINED_ALARM = 4

class malcolm.core.AlarmStatus

An alarm status

NO_STATUS = 0

DEVICE_STATUS = 1

DRIVER_STATUS = 2

RECORD_STATUS = 3

DB_STATUS = 4

CONF_STATUS = 5

UNDEFINED_STATUS = 6

CLIENT_STATUS = 7

class malcolm.core.Attribute(controller: Controller, context: malcolm.core.context.Context, data: malcolm.core.models.Model)

Represents a value with type information that may be backed elsewhere

put_value(value, timeout=None)

Put a value to the Attribute and wait for completion

class malcolm.core.AttributeModel(value: Anno(name='AValue', typ=typing.Any, description='The current value of the Attribute') = None, alarm: Anno(name='AAlarm', typ=<class 'malcolm.core.alarm.Alarm'>, description='The current alarm status') = None, timeStamp: Anno(name='ATimeStamp', typ=<class 'malcolm.core.timestamp.TimeStamp'>, description='The time when the value was last updated') = None, meta: Anno(name='AVMeta', typ=<class 'malcolm.core.models.VMeta'>, description='The validating Meta object describing our value') = None)

Data Model for an Attribute

Parameters:

• value – The current value of the Attribute
• alarm (Alarm) – The current alarm status
• timeStamp (TimeStamp) – The time when the value was last updated
• meta (VMeta) – The validating Meta object describing our value

meta

A Meta object describing the Attribute

value

The current value of the Attribute

alarm

An Alarm object indicating any problems

timeStamp

When value was last set

set_value(value: Any, set_alarm_ts: bool = True, alarm: malcolm.core.alarm.Alarm = None, ts: malcolm.core.timestamp.TimeStamp = None) → Any

Set value, calculating alarm and ts if requested

set_value_alarm_ts(value: Any, alarm: malcolm.core.alarm.Alarm, ts: malcolm.core.timestamp.TimeStamp) → None

Set value with pre-validated alarm and timeStamp

apply_change(path, *args) → None

Take a single change from a Delta and apply it to this model

exception malcolm.core.BadValueError

match_update() received one of the specified bad values

class malcolm.core.Block(controller, context, data)

Object consisting of a number of Attributes and Methods

class malcolm.core.BlockMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = True, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '', fields: Union[Anno(name='AFields', typ=<class 'str'>, description='The list of fields currently in the Block'), Sequence[str], str] = ())

Parameters:

• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element
• fields (str) – The list of fields currently in the Block

class malcolm.core.BlockModel

Data Model for a Block

class malcolm.core.BooleanArrayMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '')

Meta object containing information for a boolean array

Parameters:

• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element

validate(value: Any) → annotypes._array.Array[bool][bool]

Check if the value is valid returns it

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

default_widget() → malcolm.core.tags.Widget

Abstract function to return the default widget type

class malcolm.core.BooleanMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '')

Meta object containing information for a boolean

Parameters:

• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element

attribute_class

alias of NTScalar

validate(value: Any) → bool

Cast value to boolean and return it

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

default_widget() → malcolm.core.tags.Widget

Abstract function to return the default widget type

class malcolm.core.ChoiceArrayMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', choices: Union[Anno(name='AChoices', typ=<class 'str'>, description='Choices of valid strings'), Sequence[enum.Enum], Sequence[str]] = (), tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '')

Meta object containing information for a choice array

Parameters:

• description (str) – Description of what this element represents
• choices (str) – Choices of valid strings
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element

validate(value: Any) → annotypes._array.Array[str][str]

Check if the value is valid returns it

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

class malcolm.core.ChoiceMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', choices: Union[Anno(name='AChoices', typ=<class 'str'>, description='Choices of valid strings'), Sequence[enum.Enum], Sequence[str]] = (), tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '')

Meta object containing information for a enum

Parameters:

• description (str) – Description of what this element represents
• choices (str) – Choices of valid strings
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element

attribute_class

alias of NTScalar

validate(value: Any) → Union[enum.Enum, str]

Check if the value is valid returns it

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

default_widget() → malcolm.core.tags.Widget

Abstract function to return the default widget type

classmethod from_annotype(anno: annotypes._anno.Anno, writeable: bool, **kwargs) → malcolm.core.models.VMeta

Return an instance of this class from an Anno

class malcolm.core.Context(process: Process)

Helper allowing Future style access to Block Attributes and Methods

block_view(mri: str) → Any

Get a view of a block

Parameters:mri – The mri of the controller hosting the block Returns:The block we control Return type:Block

set_notify_dispatch_request(notify_dispatch_request, *args)

Set function to call just before requests are dispatched

Parameters:notify_dispatch_request (callable) – function will be called with request as single arg just before request is dispatched

ignore_stops_before_now()

Ignore any stops received before this point

stop()

Stops any wait_all_futures call with an AbortedError

put(path, value, timeout=None, event_timeout=None)

Puts a value to a path and returns when it completes

Parameters:

• path (list) – The path to put to
• value (object) – The value to set
• timeout (float) – time in seconds to wait for responses, wait forever if None
• event_timeout – maximum time in seconds to wait between each response event, wait forever if None

Returns:

The value after the put completes

put_async(path, value)

Puts a value to a path and returns immediately

Parameters:

• path (list) – The path to put to
• value (object) – The value to set

Returns:

A single Future which will resolve to the result

Return type:

Future

post(path, params=None, timeout=None, event_timeout=None)

Synchronously calls a method

Parameters:

• path (list) – The path to post to
• params (dict) – parameters for the call
• timeout (float) – time in seconds to wait for responses, wait forever if None
• event_timeout – maximum time in seconds to wait between each response event, wait forever if None

Returns:

the result from ‘method’

post_async(path, params=None)

Asynchronously calls a function on a child block

Parameters:

• path (list) – The path to post to
• params (dict) – parameters for the call

Returns:

as single Future that will resolve to the result

Return type:

Future

subscribe(path, callback, *args)

Subscribe to changes in a given attribute and call callback(future, value, *args) when it changes

Returns:A single Future which will resolve to the result Return type:Future

unsubscribe(future)

Terminates the subscription given by a future

Parameters:future (Future) – The future of the original subscription

unsubscribe_all(callback=False)

Send an unsubscribe for all active subscriptions

when_matches(path, good_value, bad_values=None, timeout=None, event_timeout=None)

Resolve when an path value equals value

Parameters:

• path (list) – The path to wait to
• good_value (object) – the value to wait for
• bad_values (list) – values to raise an error on
• timeout (float) – time in seconds to wait for responses, wait forever if None
• event_timeout – maximum time in seconds to wait between each response event, wait forever if None

when_matches_async(path, good_value, bad_values=None)

Wait for an attribute to become a given value

Parameters:

• path (list) – The path to wait to
• good_value – If it is a callable then expect it to return True if we are satisfied and raise on error. If it is not callable then compare each value against this one and return if it matches.
• bad_values (list) – values to raise an error on

Returns:

a single Future that will resolve when the path matches good_value or bad_values

Return type:

Future

wait_all_futures(futures: Union[List[malcolm.core.future.Future], malcolm.core.future.Future, None], timeout: float = None, event_timeout: float = None) → None

Services all futures until the list ‘futures’ are all done then returns. Calls relevant subscription callbacks as they come off the queue and raises an exception on abort

Parameters:

• futures – a Future or list of all futures that the caller wants to wait for
• timeout – maximum total time in seconds to wait for responses, wait forever if None
• event_timeout – maximum time in seconds to wait between each response event, wait forever if None

sleep(seconds)

Services all futures while waiting

Parameters:seconds (float) – Time to wait

class malcolm.core.Controller(mri: Anno(name='AMri', typ=<class 'str'>, description='The Malcolm Resource Identifier for the Block produced'), description: Anno(name='ADescription', typ=<class 'str'>, description='Description of the Block produced by the controller') = '')

Parameters:

• mri (str) – The Malcolm Resource Identifier for the Block produced
• description (str) – Description of the Block produced by the controller

make_view(context: malcolm.core.context.Context, data: malcolm.core.models.Model, child_name: str) → Any

Make a child View of data[child_name]

handle_request(request: malcolm.core.request.Request) → malcolm.core.concurrency.Spawned

Spawn a new thread that handles Request

class malcolm.core.Define(name: Anno(name='AName', typ=<class 'str'>, description='The name of the defined parameter'), value: Anno(name='AValue', typ=typing.Any, description='The value of the defined parameter'))

Parameters:

• name (str) – The name of the defined parameter
• value – The value of the defined parameter

class malcolm.core.Delta(id: Anno(name='AId', typ=<class 'int'>, description='ID that the Request was sent with') = 0, changes: Anno(name='AChanges', typ=typing.Any, description='List of [[path], value] pairs for changed values') = None)

Create a Delta Response object with the provided parameters

Parameters:

• id (int) – ID that the Request was sent with
• changes – List of [[path], value] pairs for changed values

to_dict(dict_cls=<class 'annotypes._frozen_dict.FrozenOrderedDict'>)

Create a dictionary representation of object attributes

Returns:OrderedDict serialised version of self

class malcolm.core.Display(limitLow: Union[Anno(name='ALimitLow', typ=<class 'numpy.float64'>, description='The lower bound of range within which the value must be set'), float] = 0, limitHigh: Union[Anno(name='ALimitHigh', typ=<class 'numpy.float64'>, description='The upper bound of range within which the value must be set'), float] = 0, description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', precision: Union[Anno(name='APrecision', typ=<class 'numpy.int32'>, description='Number of significant figures to display'), int] = 0, units: Anno(name='AUnits', typ=<class 'str'>, description='The units for the value') = '')

Parameters:

• limitLow (float64) – The lower bound of range within which the value must be set
• limitHigh (float64) – The upper bound of range within which the value must be set
• description (str) – Description of what this element represents
• precision (int32) – Number of significant figures to display
• units (str) – The units for the value

class malcolm.core.Error(id: Anno(name='AId', typ=<class 'int'>, description='ID that the Request was sent with') = 0, message: Anno(name='AMessage', typ=<class 'Exception'>, description='Error message exception') = Exception(''))

Create an Error Response object with the provided parameters

Parameters:

• id (int) – ID that the Request was sent with
• message (Exception) – Error message exception

exception malcolm.core.FieldError

Basically a KeyError but without quotation marks in error message

class malcolm.core.Future(context)

Represents the result of an asynchronous computation. This class has a similar API to concurrent.futures.Future but this simpler version is not thread safe

Parameters:context (Context) – The context to run under

done()

Return True if the future finished executing.

result(timeout=None)

Return the result of the call that the future represents.

Parameters:

timeout – The number of seconds to wait for the result if the future isn’t done. If None, then there is no limit on the wait time.

Returns:

The result of the call that the future represents.

Raises:

• TimeoutError – If the future didn’t finish executing before the given timeout.
• Exception – If the call raised then that exception will be raised.

exception(timeout=None)

Return the exception raised by the call that the future represents.

Parameters:timeout – The number of seconds to wait for the exception if the future isn’t done. If None, then there is no limit on the wait time. Returns:The exception raised by the call that the future represents or None if the call completed without raising. Raises:TimeoutError – If the future didn’t finish executing before the given timeout.

set_result(result)

Sets the return value of work associated with the future.

Should only be used by Task and unit tests.

set_exception(exception)

Sets the result of the future as being the given exception.

Should only be used by Task and unit tests.

class malcolm.core.Get(id: Anno(name='AId', typ=<class 'int'>, description='ID that should be used for any responses') = 0, path: Union[Anno(name='APath', typ=<class 'str'>, description='Path to target Block substructure'), Sequence[str], str] = None)

Create a Get Request object

Parameters:

• id (int) – ID that should be used for any responses
• path (str) – Path to target Block substructure

class malcolm.core.Hook(child: Anno(name='AHookable', typ=<class 'malcolm.core.hook.Hookable'>, description='The child that the hook is being passed to'), **kwargs)

Something that children can register with to be called

Parameters:child (Hookable) – The child that the hook is being passed to

prepare() → None

Override this if we need to prepare before running

stop() → None

Override this if we can stop

validate_return(ret: T) → Any

Override this if the function is expected to return something to to validate its value

class malcolm.core.Hookable

Baseclass of something that can be attached to a hook

register_hooked(hooks: Union[Type[Hook], Sequence[Type[Hook]]], func: Callable[[...], T], args_gen: Optional[Callable[[List[str]], List[str]]] = None) → None

Register func to be run when any of the hooks are run by parent

Parameters:

• hooks – A Hook class or list of Hook classes of interest
• func – The callable that should be run on that Hook
• args_gen – Optionally specify the argument names that should be passed to func. If not given then use func.call_types.keys

on_hook(hook: malcolm.core.hook.Hook) → None

Takes a hook, and optionally calls hook.run on a function

exception malcolm.core.IncompatibleError

Incompatibility between components of this scan

class malcolm.core.Info

Base class that should be inherited from when a part needs to return something from a hooked function

classmethod filter_parts(part_info: Mapping[str, Optional[Sequence[T_co]]]) → Dict[str, List[T]]

Filter the part_info dict looking for instances of our class

Parameters:part_info (dict) – {part_name: [Info] or None} as returned from Controller.run_hook() Returns:{part_name: [info]} where info is a subclass of cls Return type:dict

classmethod filter_values(part_info: Mapping[str, Optional[Sequence[T_co]]]) → List[T]

Filter the part_info dict list looking for instances of our class

Parameters:part_info (dict) – {part_name: [Info] or None} as returned from Controller.run_hook() Returns:[info] where info is a subclass of cls Return type:list

classmethod filter_single_value(part_info: Mapping[str, Optional[Sequence[T_co]]], error_msg: str = None) → T

Filter the part_info dict list looking for a single instance of our class

Parameters:

• part_info (dict) – {part_name: [Info] or None} as returned from Controller.run_hook()
• error_msg (str) – Optional specific error message to show if there isn’t a single value

Returns:

info subclass of cls

class malcolm.core.Loggable

Utility class that provides a named logger for a class instance

set_logger(**fields)

Change the name of the logger that log.* should call

Parameters:**fields – Extra fields to be logged. Logger name will be: “.”.join([<module_name>, <cls_name>] + fields_sorted_on_key)

class malcolm.core.MapMeta(elements: Optional[Anno(name='AElements', typ=(<class 'str'>, <class 'malcolm.core.models.VMeta'>), description='Meta objects that are used to describe the elements in the map')] = None, required: Union[Anno(name='ARequired', typ=<class 'str'>, description='The required elements in the map'), Sequence[str], str] = ())

An object containing a set of ScalarMeta objects

Parameters:

• elements – Meta objects that are used to describe the elements in the map
• required (str) – The required elements in the map

validate(param_dict: Optional[Mapping[str, Any]] = None, add_missing: bool = False) → Dict[str, Any]

Return a param dict in the right order, with the correct keys and values of the correct type with no extras or missing

class malcolm.core.Method(controller: Controller, context: malcolm.core.context.Context, data: malcolm.core.models.Model)

Exposes a function with metadata for arguments and return values

class malcolm.core.MethodLog(value: Anno(name='AMVValue', typ=(<class 'str'>, typing.Any), description='The last map this took/returned') = None, present: Union[Anno(name='APresent', typ=<class 'str'>, description='The elements that were supplied in the map'), Sequence[str], str] = (), alarm: Anno(name='AAlarm', typ=<class 'malcolm.core.alarm.Alarm'>, description='The current alarm status') = None, timeStamp: Anno(name='ATimeStamp', typ=<class 'malcolm.core.timestamp.TimeStamp'>, description='The time when the value was last updated') = None)

Exposes a function with metadata for arguments and return values

Parameters:

• value – The last map this took/returned
• present (str) – The elements that were supplied in the map
• alarm (Alarm) – The current alarm status
• timeStamp (TimeStamp) – The time when the value was last updated

class malcolm.core.MethodMeta(takes: Optional[Anno(name='ATakes', typ=<class 'malcolm.core.models.MapMeta'>, description='Meta for describing the arguments that should be passed')] = None, defaults: Optional[Anno(name='ADefaults', typ=(<class 'str'>, typing.Any), description='The required elements in the map')] = None, description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '', returns: Optional[Anno(name='AReturns', typ=<class 'malcolm.core.models.MapMeta'>, description='Meta for describing the arguments that will be returned')] = None)

Exposes a function with metadata for arguments and return values

Parameters:

• takes (MapMeta) – Meta for describing the arguments that should be passed
• defaults – The required elements in the map
• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element
• returns (MapMeta) – Meta for describing the arguments that will be returned

classmethod from_callable(func: Callable, description: str = None, returns: bool = True, without_takes: Sequence[str] = ()) → malcolm.core.models.MethodMeta

Return an instance of this class from a Callable

Parameters:

• func – @with_call_types decorated Callable to inspect
• description – Override description. If None use func.__doc__
• returns – If True then scan return_type too
• without_takes – A sequence of strings that should not appear in the takes structure

Returns:

A MethodMeta with takes and returns matching the input func

class malcolm.core.MethodModel(took: Anno(name='ATook', typ=<class 'malcolm.core.models.MethodLog'>, description='The last arguments that a method call took') = None, returned: Anno(name='AReturned', typ=<class 'malcolm.core.models.MethodLog'>, description='The last return value produced by a method call') = None, meta: Anno(name='AMethodMeta', typ=<class 'malcolm.core.models.MethodMeta'>, description='Meta for describing the arguments that will be returned') = None)

Exposes a function with last took and returned arguments

Parameters:

• took (MethodLog) – The last arguments that a method call took
• returned (MethodLog) – The last return value produced by a method call
• meta (MethodMeta) – Meta for describing the arguments that will be returned

class malcolm.core.Model

set_notifier_path(notifier: Union[malcolm.core.notifier.Notifier, malcolm.core.notifier.DummyNotifier], path: List[str]) → None

Sets the notifier, and the path from the path from block root

Parameters:

• notifier (Notifier) – The Notifier to tell when endpoint data changes
• path (list) – The absolute path to get to this object

apply_change(path: List[str], *args) → None

Take a single change from a Delta and apply it to this model

class malcolm.core.NTScalar(value: Anno(name='AValue', typ=typing.Any, description='The current value of the Attribute') = None, alarm: Anno(name='AAlarm', typ=<class 'malcolm.core.alarm.Alarm'>, description='The current alarm status') = None, timeStamp: Anno(name='ATimeStamp', typ=<class 'malcolm.core.timestamp.TimeStamp'>, description='The time when the value was last updated') = None, meta: Anno(name='AVMeta', typ=<class 'malcolm.core.models.VMeta'>, description='The validating Meta object describing our value') = None)

AttributeModel containing a StringMeta, BooleanMeta, NumberMeta or ChoiceMeta

Parameters:

• value – The current value of the Attribute
• alarm (Alarm) – The current alarm status
• timeStamp (TimeStamp) – The time when the value was last updated
• meta (VMeta) – The validating Meta object describing our value

class malcolm.core.NTScalarArray(value: Anno(name='AValue', typ=typing.Any, description='The current value of the Attribute') = None, alarm: Anno(name='AAlarm', typ=<class 'malcolm.core.alarm.Alarm'>, description='The current alarm status') = None, timeStamp: Anno(name='ATimeStamp', typ=<class 'malcolm.core.timestamp.TimeStamp'>, description='The time when the value was last updated') = None, meta: Anno(name='AVMeta', typ=<class 'malcolm.core.models.VMeta'>, description='The validating Meta object describing our value') = None)

AttributeModel containing a VArrayMeta

Parameters:

• value – The current value of the Attribute
• alarm (Alarm) – The current alarm status
• timeStamp (TimeStamp) – The time when the value was last updated
• meta (VMeta) – The validating Meta object describing our value

class malcolm.core.NTTable(value: Anno(name='AValue', typ=typing.Any, description='The current value of the Attribute') = None, alarm: Anno(name='AAlarm', typ=<class 'malcolm.core.alarm.Alarm'>, description='The current alarm status') = None, timeStamp: Anno(name='ATimeStamp', typ=<class 'malcolm.core.timestamp.TimeStamp'>, description='The time when the value was last updated') = None, meta: Anno(name='AVMeta', typ=<class 'malcolm.core.models.VMeta'>, description='The validating Meta object describing our value') = None)

AttributeModel containing a TableMeta

Parameters:

• value – The current value of the Attribute
• alarm (Alarm) – The current alarm status
• timeStamp (TimeStamp) – The time when the value was last updated
• meta (VMeta) – The validating Meta object describing our value

set_value_alarm_ts(value: Anno(name='AValue', typ=typing.Any, description='The current value of the Attribute'), alarm: malcolm.core.alarm.Alarm, ts: malcolm.core.timestamp.TimeStamp) → None

Set value with pre-validated alarm and timeStamp

class malcolm.core.NTUnion(value: Anno(name='AValue', typ=typing.Any, description='The current value of the Attribute') = None, alarm: Anno(name='AAlarm', typ=<class 'malcolm.core.alarm.Alarm'>, description='The current alarm status') = None, timeStamp: Anno(name='ATimeStamp', typ=<class 'malcolm.core.timestamp.TimeStamp'>, description='The time when the value was last updated') = None, meta: Anno(name='AVMeta', typ=<class 'malcolm.core.models.VMeta'>, description='The validating Meta object describing our value') = None)

AttributeModel containing a meta producing some object structure

Parameters:

• value – The current value of the Attribute
• alarm (Alarm) – The current alarm status
• timeStamp (TimeStamp) – The time when the value was last updated
• meta (VMeta) – The validating Meta object describing our value

exception malcolm.core.NotWriteableError

The field is not currently writeable, so cannot Put or Post to it

class malcolm.core.Notifier(mri: str, lock: cothread.cothread.RLock, block: BlockModel)

Object that can service callbacks on given endpoints

handle_subscribe(request: malcolm.core.request.Subscribe) → CallbackResponses

Handle a Subscribe request from outside. Called with lock taken

handle_unsubscribe(request: malcolm.core.request.Unsubscribe) → CallbackResponses

Handle a Unsubscribe request from outside. Called with lock taken

changes_squashed

Context manager to allow multiple calls to notify_change() to be made and all changes squashed into one consistent set. E.g:

with notifier.changes_squashed:

attr.set_value(1) attr.set_alarm(MINOR)

add_squashed_change(path: List[str], data: Any) → None

Register a squashed change to a particular path

Parameters:

• path (list) – The path of what has changed, relative from Block
• data (object) – The new data

add_squashed_delete(path: List[str]) → None

Register a squashed deletion of a particular path

Parameters:path (list) – The path of what has changed, relative from Block

class malcolm.core.NumberArrayMeta(dtype: Anno(name='ADtype', typ=<class 'str'>, description='Numpy dtype string') = 'float64', description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '', display: Anno(name='ADisplay', typ=<class 'malcolm.core.models.Display'>, description='Display info meta object') = None)

Meta object containing information for an array of numerical values

Parameters:

• dtype (str) – Numpy dtype string
• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element
• display (Display) – Display info meta object

validate(value: Any) → annotypes._array.Array

Check if the value is valid returns it

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

class malcolm.core.NumberMeta(dtype: Anno(name='ADtype', typ=<class 'str'>, description='Numpy dtype string') = 'float64', description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '', display: Anno(name='ADisplay', typ=<class 'malcolm.core.models.Display'>, description='Display info meta object') = None)

Meta object containing information for a numerical value

Parameters:

• dtype (str) – Numpy dtype string
• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element
• display (Display) – Display info meta object

attribute_class

alias of NTScalar

validate(value: Any) → numpy.number

Check if the value is valid returns it

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

default_widget() → malcolm.core.tags.Widget

Abstract function to return the default widget type

classmethod from_annotype(anno: annotypes._anno.Anno, writeable: bool, **kwargs) → malcolm.core.models.VMeta

Return an instance of this class from an Anno

class malcolm.core.Part(name: Anno(name='APartName', typ=<class 'str'>, description='The name of the Part within the Controller'))

Parameters:name (str) – The name of the Part within the Controller

setup(registrar: malcolm.core.part.PartRegistrar) → None

Use the given PartRegistrar to populate the hooks and fields. This function is called for all parts in a block when the block’s Controller is added to a Process

notify_dispatch_request(request: malcolm.core.request.Request) → None

Will be called when a context passed to a hooked function is about to dispatch a request

class malcolm.core.PartRegistrar(field_registry: malcolm.core.part.FieldRegistry, info_registry: malcolm.core.part.InfoRegistry, part: malcolm.core.part.Part)

Utility object that allows Parts to register Methods and Attributes with their parent Controller that will appear in the Block

hook(hooks: Union[Type[malcolm.core.hook.Hook], Sequence[Type[malcolm.core.hook.Hook]]], func: Callable[[...], T], args_gen: Optional[Callable[[List[str]], List[str]]] = None)

Register func to be run when any of the hooks are run by parent

Parameters:

• hooks – A Hook class or list of Hook classes of interest
• func – The callable that should be run on that Hook
• args_gen – Optionally specify the argument names that should be passed to func. If not given then use func.call_types.keys

add_method_model(func: Callable, name: Optional[str] = None, description: Optional[str] = None, needs_context: bool = False) → malcolm.core.models.MethodModel

Register a function to be added to the Block as a MethodModel

Parameters:

• func – The callable that will be called when the Method is called
• name – Override name, if None then take function __name__
• description – Override description, if None take function.__doc__
• needs_context – If True the “context” argument will be supplied to func with a newly created Context instance

add_attribute_model(name: str, attr: malcolm.core.models.AttributeModel, writeable_func: Optional[Callable] = None, needs_context: bool = False) → malcolm.core.models.AttributeModel

Register a pre-existing AttributeModel to be added to the Block

report(info: malcolm.core.info.Info) → None

Report an Info to the parent Controller

class malcolm.core.PathRequest(id: Anno(name='AId', typ=<class 'int'>, description='ID that should be used for any responses') = 0, path: Union[Anno(name='APath', typ=<class 'str'>, description='Path to target Block substructure'), Sequence[str], str] = None)

Parameters:

• id (int) – ID that should be used for any responses
• path (str) – Path to target Block substructure

class malcolm.core.Port

Enum with all the known flowgraph port tags to appear on Attribute Metas

BOOL = 'bool'

Boolean value. Typically used in PandA

INT32 = 'int32'

32-bit signed integer. Typically used in PandA

NDARRAY = 'NDArray'

areaDetector NDArray port

MOTOR = 'motor'

Motor record connection to CS or controller

BLOCK = 'block'

Malcolm level connection to another Block

sink_port_tag(disconnected_value)

Add a tag indicating this is a Sink Port of the given type

Parameters:disconnected_value – What value should the Attribute be set to when the port is disconnected

source_port_tag(connected_value)

Add a tag indicating this is a Source Port of the given type

Parameters:connected_value – What value should a Sink Port be set to if it is connected to this port

with_source_port_tag(tags, connected_value)

Add a Source Port tag to the tags list, removing any other Source Ports

class malcolm.core.Post(id: Anno(name='AId', typ=<class 'int'>, description='ID that should be used for any responses') = 0, path: Union[Anno(name='APath', typ=<class 'str'>, description='Path to target Block substructure'), Sequence[str], str] = None, parameters: Anno(name='AParameters', typ=(<class 'str'>, typing.Any), description='Parameters to use in a method Post') = None)

Create a Post Request object

Parameters:

• id (int) – ID that should be used for any responses
• path (str) – Path to target Block substructure
• parameters – Parameters to use in a method Post

class malcolm.core.Process(name: str = 'Process')

Hosts a number of Controllers and provides spawn capabilities

start(timeout=10.0)

Start the process going

Parameters:timeout (float) – Maximum amount of time to wait for each spawned process. None means forever

stop(timeout=10.0)

Stop the process and wait for it to finish

Parameters:timeout (float) – Maximum amount of time to wait for each spawned object. None means forever

spawn(function: Callable[[...], Any], *args, **kwargs) → malcolm.core.concurrency.Spawned

Runs the function in a worker thread, returning a Result object

Parameters:

• function – Function to run
• args – Positional arguments to run the function with
• kwargs – Keyword arguments to run the function with

Returns:

Something you can call wait(timeout) on to see when it’s

finished executing

Return type:

Spawned

add_controllers(controllers: List[malcolm.core.controller.Controller], timeout: float = None) → None

Add many controllers to be hosted by this process

Parameters:

• controllers (List[Controller]) – List of its controller
• timeout (float) – Maximum amount of time to wait for each spawned object. None means forever

add_controller(controller: malcolm.core.controller.Controller, timeout: float = None) → None

Add a controller to be hosted by this process

Parameters:

• controller (Controller) – Its controller
• timeout (float) – Maximum amount of time to wait for each spawned object. None means forever

get_controller(mri: str) → malcolm.core.controller.Controller

Get controller which can make Block views for this mri

block_view(mri: str) → Any

Get a Block view from a Controller with given mri

class malcolm.core.ProcessPublishHook(child: Anno(name='AHookable', typ=<class 'malcolm.core.hook.Hookable'>, description='The child that the hook is being passed to'), published: Anno(name='APublished', typ=<class 'str'>, description='The list of currently published Controller mris'))

Called when a new block is added

Parameters:

• child (Hookable) – The child that the hook is being passed to
• published (str) – The list of currently published Controller mris

class malcolm.core.ProcessStartHook(child: Anno(name='AHookable', typ=<class 'malcolm.core.hook.Hookable'>, description='The child that the hook is being passed to'), **kwargs)

Called at start() to start all child controllers

Parameters:child (Hookable) – The child that the hook is being passed to

validate_return(ret: Union[Anno(name='AUnpublishedInfos', typ=<class 'malcolm.core.process.UnpublishedInfo'>, description='Each of these reports that the controller should not be published'), Sequence[malcolm.core.process.UnpublishedInfo], malcolm.core.process.UnpublishedInfo, None]) -> Anno(name='AUnpublishedInfos', typ=<class 'malcolm.core.process.UnpublishedInfo'>, description='Each of these reports that the controller should not be published')

Check that all returns are UnpublishedInfo objects indicating that the controller shouldn’t be published via any server comms

class malcolm.core.ProcessStopHook(child: Anno(name='AHookable', typ=<class 'malcolm.core.hook.Hookable'>, description='The child that the hook is being passed to'), **kwargs)

Called at stop() to gracefully stop all child controllers

Parameters:child (Hookable) – The child that the hook is being passed to

class malcolm.core.Put(id: Anno(name='AId', typ=<class 'int'>, description='ID that should be used for any responses') = 0, path: Union[Anno(name='APath', typ=<class 'str'>, description='Path to target Block substructure'), Sequence[str], str] = None, value: Anno(name='AValue', typ=typing.Any, description='Value to put') = None, get: Anno(name='AGet', typ=<class 'bool'>, description='If set then return the current value in Return when Put completes') = False)

Create a Put Request object

Parameters:

• id (int) – ID that should be used for any responses
• path (str) – Path to target Block substructure
• value – Value to put
• get (bool) – If set then return the current value in Return when Put completes

class malcolm.core.Queue

Threadsafe and cothreadsafe queue with gets in calling thread

class malcolm.core.RLock

A reentrant lock.

acquire(timeout=None)

Acquire the lock if necessary and increment the recursion level.

release()

Release a lock, decrementing the recursion level.

class malcolm.core.Request(id: Anno(name='AId', typ=<class 'int'>, description='ID that should be used for any responses') = 0)

Request object that registers a callback for when action is complete.

Parameters:id (int) – ID that should be used for any responses

set_callback(callback: Callable[[malcolm.core.response.Response], None]) → None

Set the callback to be called on response

return_response(value: Any = None) → Tuple[Callable[[malcolm.core.response.Response], None], malcolm.core.response.Return]

Create a Return Response object to signal a return value

error_response(exception: Exception) → Tuple[Callable[[malcolm.core.response.Response], None], malcolm.core.response.Error]

Create an Error Response object to signal an error

generate_key() → Tuple[Callable[[malcolm.core.response.Response], None], int]

A key that will uniquely identify this request, for matching Subscribes up to Unsubscribes

class malcolm.core.Response(id: Anno(name='AId', typ=<class 'int'>, description='ID that the Request was sent with') = 0)

Represents a response to a Request

Parameters:id (int) – ID that the Request was sent with

exception malcolm.core.ResponseError

A Put or Post raised an Error

class malcolm.core.Return(id: Anno(name='AId', typ=<class 'int'>, description='ID that the Request was sent with') = 0, value: Anno(name='AValue', typ=typing.Any, description='Return value of the request') = None)

Represents a return from a Put or Post

Parameters:

• id (int) – ID that the Request was sent with
• value – Return value of the request

class malcolm.core.Spawned(func: Callable[[...], Any], args: Tuple, kwargs: Dict[KT, VT])

Internal object keeping track of a spawned function

ready() → bool

Return True if the spawned result has returned or errored

get(timeout: float = None) → T

Return the result or raise the error the function has produced

class malcolm.core.StringArrayMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '')

Meta object containing information for a string array

Parameters:

• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element

validate(value: Any) → annotypes._array.Array

Check if the value is valid returns it

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

default_widget() → malcolm.core.tags.Widget

Abstract function to return the default widget type

class malcolm.core.StringMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '')

Meta object containing information for a string

Parameters:

• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element

attribute_class

alias of NTScalar

validate(value: Any) → str

Check if the value is valid returns it

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

default_widget() → malcolm.core.tags.Widget

Abstract function to return the default widget type

class malcolm.core.Subscribe(id: Anno(name='AId', typ=<class 'int'>, description='ID that should be used for any responses') = 0, path: Union[Anno(name='APath', typ=<class 'str'>, description='Path to target Block substructure'), Sequence[str], str] = None, delta: Anno(name='ADifferences', typ=<class 'bool'>, description='Notify of differences only') = False)

Create a Subscribe Request object

Parameters:

• id (int) – ID that should be used for any responses
• path (str) – Path to target Block substructure
• delta (bool) – Notify of differences only

update_response(value: Any) → Tuple[Callable[[malcolm.core.response.Response], None], malcolm.core.response.Update]

Create an Update Response object to handle the request

delta_response(changes: List[List[Union[List[str], Any]]]) → Tuple[Callable[[malcolm.core.response.Response], None], malcolm.core.response.Delta]

“Create a Delta Response object to handle the request

class malcolm.core.Table

class malcolm.core.TableMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '', elements: Anno(name='ATableElements', typ=(<class 'str'>, <class 'malcolm.core.models.VArrayMeta'>), description='Elements that should appear in the table instance') = None)

Parameters:

• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element
• elements – Elements that should appear in the table instance

attribute_class

alias of NTTable

set_elements(elements: Anno(name='ATableElements', typ=(<class 'str'>, <class 'malcolm.core.models.VArrayMeta'>), description='Elements that should appear in the table instance')) -> Anno(name='ATableElements', typ=(<class 'str'>, <class 'malcolm.core.models.VArrayMeta'>), description='Elements that should appear in the table instance')

Set the elements dict from a serialized dict

validate(value: Any) → Any

Abstract function to validate a given value

Parameters:value – Value to validate Returns:The validated value if it passes

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

default_widget() → malcolm.core.tags.Widget

Abstract function to return the default widget type

classmethod from_table(table_cls: Type[malcolm.core.table.Table], description: str, widget: malcolm.core.tags.Widget = None, writeable: List[str] = [], extra_tags: List[str] = []) → malcolm.core.models.TableMeta

Create a TableMeta object, using a Table subclass as the spec

Parameters:

• table_cls – The Table class to read __init__ args from
• description – The description of the created Meta
• widget – The widget of the created Meta
• writeable – A list of the writeable field names. If there are any writeable fields then the whole Meta is writeable
• extra_tags – A list of tags to be added to the table meta

classmethod from_annotype(anno: annotypes._anno.Anno, writeable: bool, **kwargs) → malcolm.core.models.VMeta

Return an instance of this class from an Anno

class malcolm.core.TimeStamp(secondsPastEpoch: Anno(name='ASecondsPastEpoch', typ=<class 'numpy.int64'>, description='Seconds since Jan 1, 1970 00:00:00 UTC') = None, nanoseconds: Anno(name='ANanoseconds', typ=<class 'numpy.int32'>, description='Nanoseconds relative to the secondsPastEpoch field') = None, userTag: Anno(name='AUserTag', typ=<class 'numpy.int32'>, description='An integer value whose interpretation is deliberately undefined') = 0)

Parameters:

• secondsPastEpoch (int64) – Seconds since Jan 1, 1970 00:00:00 UTC
• nanoseconds (int32) – Nanoseconds relative to the secondsPastEpoch field
• userTag (int32) – An integer value whose interpretation is deliberately undefined

zero = TimeStamp(secondsPastEpoch=0, nanoseconds=0, userTag=0)

exception malcolm.core.TimeoutError

The operation exceeded the given deadline

exception malcolm.core.UnexpectedError

We received an object we weren’t expecting

class malcolm.core.UnpublishedInfo(mri: str)

class malcolm.core.Unsubscribe(id: Anno(name='AId', typ=<class 'int'>, description='ID that should be used for any responses') = 0)

Create an Unsubscribe Request object

Parameters:id (int) – ID that should be used for any responses

class malcolm.core.Update(id: Anno(name='AId', typ=<class 'int'>, description='ID that the Request was sent with') = 0, value: Anno(name='AValue', typ=typing.Any, description='Return value of the request') = None)

Create an Update Response object with the provided parameters

Parameters:

• id (int) – ID that the Request was sent with
• value – Return value of the request

class malcolm.core.VArrayMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '')

Intermediate abstract class so TableMeta can say “only arrays”

Parameters:

• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element

attribute_class

alias of NTScalarArray

class malcolm.core.VMeta(description: Anno(name='AMetaDescription', typ=<class 'str'>, description='Description of what this element represents') = '', tags: Union[Anno(name='ATags', typ=<class 'str'>, description='Generic text tags for client tools to interpret'), Sequence[str], str] = (), writeable: Anno(name='AWriteable', typ=<class 'bool'>, description='Whether this element is currently writeable') = False, label: Anno(name='ALabel', typ=<class 'str'>, description='A human readable label for the element') = '')

Abstract base class for validating the values of Attributes

Parameters:

• description (str) – Description of what this element represents
• tags (str) – Generic text tags for client tools to interpret
• writeable (bool) – Whether this element is currently writeable
• label (str) – A human readable label for the element

validate(value: Any) → Any

Abstract function to validate a given value

Parameters:value – Value to validate Returns:The validated value if it passes

create_attribute_model(initial_value: Any = None) → malcolm.core.models.AttributeModel

Make an AttributeModel instance of the correct type for this Meta

Parameters:initial_value – The initial value the Attribute should take Returns:The created attribute model instance Return type:AttributeModel

doc_type_string() → str

Abstract function to return the python type string.

For example, “str” or “numpy.int32”

default_widget() → malcolm.core.tags.Widget

Abstract function to return the default widget type

classmethod from_annotype(anno: annotypes._anno.Anno, writeable: bool, **kwargs) → malcolm.core.models.VMeta

Return an instance of this class from an Anno

classmethod register_annotype_converter(types: Union[Sequence[type], type], is_array: bool = False, is_mapping: bool = False) → annotypes._anno.Anno

Register this class as a converter for Anno instances

classmethod lookup_annotype_converter(anno: annotypes._anno.Anno) → Type[malcolm.core.models.VMeta]

Look up a vmeta based on an Anno

class malcolm.core.Widget

Enum with all the known widget tags to appear on Attribute Metas

NONE = ''

Force no widget

TEXTINPUT = 'textinput'

Editable text input box

TEXTUPDATE = 'textupdate'

Read only text update

MULTILINETEXTUPDATE = 'multilinetextupdate'

Multi line text update

LED = 'led'

On/Off LED indicator

COMBO = 'combo'

Select from a number of choice values

ICON = 'icon'

This field gives the SVG icon for the whole Block

HELP = 'help'

Gives a URL for the help documentation for the Block

GROUP = 'group'

Expandable section that other Attributes can appear in

TABLE = 'table'

Table of rows with a widget type for each column

CHECKBOX = 'checkbox'

A box that can be checked or not

FLOWGRAPH = 'flowgraph'

Boxes with lines for child block connections

TREE = 'tree'

A nested tree of object models editor

PLOT = 'plot'

Waveform plot

METER = 'meter'

progress meter

IMGURL = 'imgurl'

url of an image

tag()

Return the actual tag for the given Widget

E.g.

Widget.LED.tag() -> “widget:led”

exception malcolm.core.YamlError

When instantiating some YAML raises an error

malcolm.core.badge_value_tag(mri: str, attribute_name: str, display: str = 'plus') → str

Marks this Port as having another attribute in another Block that should be displayed as its badge.

Parameters:

• mri – The mri of the Block that provides the badge value
• attribute_name – The attribute name in the Block given by mri that provides the badge value
• display – The type of formatting to apply to the badge value: “plus”: If value < 1 then hide, otherwise display +%d

malcolm.core.camel_to_title(name)

Takes a camelCaseFieldName and returns an Title Case Field Name

Parameters:name (str) – E.g. camelCaseFieldName Returns:Title Case converted name. E.g. Camel Case Field Name Return type:str

malcolm.core.config_tag(iteration: int = 1) → str

Marks this field as a value that should be saved and loaded at config

Parameters:iteration – All iterations are sorted in increasing order and done in batches of the same iteration number

malcolm.core.get_config_tag(tags: Sequence[str]) → Optional[str]

Get the config_tag from tags or return None

malcolm.core.group_tag(group_name: str) → str

Marks this field as belonging to a group

malcolm.core.linked_value_tag(mri: str, attribute_name: str) → str

Marks this field as having another attribute in another Block that should be displayed below it as a linked value

malcolm.core.method_return_unpacked()

This method has a single element returns, and when called will return this single element unpacked rather than in a single element map

E.g.

hello.greet(“me”) -> “Hello me” not {“return”: “Hello me”}

malcolm.core.sleep(timeout)

Sleep until the specified timeout has expired.

malcolm.core.snake_to_camel(name)

Takes a snake_field_name and returns a camelCaseFieldName

Parameters:name (str) – E.g. snake_field_name or SNAKE_FIELD_NAME Returns:camelCase converted name. E.g. capsFieldName Return type:str

malcolm.core.version_tag()

Tag with the current version of pymalcolm

malcolm.core.without_config_tags(tags)

Return a new list of tags without any config tags

malcolm.core.without_group_tags(tags: Sequence[str]) → List[str]

Return a new list of tags without any group tags

malcolm.core.without_linked_value_tags(tags: Sequence[str]) → List[str]

Return a new list of tags without any group tags