labgrid package

Subpackages

Submodules

labgrid.binding module

exception labgrid.binding.StateError(msg)[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='msg'),)
__firstlineno__ = 8
__init__(msg) → None

Method generated by attrs for class StateError.

__match_args__ = ('msg',)
__module__ = 'labgrid.binding'
__replace__(**changes)

Method generated by attrs for class StateError.

__repr__()

Method generated by attrs for class StateError.

__static_attributes__ = ()
__weakref__

list of weak references to the object

exception labgrid.binding.BindingError(msg)[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='msg'),)
__firstlineno__ = 13
__init__(msg) → None

Method generated by attrs for class BindingError.

__match_args__ = ('msg',)
__module__ = 'labgrid.binding'
__replace__(**changes)

Method generated by attrs for class BindingError.

__repr__()

Method generated by attrs for class BindingError.

__static_attributes__ = ()
__weakref__

list of weak references to the object

class labgrid.binding.BindingState(*values)[source]

Bases: Enum

error = -1
idle = 0
bound = 1
active = 2
__module__ = 'labgrid.binding'
class labgrid.binding.BindingMixin(target, name)[source]

Bases: object

Handles the binding and activation of drivers and their supplying resources and drivers.

One client can be bound to many suppliers, and one supplier can be bound by many clients.

Conflicting access to one supplier can be avoided by deactivating conflicting clients before activation (using the resolve_conflicts callback).

bindings: Dict[str, Any] = {}
__attrs_post_init__()[source]
property display_name
on_supplier_bound(supplier)[source]

Called by the Target after a new supplier has been bound

on_client_bound(client)[source]

Called by the Target after a new client has been bound

on_activate()[source]

Called by the Target when this object has been activated

on_deactivate()[source]

Called by the Target when this object has been deactivated

resolve_conflicts(client)[source]

Called by the Target to allow this object to deactivate conflicting clients.

get_bound_resources()[source]

Called by the Target to find the correct driver corresponding to a resource

classmethod check_active(func)[source]
classmethod check_bound(func)[source]
class NamedBinding(value)[source]

Bases: object

Marks a binding (or binding set) as requiring an explicit name.

__init__(value)[source]
__repr__()[source]

Return repr(self).

__dict__ = mappingproxy({'__module__': 'labgrid.binding', '__firstlineno__': 122, '__doc__': '\nMarks a binding (or binding set) as requiring an explicit name.\n', '__init__': <function BindingMixin.NamedBinding.__init__>, '__repr__': <function BindingMixin.NamedBinding.__repr__>, '__static_attributes__': ('value',), '__dict__': <attribute '__dict__' of 'NamedBinding' objects>, '__weakref__': <attribute '__weakref__' of 'NamedBinding' objects>, '__annotations__': {}})
__firstlineno__ = 122
__module__ = 'labgrid.binding'
__static_attributes__ = ('value',)
__weakref__

list of weak references to the object

__annotations__ = {'bindings': typing.Dict[str, typing.Any]}
__attrs_attrs__ = (Attribute(name='target', default=NOTHING, validator=None, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='target'), Attribute(name='name', default=NOTHING, validator=<optional validator for <instance_of validator for type <class 'str'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='name'), Attribute(name='state', default=<BindingState.idle: 0>, validator=None, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=False, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='state'))
__dict__ = mappingproxy({'__module__': 'labgrid.binding', '__firstlineno__': 26, '__annotations__': {'bindings': typing.Dict[str, typing.Any]}, '__doc__': '\nHandles the binding and activation of drivers and their supplying resources\nand drivers.\n\nOne client can be bound to many suppliers, and one supplier can be bound by\nmany clients.\n\nConflicting access to one supplier can be avoided by deactivating\nconflicting clients before activation (using the resolve_conflicts\ncallback).\n', 'bindings': {}, '__attrs_post_init__': <function BindingMixin.__attrs_post_init__>, 'display_name': <property object>, 'on_supplier_bound': <function BindingMixin.on_supplier_bound>, 'on_client_bound': <function BindingMixin.on_client_bound>, 'on_activate': <function BindingMixin.on_activate>, 'on_deactivate': <function BindingMixin.on_deactivate>, 'resolve_conflicts': <function BindingMixin.resolve_conflicts>, 'get_bound_resources': <function BindingMixin.get_bound_resources>, 'check_active': <classmethod(<function BindingMixin.check_active>)>, 'check_bound': <classmethod(<function BindingMixin.check_bound>)>, 'NamedBinding': <class 'labgrid.binding.BindingMixin.NamedBinding'>, '__static_attributes__': ('clients', 'suppliers', 'target'), '__dict__': <attribute '__dict__' of 'BindingMixin' objects>, '__weakref__': <attribute '__weakref__' of 'BindingMixin' objects>, '__attrs_attrs__': (Attribute(name='target', default=NOTHING, validator=None, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='target'), Attribute(name='name', default=NOTHING, validator=<optional validator for <instance_of validator for type <class 'str'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='name'), Attribute(name='state', default=<BindingState.idle: 0>, validator=None, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=False, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='state')), '__replace__': <function BindingMixin.<lambda>>, '__match_args__': ('target', 'name'), '__repr__': <function BindingMixin.__repr__>, '__init__': <function BindingMixin.__init__>})
__firstlineno__ = 26
__init__(target, name) → None

Method generated by attrs for class BindingMixin.

__match_args__ = ('target', 'name')
__module__ = 'labgrid.binding'
__replace__(**changes)

Method generated by attrs for class BindingMixin.

__repr__()

Method generated by attrs for class BindingMixin.

__static_attributes__ = ('clients', 'suppliers', 'target')
__weakref__

list of weak references to the object

labgrid.config module

Config convenience class

This class encapsulates access functions to the environment configuration

class labgrid.config.Config(filename)[source]

Bases: object

__attrs_post_init__()[source]
resolve_path(path)[source]

Resolve an absolute path

Parameters:

path (str) – path to resolve

Returns:

the absolute path

Return type:

str

resolve_path_str_or_list(path)[source]

Resolves a single path or multiple paths. Always returns a list (containing a single or multiple resolved paths).

Parameters:

path (str, list) – path(s) to resolve

Returns:

absolute path(s

Return type:

list

Raises:

TypeError – if input is neither str nor list

get_tool(tool)[source]

Retrieve an entry from the tools subkey

Parameters:

tool (str) – the tool to retrieve the path for

Returns:

path to the requested tools

Return type:

str

get_image_path(kind)[source]

Retrieve an entry from the images subkey

Parameters:

kind (str) – the kind of the image to retrieve the path for

Returns:

path to the image

Return type:

str

Raises:

KeyError – if the requested image can not be found in the configuration

get_path(kind)[source]

Retrieve an entry from the paths subkey

Parameters:

kind (str) – the type of path to retrieve the path for

Returns:

path to the path

Return type:

str

Raises:

KeyError – if the requested image can not be found in the configuration

get_option(name, default=None)[source]

Retrieve an entry from the options subkey

Parameters:

  • name (str) – name of the option

  • default (any) – A default parameter in case the option can not be found

Returns:

value of the option or default parameter

Return type:

any

Raises:

KeyError – if the requested image can not be found in the configuration

set_option(name, value)[source]

Set an entry in the options subkey

Parameters:

  • name (str) – name of the option

  • value (any) – the new value

get_target_option(target, name, default=None)[source]
Retrieve an entry from the options subkey under the specified target

subkey

Parameters:

  • target (str) – name of the target

  • name (str) – name of the option

  • default (any) – A default parameter in case the option can not be found

Returns:

value of the option or default parameter

Return type:

any

Raises:

KeyError – if the requested key can not be found in the configuration, or if the target can not be found in the configuration.

set_target_option(target, name, value)[source]

Set an entry in the options subkey under the specified target subkey

Parameters:

  • target (str) – name of the target

  • name (str) – name of the option

  • value (any) – the new value

Raises:

KeyError – if the requested target can not be found in the configuration

get_targets()[source]
get_imports()[source]

Helper function that returns the list of all imports

Returns:

List of files which should be imported

Return type:

List

get_paths()[source]

Helper function that returns the subdict of all paths

Returns:

Dictionary containing all path definitions

Return type:

Dict

get_images()[source]

Helper function that returns the subdict of all images

Returns:

Dictionary containing all image definitions

Return type:

Dict

get_features()[source]
__attrs_attrs__ = (Attribute(name='filename', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='filename'),)
__dict__ = mappingproxy({'__module__': 'labgrid.config', '__firstlineno__': 14, '__attrs_post_init__': <function Config.__attrs_post_init__>, 'resolve_path': <function Config.resolve_path>, 'resolve_path_str_or_list': <function Config.resolve_path_str_or_list>, 'get_tool': <function Config.get_tool>, 'get_image_path': <function Config.get_image_path>, 'get_path': <function Config.get_path>, 'get_option': <function Config.get_option>, 'set_option': <function Config.set_option>, 'get_target_option': <function Config.get_target_option>, 'set_target_option': <function Config.set_target_option>, 'get_targets': <function Config.get_targets>, 'get_imports': <function Config.get_imports>, 'get_paths': <function Config.get_paths>, 'get_images': <function Config.get_images>, 'get_features': <function Config.get_features>, '__static_attributes__': ('base', 'data'), '__dict__': <attribute '__dict__' of 'Config' objects>, '__weakref__': <attribute '__weakref__' of 'Config' objects>, '__doc__': None, '__attrs_attrs__': (Attribute(name='filename', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='filename'),), '__replace__': <function Config.<lambda>>, '__match_args__': ('filename',), '__repr__': <function Config.__repr__>, '__init__': <function Config.__init__>, '__annotations__': {}})
__firstlineno__ = 14
__init__(filename) → None

Method generated by attrs for class Config.

__match_args__ = ('filename',)
__module__ = 'labgrid.config'
__replace__(**changes)

Method generated by attrs for class Config.

__repr__()

Method generated by attrs for class Config.

__static_attributes__ = ('base', 'data')
__weakref__

list of weak references to the object

labgrid.consoleloggingreporter module

class labgrid.consoleloggingreporter.ConsoleLoggingReporter(logpath)[source]

Bases: object

ConsoleLoggingReporter - Reporter that writes console log files

Parameters:

logpath (str) – path to store the logfiles in

instance = None
classmethod start(path)[source]

starts the ConsoleLoggingReporter

classmethod stop()[source]

stops the ConsoleLoggingReporter

__init__(logpath)[source]
get_logfile(event)[source]

Returns the correct file handle from cache or creates a new file handle

notify(event)[source]

This is the callback function for steps

__dict__ = mappingproxy({'__module__': 'labgrid.consoleloggingreporter', '__firstlineno__': 8, '__doc__': 'ConsoleLoggingReporter - Reporter that writes console log files\n\nArgs:\n    logpath (str): path to store the logfiles in\n', 'instance': None, 'start': <classmethod(<function ConsoleLoggingReporter.start>)>, 'stop': <classmethod(<function ConsoleLoggingReporter.stop>)>, '__init__': <function ConsoleLoggingReporter.__init__>, '_stop': <function ConsoleLoggingReporter._stop>, 'get_logfile': <function ConsoleLoggingReporter.get_logfile>, 'notify': <function ConsoleLoggingReporter.notify>, '__static_attributes__': ('_logcache', 'logpath'), '__dict__': <attribute '__dict__' of 'ConsoleLoggingReporter' objects>, '__weakref__': <attribute '__weakref__' of 'ConsoleLoggingReporter' objects>, '__annotations__': {}})
__firstlineno__ = 8
__module__ = 'labgrid.consoleloggingreporter'
__static_attributes__ = ('_logcache', 'logpath')
__weakref__

list of weak references to the object

labgrid.environment module

class labgrid.environment.Environment(config_file='config.yaml', interact=<built-in function input>)[source]

Bases: object

An environment encapsulates targets.

__attrs_post_init__()[source]
get_target(role: str = 'main') → Target | None[source]

Returns the specified target or None if not found.

Each target is initialized as needed.

get_features()[source]
get_target_features()[source]
cleanup()[source]
__attrs_attrs__ = (Attribute(name='config_file', default='config.yaml', validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='config_file'), Attribute(name='interact', default=<built-in function input>, validator=None, repr=False, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='interact'))
__dict__ = mappingproxy({'__module__': 'labgrid.environment', '__firstlineno__': 9, '__doc__': 'An environment encapsulates targets.', '__attrs_post_init__': <function Environment.__attrs_post_init__>, 'get_target': <function Environment.get_target>, 'get_features': <function Environment.get_features>, 'get_target_features': <function Environment.get_target_features>, 'cleanup': <function Environment.cleanup>, '__static_attributes__': ('config', 'targets'), '__dict__': <attribute '__dict__' of 'Environment' objects>, '__weakref__': <attribute '__weakref__' of 'Environment' objects>, '__attrs_attrs__': (Attribute(name='config_file', default='config.yaml', validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='config_file'), Attribute(name='interact', default=<built-in function input>, validator=None, repr=False, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='interact')), '__replace__': <function Environment.<lambda>>, '__match_args__': ('config_file', 'interact'), '__repr__': <function Environment.__repr__>, '__init__': <function Environment.__init__>, '__annotations__': {}})
__firstlineno__ = 9
__init__(config_file='config.yaml', interact=<built-in function input>) → None

Method generated by attrs for class Environment.

__match_args__ = ('config_file', 'interact')
__module__ = 'labgrid.environment'
__replace__(**changes)

Method generated by attrs for class Environment.

__repr__()

Method generated by attrs for class Environment.

__static_attributes__ = ('config', 'targets')
__weakref__

list of weak references to the object

labgrid.exceptions module

exception labgrid.exceptions.NoConfigFoundError(msg)[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='msg'),)
__firstlineno__ = 4
__init__(msg) → None

Method generated by attrs for class NoConfigFoundError.

__match_args__ = ('msg',)
__module__ = 'labgrid.exceptions'
__replace__(**changes)

Method generated by attrs for class NoConfigFoundError.

__repr__()

Method generated by attrs for class NoConfigFoundError.

__static_attributes__ = ()
__weakref__

list of weak references to the object

exception labgrid.exceptions.NoSupplierFoundError(msg, filter=None)[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='msg'), Attribute(name='filter', default=None, validator=<optional validator for <instance_of validator for type <class 'set'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='filter'))
__firstlineno__ = 9
__init__(msg, filter=None) → None

Method generated by attrs for class NoSupplierFoundError.

__match_args__ = ('msg', 'filter')
__module__ = 'labgrid.exceptions'
__replace__(**changes)

Method generated by attrs for class NoSupplierFoundError.

__repr__()

Method generated by attrs for class NoSupplierFoundError.

__static_attributes__ = ()
__weakref__

list of weak references to the object

exception labgrid.exceptions.InvalidConfigError(msg)[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='msg'),)
__firstlineno__ = 18
__init__(msg) → None

Method generated by attrs for class InvalidConfigError.

__match_args__ = ('msg',)
__module__ = 'labgrid.exceptions'
__replace__(**changes)

Method generated by attrs for class InvalidConfigError.

__repr__()

Method generated by attrs for class InvalidConfigError.

__static_attributes__ = ()
__weakref__

list of weak references to the object

exception labgrid.exceptions.NoDriverFoundError(msg, filter=None)[source]

Bases: NoSupplierFoundError

__annotations__ = {}
__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=True, on_setattr=None, alias='msg'), Attribute(name='filter', default=None, validator=<optional validator for <instance_of validator for type <class 'set'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=True, on_setattr=None, alias='filter'))
__firstlineno__ = 23
__init__(msg, filter=None) → None

Method generated by attrs for class NoDriverFoundError.

__match_args__ = ('msg', 'filter')
__module__ = 'labgrid.exceptions'
__replace__(**changes)

Method generated by attrs for class NoDriverFoundError.

__repr__()

Method generated by attrs for class NoDriverFoundError.

__static_attributes__ = ()
exception labgrid.exceptions.NoResourceFoundError(msg, filter=None, found=None)[source]

Bases: NoSupplierFoundError

__annotations__ = {}
__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=True, on_setattr=None, alias='msg'), Attribute(name='filter', default=None, validator=<optional validator for <instance_of validator for type <class 'set'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=True, on_setattr=None, alias='filter'), Attribute(name='found', default=None, validator=<optional validator for <instance_of validator for type <class 'list'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='found'))
__firstlineno__ = 28
__init__(msg, filter=None, found=None) → None

Method generated by attrs for class NoResourceFoundError.

__match_args__ = ('msg', 'filter', 'found')
__module__ = 'labgrid.exceptions'
__replace__(**changes)

Method generated by attrs for class NoResourceFoundError.

__repr__()

Method generated by attrs for class NoResourceFoundError.

__static_attributes__ = ()
exception labgrid.exceptions.NoStrategyFoundError(msg, filter=None)[source]

Bases: NoSupplierFoundError

__annotations__ = {}
__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=True, on_setattr=None, alias='msg'), Attribute(name='filter', default=None, validator=<optional validator for <instance_of validator for type <class 'set'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=True, on_setattr=None, alias='filter'))
__firstlineno__ = 36
__init__(msg, filter=None) → None

Method generated by attrs for class NoStrategyFoundError.

__match_args__ = ('msg', 'filter')
__module__ = 'labgrid.exceptions'
__replace__(**changes)

Method generated by attrs for class NoStrategyFoundError.

__repr__()

Method generated by attrs for class NoStrategyFoundError.

__static_attributes__ = ()
exception labgrid.exceptions.RegistrationError(msg)[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='msg'),)
__firstlineno__ = 41
__init__(msg) → None

Method generated by attrs for class RegistrationError.

__match_args__ = ('msg',)
__module__ = 'labgrid.exceptions'
__replace__(**changes)

Method generated by attrs for class RegistrationError.

__repr__()

Method generated by attrs for class RegistrationError.

__static_attributes__ = ()
__weakref__

list of weak references to the object

labgrid.factory module

class labgrid.factory.TargetFactory[source]

Bases: object

__init__()[source]
reg_resource(cls)[source]

Register a resource with the factory.

Returns the class to allow using it as a decorator.

reg_driver(cls)[source]

Register a driver with the factory.

Returns the class to allow using it as a decorator.

static normalize_config(config)[source]
make_resource(target, resource, name, args)[source]
make_driver(target, driver, name, args)[source]
make_target(name, config, *, env=None)[source]
class_from_string(string: str)[source]
__dict__ = mappingproxy({'__module__': 'labgrid.factory', '__firstlineno__': 7, '__init__': <function TargetFactory.__init__>, 'reg_resource': <function TargetFactory.reg_resource>, 'reg_driver': <function TargetFactory.reg_driver>, '_convert_to_named_list': <staticmethod(<function TargetFactory._convert_to_named_list>)>, 'normalize_config': <staticmethod(<function TargetFactory.normalize_config>)>, 'make_resource': <function TargetFactory.make_resource>, 'make_driver': <function TargetFactory.make_driver>, 'make_target': <function TargetFactory.make_target>, 'class_from_string': <function TargetFactory.class_from_string>, '_insert_into_all': <function TargetFactory._insert_into_all>, '__static_attributes__': ('all_classes', 'drivers', 'resources'), '__dict__': <attribute '__dict__' of 'TargetFactory' objects>, '__weakref__': <attribute '__weakref__' of 'TargetFactory' objects>, '__doc__': None, '__annotations__': {}})
__firstlineno__ = 7
__module__ = 'labgrid.factory'
__static_attributes__ = ('all_classes', 'drivers', 'resources')
__weakref__

list of weak references to the object

labgrid.factory.target_factory = <labgrid.factory.TargetFactory object>

Global TargetFactory instance

This instance is used to register Resource and Driver classes so that Targets can be created automatically from YAML files.

labgrid.logging module

labgrid.logging.basicConfig(**kwargs)[source]
class labgrid.logging.StepFormatter(*args, indent=True, color=None, parent=None, **kwargs)[source]

Bases: object

__init__(*args, indent=True, color=None, parent=None, **kwargs)[source]
format(record)[source]
__dict__ = mappingproxy({'__module__': 'labgrid.logging', '__firstlineno__': 25, '__init__': <function StepFormatter.__init__>, 'format': <function StepFormatter.format>, '__static_attributes__': ('bufs', 'color', 'formatter', 'indent', 'indent_level'), '__dict__': <attribute '__dict__' of 'StepFormatter' objects>, '__weakref__': <attribute '__weakref__' of 'StepFormatter' objects>, '__doc__': None, '__annotations__': {}})
__firstlineno__ = 25
__module__ = 'labgrid.logging'
__static_attributes__ = ('bufs', 'color', 'formatter', 'indent', 'indent_level')
__weakref__

list of weak references to the object

class labgrid.logging.SerialLoggingReporter(bufs=NOTHING, loggers=NOTHING, lastevent=None)[source]

Bases: object

__attrs_post_init__()[source]
vt100_replace_cr_nl(buf)[source]
notify(event)[source]
flush()[source]
__attrs_attrs__ = (Attribute(name='bufs', default=Factory(factory=<class 'dict'>, takes_self=False), validator=<instance_of validator for type <class 'dict'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='bufs'), Attribute(name='loggers', default=Factory(factory=<class 'dict'>, takes_self=False), validator=<instance_of validator for type <class 'dict'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='loggers'), Attribute(name='lastevent', default=None, validator=<optional validator for <instance_of validator for type <class 'labgrid.step.StepEvent'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='lastevent'))
__dict__ = mappingproxy({'__module__': 'labgrid.logging', '__firstlineno__': 54, '__attrs_post_init__': <function SerialLoggingReporter.__attrs_post_init__>, 'vt100_replace_cr_nl': <function SerialLoggingReporter.vt100_replace_cr_nl>, '_create_message': <function SerialLoggingReporter._create_message>, 'notify': <function SerialLoggingReporter.notify>, 'flush': <function SerialLoggingReporter.flush>, '__static_attributes__': ('lastevent',), '__dict__': <attribute '__dict__' of 'SerialLoggingReporter' objects>, '__weakref__': <attribute '__weakref__' of 'SerialLoggingReporter' objects>, '__doc__': None, '__attrs_attrs__': (Attribute(name='bufs', default=Factory(factory=<class 'dict'>, takes_self=False), validator=<instance_of validator for type <class 'dict'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='bufs'), Attribute(name='loggers', default=Factory(factory=<class 'dict'>, takes_self=False), validator=<instance_of validator for type <class 'dict'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='loggers'), Attribute(name='lastevent', default=None, validator=<optional validator for <instance_of validator for type <class 'labgrid.step.StepEvent'>> or None>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='lastevent')), '__ne__': <function __ne__>, '__lt__': <function SerialLoggingReporter.__lt__>, '__le__': <function SerialLoggingReporter.__le__>, '__gt__': <function SerialLoggingReporter.__gt__>, '__ge__': <function SerialLoggingReporter.__ge__>, '__hash__': None, '__replace__': <function SerialLoggingReporter.<lambda>>, '__match_args__': ('bufs', 'loggers', 'lastevent'), '__repr__': <function SerialLoggingReporter.__repr__>, '__eq__': <function SerialLoggingReporter.__eq__>, '__init__': <function SerialLoggingReporter.__init__>, '__annotations__': {}})
__eq__(other)

Method generated by attrs for class SerialLoggingReporter.

__firstlineno__ = 54
__ge__(other)

Method generated by attrs for class SerialLoggingReporter.

__gt__(other)

Method generated by attrs for class SerialLoggingReporter.

__hash__ = None
__init__(bufs=NOTHING, loggers=NOTHING, lastevent=None) → None

Method generated by attrs for class SerialLoggingReporter.

__le__(other)

Method generated by attrs for class SerialLoggingReporter.

__lt__(other)

Method generated by attrs for class SerialLoggingReporter.

__match_args__ = ('bufs', 'loggers', 'lastevent')
__module__ = 'labgrid.logging'
__ne__(other)

Check equality and either forward a NotImplemented or return the result negated.

__replace__(**changes)

Method generated by attrs for class SerialLoggingReporter.

__repr__()

Method generated by attrs for class SerialLoggingReporter.

__static_attributes__ = ('lastevent',)
__weakref__

list of weak references to the object

class labgrid.logging.StepLogger[source]

Bases: object

__attrs_post_init__()[source]
classmethod start(length_limit=None)[source]

starts the StepLogger

classmethod stop()[source]

stops the StepLogger

classmethod get_prefix(event)[source]
static format_arguments(args)[source]
static format_duration(duration)[source]
classmethod format_result(result)[source]
classmethod get_next_indent(event)[source]
classmethod notify(event)[source]
__dict__ = mappingproxy({'__module__': 'labgrid.logging', '__firstlineno__': 129, '_started': False, '_logger': None, '_serial_logger': None, '_length_limit': 100, '__attrs_post_init__': <function StepLogger.__attrs_post_init__>, 'start': <classmethod(<function StepLogger.start>)>, 'stop': <classmethod(<function StepLogger.stop>)>, 'get_prefix': <classmethod(<function StepLogger.get_prefix>)>, 'format_arguments': <staticmethod(<function StepLogger.format_arguments>)>, 'format_duration': <staticmethod(<function StepLogger.format_duration>)>, 'format_result': <classmethod(<function StepLogger.format_result>)>, '_StepLogger__get_message': <classmethod(<function StepLogger.__get_message>)>, 'get_next_indent': <classmethod(<function StepLogger.get_next_indent>)>, 'notify': <classmethod(<function StepLogger.notify>)>, '__static_attributes__': (), '__dict__': <attribute '__dict__' of 'StepLogger' objects>, '__weakref__': <attribute '__weakref__' of 'StepLogger' objects>, '__doc__': None, '__annotations__': {}})
__firstlineno__ = 129
__module__ = 'labgrid.logging'
__static_attributes__ = ()
__weakref__

list of weak references to the object

labgrid.step module

class labgrid.step.Steps[source]

Bases: object

__init__()[source]
get_current()[source]
get_new(title, tag, source, sourceinfo)[source]
push(step)[source]
pop(step)[source]
subscribe(callback)[source]
unsubscribe(callback)[source]
notify(event)[source]
__dict__ = mappingproxy({'__module__': 'labgrid.step', '__firstlineno__': 10, '__init__': <function Steps.__init__>, 'get_current': <function Steps.get_current>, 'get_new': <function Steps.get_new>, 'push': <function Steps.push>, 'pop': <function Steps.pop>, 'subscribe': <function Steps.subscribe>, 'unsubscribe': <function Steps.unsubscribe>, 'notify': <function Steps.notify>, '__static_attributes__': ('_stack', '_subscribers'), '__dict__': <attribute '__dict__' of 'Steps' objects>, '__weakref__': <attribute '__weakref__' of 'Steps' objects>, '__doc__': None, '__annotations__': {}})
__firstlineno__ = 10
__module__ = 'labgrid.step'
__static_attributes__ = ('_stack', '_subscribers')
__weakref__

list of weak references to the object

class labgrid.step.StepEvent(step, data, *, resource=None, stream=False)[source]

Bases: object

__init__(step, data, *, resource=None, stream=False)[source]
__str__()[source]

Return str(self).

__setitem__(k, v)[source]
merge(other)[source]
property age
__dict__ = mappingproxy({'__module__': 'labgrid.step', '__firstlineno__': 50, '__init__': <function StepEvent.__init__>, '__str__': <function StepEvent.__str__>, '__setitem__': <function StepEvent.__setitem__>, '_invalidate': <function StepEvent._invalidate>, 'merge': <function StepEvent.merge>, 'age': <property object>, '__static_attributes__': ('data', 'resource', 'step', 'stream', 'ts'), '__dict__': <attribute '__dict__' of 'StepEvent' objects>, '__weakref__': <attribute '__weakref__' of 'StepEvent' objects>, '__doc__': None, '__annotations__': {}})
__firstlineno__ = 50
__module__ = 'labgrid.step'
__static_attributes__ = ('data', 'resource', 'step', 'stream', 'ts')
__weakref__

list of weak references to the object

class labgrid.step.Step(title, level, tag, source, sourceinfo)[source]

Bases: object

__init__(title, level, tag, source, sourceinfo)[source]
__repr__()[source]

Return repr(self).

property duration
property status
property is_active
property is_done
start()[source]
skip(reason)[source]
stop()[source]
__del__()[source]
__dict__ = mappingproxy({'__module__': 'labgrid.step', '__firstlineno__': 100, '__init__': <function Step.__init__>, '__repr__': <function Step.__repr__>, 'duration': <property object>, 'status': <property object>, 'is_active': <property object>, 'is_done': <property object>, '_notify': <function Step._notify>, 'start': <function Step.start>, 'skip': <function Step.skip>, 'stop': <function Step.stop>, '__del__': <function Step.__del__>, '__static_attributes__': ('_skipped', '_start_ts', '_stop_ts', 'args', 'exception', 'level', 'result', 'source', 'sourceinfo', 'tag', 'title'), '__dict__': <attribute '__dict__' of 'Step' objects>, '__weakref__': <attribute '__weakref__' of 'Step' objects>, '__doc__': None, '__annotations__': {}})
__firstlineno__ = 100
__module__ = 'labgrid.step'
__static_attributes__ = ('_skipped', '_start_ts', '_stop_ts', 'args', 'exception', 'level', 'result', 'source', 'sourceinfo', 'tag', 'title')
__weakref__

list of weak references to the object

labgrid.step.step(*, title=None, args=[], result=False, tag=None)[source]

labgrid.stepreporter module

class labgrid.stepreporter.StepReporter[source]

Bases: object

__init__()[source]
classmethod start()[source]

starts the StepReporter

classmethod stop()[source]

stops the StepReporter

static notify(event)[source]
__dict__ = mappingproxy({'__module__': 'labgrid.stepreporter', '__firstlineno__': 4, '_started': False, '__init__': <function StepReporter.__init__>, 'start': <classmethod(<function StepReporter.start>)>, 'stop': <classmethod(<function StepReporter.stop>)>, 'notify': <staticmethod(<function StepReporter.notify>)>, '__static_attributes__': (), '__dict__': <attribute '__dict__' of 'StepReporter' objects>, '__weakref__': <attribute '__weakref__' of 'StepReporter' objects>, '__doc__': None, '__annotations__': {}})
__firstlineno__ = 4
__module__ = 'labgrid.stepreporter'
__static_attributes__ = ()
__weakref__

list of weak references to the object

labgrid.target module

class labgrid.target.Target(name, env=None)[source]

Bases: object

__attrs_post_init__()[source]
interact(msg)[source]
update_resources()[source]

Iterate over this target’s resources, deactivate any active but unavailable resources and also deactivate any drivers using them. This ensures a consistent binding states for this target.

await_resources(resources, timeout=None, avail=True)[source]

Poll the given resources and wait until they are (un-)available.

Parameters:

  • resources (List) – the resources to poll

  • timeout (float) – optional timeout

  • avail (bool) – optionally wait until the resources are unavailable with avail=False

get_resource(cls, *, name=None, wait_avail=True)[source]

Helper function to get a resource of the target. Returns the first valid resource found, otherwise a NoResourceFoundError is raised.

Arguments: cls – resource-class to return as a resource name – optional name to use as a filter wait_avail – wait for the resource to become available (default True)

get_active_driver(cls, *, name=None, resource=None)[source]

Helper function to get the active driver of the target. Returns the active driver found, otherwise None.

Arguments: cls – driver-class to return as a resource name – optional name to use as a filter resource – optional resource to use as a filter

get_driver(cls, *, name=None, resource=None, activate=True)[source]

Helper function to get a driver of the target. Returns the first valid driver found, otherwise None.

Arguments: cls – driver-class to return as a resource name – optional name to use as a filter resource – optional resource to use as a filter activate – activate the driver (default True)

get_strategy()[source]

Helper function to get the strategy of the target.

Returns the Strategy, if exactly one exists and raises a NoStrategyFoundError otherwise.

__getitem__(key)[source]

Syntactic sugar to access drivers by class (optionally filtered by name).

>>> from labgrid import Target
>>> from labgrid.driver.fake import FakeConsoleDriver
>>>
>>> target = Target('main')
>>> console = FakeConsoleDriver(target, 'console')
>>> target.activate(console)
>>> target[FakeConsoleDriver]
FakeConsoleDriver(target=Target(name='main', env=None), name='console', state=<BindingState.active: 2>, txdelay=0.0)
>>> target[FakeConsoleDriver, 'console']
FakeConsoleDriver(target=Target(name='main', env=None), name='console', state=<BindingState.active: 2>, txdelay=0.0)
set_binding_map(mapping)[source]

Configure the binding name mapping for the next driver only.

bind_resource(resource)[source]

Bind the resource to this target.

bind_driver(client)[source]

Bind the driver to all suppliers (resources and other drivers).

Currently, we only support binding all suppliers at once.

bind(bindable)[source]
activate(client, name=None)[source]

Activate the client by activating all bound suppliers. This may require deactivating other clients.

deactivate(client, name=None)[source]

Recursively deactivate the client’s clients and itself.

This is needed to ensure that no client has an inactive supplier.

Returns the list of all objects that were deactivated

deactivate_all_drivers()[source]

Deactivates all drivers in reversed order they were activated

export()[source]

Export information from drivers.

All drivers are deactivated before being exported, unless their skip_deactivate_on_export property is true.

The Strategy can decide for which driver the export method is called and with which name. Otherwise, all drivers are exported.

cleanup()[source]

Clean up connected drivers and resources in reversed order

__attrs_attrs__ = (Attribute(name='name', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='name'), Attribute(name='env', default=None, validator=None, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='env'))
__dict__ = mappingproxy({'__module__': 'labgrid.target', '__firstlineno__': 18, '__attrs_post_init__': <function Target.__attrs_post_init__>, 'interact': <function Target.interact>, 'update_resources': <function Target.update_resources>, 'await_resources': <function Target.await_resources>, 'get_resource': <function Target.get_resource>, '_get_driver': <function Target._get_driver>, 'get_active_driver': <function Target.get_active_driver>, 'get_driver': <function Target.get_driver>, 'get_strategy': <function Target.get_strategy>, '__getitem__': <function Target.__getitem__>, 'set_binding_map': <function Target.set_binding_map>, 'bind_resource': <function Target.bind_resource>, 'bind_driver': <function Target.bind_driver>, 'bind': <function Target.bind>, 'activate': <function Target.activate>, 'deactivate': <function Target.deactivate>, 'deactivate_all_drivers': <function Target.deactivate_all_drivers>, '_atexit_cleanup': <function Target._atexit_cleanup>, 'export': <function Target.export>, 'cleanup': <function Target.cleanup>, '__static_attributes__': ('_binding_map', '_lookup_table', 'drivers', 'last_update', 'log', 'resources'), '__dict__': <attribute '__dict__' of 'Target' objects>, '__weakref__': <attribute '__weakref__' of 'Target' objects>, '__doc__': None, '__attrs_attrs__': (Attribute(name='name', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='name'), Attribute(name='env', default=None, validator=None, repr=True, eq=True, eq_key=None, order=True, order_key=None, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False, inherited=False, on_setattr=None, alias='env')), '__replace__': <function Target.<lambda>>, '__match_args__': ('name', 'env'), '__repr__': <function Target.__repr__>, '__init__': <function Target.__init__>, '__annotations__': {}})
__firstlineno__ = 18
__init__(name, env=None) → None

Method generated by attrs for class Target.

__match_args__ = ('name', 'env')
__module__ = 'labgrid.target'
__replace__(**changes)

Method generated by attrs for class Target.

__repr__()

Method generated by attrs for class Target.

__static_attributes__ = ('_binding_map', '_lookup_table', 'drivers', 'last_update', 'log', 'resources')
__weakref__

list of weak references to the object