Command

A ConnectionObserver subclass that sends a command string over a connection and parses its output.

Overview

Command extends ConnectionObserver to model device commands — operations like ls, ping, or show version that are initiated by sending a string to the device and conclude when the device’s output matches an expected pattern.

moler.command.Command
  └── moler.connection_observer.ConnectionObserver

A Command object:

1. Sends command_string over the connection when started.
2. Parses the subsequent output in data_received().
3. Calls set_result() once parsing is complete.

Command is an abstract class. You must subclass it, set self.command_string, and implement data_received().

Constructor

Command(connection=None, runner=None)

connection · AbstractMolerConnection · default: "None"

The connection used to send the command string and receive its output.

runner · ConnectionObserverRunner · default: "None"

The runner managing background execution. Resolved from the connection or the global default if None.

After construction the following instance attributes are available:

Attribute	Type	Description
command_string	str | None	The string to be sent to the device. Must be set before start() is called.
cmd_name	str	Derived class name in lower-case underscore form (from Command.observer_name).

Public methods

send_command

cmd.send_command() -> None

Send self.command_string over the connection using connection.sendline(). This is called automatically by the runner when the command is submitted for execution. You generally do not need to call it manually.

Raises AttributeError if connection is None.

is_command

cmd.is_command() -> bool

Returns True. This distinguishes Command instances from plain ConnectionObserver instances at runtime, for example when the CommandScheduler serializes commands on a shared connection.

get_long_desc

cmd.get_long_desc() -> str

Returns a string of the form "Command <module>.<ClassName>(\"<command_string>\", id:...)".

get_short_desc

cmd.get_short_desc() -> str

Returns a string of the form "Command <module>.<ClassName>(id:...)".

Inherited API

All methods from ConnectionObserver are available. The most commonly used ones:

Method	Description
start(timeout)	Begin sending and parsing. Raises NoCommandStringProvided if command_string is None.
await_done(timeout)	Block until the command completes and return its result.
result()	Retrieve the parsed result.
cancel()	Cancel the running command.
set_result(result)	Call from data_received() when parsing is done.
set_exception(exc)	Call from data_received() to signal a failure.
on_timeout()	Override for custom timeout behavior.

Subclassing guide

To implement a command:

1. Subclass Command.
2. Set self.command_string in __init__ (or accept it as a parameter).
3. Implement data_received(data, recv_time) to parse lines and call set_result() when done.

Minimal command subclass

import re
from moler.command import Command

class Uname(Command):
    """Runs 'uname -a' and returns the kernel string."""

    def __init__(self, connection=None, runner=None):
        super().__init__(connection=connection, runner=runner)
        self.command_string = 'uname -a'
        self._result_pattern = re.compile(r'^Linux\s+(\S+)\s+(\S+)')

    def data_received(self, data, recv_time):
        for line in data:
            match = self._result_pattern.search(line)
            if match:
                self.set_result({
                    'hostname': match.group(1),
                    'kernel': match.group(2),
                })
                return

Set self.command_string to end with \n if the device requires it, or rely on sendline() which appends the connection’s configured newline character automatically.

Starting a Command with command_string = None raises NoCommandStringProvided. Always assign the command string in __init__ or before calling start().