Initial import of MultiRIOTCtrl

Author: miri64

riotctrl/ctrl.py

@@ -131,15 +131,22 @@ def start_term(self, **spawnkwargs):
         The function is blocking until it is ready.
         It waits some time until the terminal is ready and resets the ctrl.
         """
+        self.start_term_wo_sleep(**spawnkwargs)
+        # on many platforms, the termprog needs a short while to be ready
+        time.sleep(self.TERM_STARTED_DELAY)
+
+    def start_term_wo_sleep(self, **spawnkwargs):
+        """Start the terminal.
+
+        The function is blocking until it is ready.
+        It does not wait until the terminal is ready and resets the ctrl.
+        """
         self.stop_term()
 
         term_cmd = self.make_command(['term'])
         self.term = self.TERM_SPAWN_CLASS(term_cmd[0], args=term_cmd[1:],
                                           env=self.env, **spawnkwargs)
 
-        # on many platforms, the termprog needs a short while to be ready
-        time.sleep(self.TERM_STARTED_DELAY)
-
     def _term_pid(self):
         """Terminal pid or None."""
         return getattr(self.term, 'pid', None)

riotctrl/multictrl/__init__.py

@@ -0,0 +1,95 @@
+"""Abstraction for multiple RIOTCtrl objects
+
+Defines a class to abstract access to multiple RIOTCtrls.
+"""
+
+import riotctrl.ctrl
+
+from riotctrl.multictrl.utils import MultiKeyDict
+
+
+# Intentionally do not inherit from TermSpawn so we do not have to rewrite
+# all of `pexpect` ;-)
+# pylint: disable=too-many-ancestors
+class MultiTermSpawn(MultiKeyDict):
+    """Allows for access and control of multiple RIOTCtrl objects
+    """
+    def expect(self, pattern, *args, **kwargs):
+        """
+        mirroring riotctrl.ctrl.TermSpawn.expect()
+        """
+        return {k: v.expect(pattern, *args, **kwargs)
+                for k, v in self.items()}
+
+    def expect_exact(self, pattern, *args, **kwargs):
+        """
+        mirroring riotctrl.ctrl.TermSpawn.expect()
+        """
+        return {k: v.expect_exact(pattern, *args, **kwargs)
+                for k, v in self.items()}
+
+    def sendline(self, *args, **kwargs):
+        """
+        mirroring riotctrl.ctrl.TermSpawn.expect()
+        """
+        return {k: v.sendline(*args, **kwargs)
+                for k, v in self.items()}
+
+
+# pylint: disable=too-many-ancestors
+class MultiRIOTCtrl(MultiKeyDict, riotctrl.ctrl.RIOTCtrl):
+    """Allows for access and control of multiple RIOTCtrl objects
+
+    >>> ctrl = MultiRIOTCtrl({'a': riotctrl.ctrl.RIOTCtrl(env={'BOARD': 'A'}),
+    ...                       'b': riotctrl.ctrl.RIOTCtrl(env={'BOARD': 'B'})})
+    >>> ctrl.board()
+    {'a': 'A', 'b': 'B'}
+    >>> ctrl['a','b'].board()
+    {'a': 'A', 'b': 'B'}
+    >>> ctrl['a'].board()
+    'A'
+    """
+    TERM_SPAWN_CLASS = MultiTermSpawn
+
+    def __init__(self, ctrls=None):
+        super().__init__(ctrls)
+        self.term = None  # type: MultiRIOTCtrl.TERM_SPAWN_CLASS
+
+    @property
+    def application_directory(self):
+        """Absolute path to the containing RIOTCtrls current directory as
+        dictionary."""
+        return {k: ctrl.application_directory for k, ctrl in self.items()}
+
+    def board(self):
+        """Return board type of containing RIOTCtrls as dictionary."""
+        return {k: ctrl.board() for k, ctrl in self.items()}
+
+    def start_term_wo_sleep(self, **spawnkwargs):
+        """Start the terminal (without waiting) for containing ctrls.
+        """
+        res = {}
+        for k, ctrl in self.items():
+            ctrl.start_term_wo_sleep(**spawnkwargs)
+            res[k] = ctrl.term
+        self.term = self.TERM_SPAWN_CLASS(res)
+
+    def make_run(self, targets, *runargs, **runkwargs):
+        """Call make `targets` for containing RIOTctrl contexts.
+
+        It is using `subprocess.run` internally.
+
+        :param targets: make targets
+        :param *runargs: args passed to subprocess.run
+        :param *runkwargs: kwargs passed to subprocess.run
+        :return: subprocess.CompletedProcess object
+        """
+        return {k: ctrl.make_run(targets, *runargs, **runkwargs)
+                for k, ctrl in self.items()}
+
+    def make_command(self, targets):
+        """Dictionary of make command for context of containing RIOTctrls.
+
+        :return: list of command arguments (for example for subprocess)
+        """
+        return {k: ctrl.make_command(targets) for k, ctrl in self.items()}

riotctrl/multictrl/ctrl.py

@@ -0,0 +1,67 @@
+"""
+Shell interaction extenison for riotctrl.multictrl
+
+Defines classes to abstract interactions with RIOT shell commands using
+riotctrl.multictrl.MultiRIOTCtrl objects
+"""
+
+import pexpect
+
+import riotctrl.shell
+import riotctrl.multictrl.ctrl
+
+from riotctrl.multictrl.ctrl import MultiKeyDict
+
+
+class MultiShellInteractionMixin(riotctrl.shell.ShellInteraction):
+    """
+    Mixin class for shell interactions of riotctrl.multictrl.ctrl.MultiRIOTCtrl
+
+    :param ctrl: a MultiRIOTCtrl object
+    """
+    # pylint: disable=super-init-not-called
+    # intentionally do not call super-init, to not cause TypeError
+    def __init__(self, ctrl):
+        # used in __del__, so initialize before raising exception
+        self.term_was_started = False
+        if not isinstance(ctrl, riotctrl.multictrl.ctrl.MultiRIOTCtrl):
+            raise TypeError(
+                "{} not compatible with non multi-RIOTCtrl {}. Use {} instead."
+                .format(type(self), type(ctrl),
+                        riotctrl.shell.ShellInteraction)
+            )
+        self.riotctrl = ctrl
+        self.replwrap = MultiKeyDict()
+
+    def _start_replwrap(self):
+        if not self.replwrap or \
+           (any(key not in self.replwrap for key in self.riotctrl) and
+            any(self.replwrap[key].child != self.riotctrl[key].term
+                for key in self.riotctrl)):
+            for key, ctrl in self.riotctrl.items():
+                # consume potentially shown prompt to be on the same ground as
+                # if it is not shown
+                ctrl.term.expect_exact(["> ", pexpect.TIMEOUT], timeout=.1)
+                # enforce prompt to be shown by sending newline
+                ctrl.term.sendline("")
+                self.replwrap[key] = pexpect.replwrap.REPLWrapper(
+                    ctrl.term, orig_prompt="> ", prompt_change=None,
+                )
+
+    # pylint: disable=arguments-differ
+    def cmd(self, cmd, timeout=-1, async_=False, ctrls=None):
+        """
+        Sends a command via the MultiShellInteractionMixin `riotctrl`s
+
+        :param  cmd: A shell command as string.
+        :param  ctrls: ctrls to run command on
+
+        :return: Output of the command as a string
+        """
+        self._start_replwrap()
+        if ctrls is None:
+            ctrls = self.riotctrl.keys()
+        elif not isinstance(ctrls, (tuple, list)):
+            ctrls = (ctrls,)
+        return {k: rw.run_command(cmd, timeout=timeout, async_=async_)
+                for k, rw in self.replwrap.items() if k in ctrls}

riotctrl/multictrl/shell.py

@@ -0,0 +1,70 @@
+"""Helper utilities.
+"""
+
+import collections.abc
+
+
+class MultiKeyDict(collections.abc.MutableMapping):
+    """Works like a dict, but returns another dict, when used with tuples
+    as a key:
+
+    >>> a = MultiKeyDict({0: 'zero', 1: 'one', 2: 'two'})
+    >>> len(a)
+    3
+    >>> a[0]
+    'zero'
+    >>> a[1]
+    'one'
+    >>> a[2]
+    'two'
+    >>> a[0,1]
+    {0: 'zero', 1: 'one'}
+    >>> a[3] = 'three'
+    >>> a[0,1] = 'foobar'
+    >>> a
+    {0: 'foobar', 1: 'foobar', 2: 'two', 3: 'three'}
+    >>> del a[1,2]
+    >>> a
+    {0: 'foobar', 3: 'three'}
+    >>> del a[0]
+    >>> a
+    {3: 'three'}
+    """
+    def __init__(self, dictionary=None):
+        if dictionary is None:
+            self._dict = {}
+        else:
+            self._dict = dict(dictionary)
+
+    def __str__(self):
+        return str(self._dict)
+
+    def __repr__(self):
+        return str(self)
+
+    def __getitem__(self, key):
+        if isinstance(key, tuple):
+            return type(self)(
+                {k: self._dict[k] for k in key}
+            )
+        return self._dict[key]
+
+    def __setitem__(self, key, value):
+        if isinstance(key, tuple):
+            for k in key:
+                self._dict[k] = value
+        else:
+            self._dict[key] = value
+
+    def __delitem__(self, key):
+        if isinstance(key, tuple):
+            for k in key:
+                del self._dict[k]
+        else:
+            del self._dict[key]
+
+    def __iter__(self):
+        return iter(self._dict)
+
+    def __len__(self):
+        return len(self._dict)

riotctrl/multictrl/utils.py

@@ -9,6 +9,8 @@
 import pexpect
 import pexpect.replwrap
 
+import riotctrl.multictrl.ctrl as multictrl
+
 
 # pylint: disable=R0903
 class ShellInteractionParser(abc.ABC):
@@ -30,9 +32,16 @@ class ShellInteraction():
     :param riotctrl: a RIOTCtrl object
     """
     def __init__(self, riotctrl):
+        # used in __del__, so initialize before raising exception
+        self.term_was_started = False
+        if isinstance(riotctrl, multictrl.MultiRIOTCtrl):
+            raise TypeError(
+                "{} not compatible with multi-RIOTCtrl {}. Use {} instead."
+                .format(type(self), type(riotctrl),
+                        'riotctrl.multictrl.shell.MultiShellInteractionMixin')
+            )
         self.riotctrl = riotctrl
         self.replwrap = None
-        self.term_was_started = False
 
     def __del__(self):
         if self.term_was_started: