Creating custom commands

Subclass CommandTextualGeneric (or GenericUnixCommand) to build your own parsing commands following the same patterns as built-in commands.

All built-in Moler commands follow the same construction pattern, which makes creating new ones straightforward. This guide walks through the pattern using the actual source code of ping.py and ps.py as reference, then shows a complete minimal example.

The command class hierarchy

ConnectionObserver
  └── Command
        └── CommandTextualGeneric       (moler/cmd/commandtextualgeneric.py)
              └── GenericUnixCommand    (moler/cmd/unix/genericunix.py)
                    └── Ping, Ps, Ls, ...  (moler/cmd/unix/*.py)

For Unix/Linux commands, subclass GenericUnixCommand. For non-Unix textual commands, subclass CommandTextualGeneric directly.

Key methods to implement

Every command must implement two methods:

Method	Purpose
build_command_string()	Returns the shell command string to send to the device
on_new_line(line, is_full_line)	Called for each line of output; put your parsing logic here

There are also optional lifecycle callbacks:

Method	When called
on_done()	Just before the command finishes (success or failure)
on_success()	Just before the command finishes successfully
on_failure()	Just before the command finishes with a failure

How build_command_string() works

CommandTextualGeneric has a command_string property backed by build_command_string(). The property is called lazily — the first time the command string is needed. Here is the pattern from ping.py:

def build_command_string(self):
    """
    Builds command string from parameters passed to object.
    :return: String representation of command to send over connection to device.
    """
    if ":" in self.destination:
        cmd = f"ping6 {self.destination}"
    else:
        cmd = f"ping {self.destination}"
    if self.options:
        cmd = f"{cmd} {self.options}"
    return cmd

And from ps.py:

def build_command_string(self):
    """
    Builds string with command.
    :return: String with command.
    """
    cmd = "ps"
    if self.options:
        cmd = f"{cmd} {self.options}"
    return cmd

The method simply constructs the shell command string from instance attributes. Moler sends this string over the connection when the command starts.

How on_new_line() works

on_new_line(line, is_full_line) is called by the framework for each line of output received after the command echo. The line argument has newline characters stripped. is_full_line is True if the line ended with a newline character (i.e., it is a complete line), False if it is a partial line (still being received).

The base class on_new_line() in CommandTextualGeneric checks whether the line matches the expected shell prompt. If it does, the command is marked as done and self.current_ret is stored as the result. You must call super().on_new_line(line, is_full_line) at the end of your override so the prompt detection still happens.

The ParsingDone exception is the standard early-exit mechanism — raise it inside a parse helper when a line has been fully handled to skip remaining parse methods:

# From ping.py — pattern for on_new_line
def on_new_line(self, line, is_full_line):
    if is_full_line:
        try:
            self._parse_trans_recv_loss_time(line)
            self._parse_min_avg_max_mdev_unit_time(line)
        except ParsingDone:
            pass  # line has been fully parsed by one of the above methods
    return super(Ping, self).on_new_line(line, is_full_line)

Each _parse_* helper checks a regex against the line. If the regex matches, it populates self.current_ret with extracted values and raises ParsingDone.

The current_ret pattern

self.current_ret is a dict (or list, for commands like ps) that accumulates parsed data as output lines arrive. When the command detects the prompt and marks itself as done, the framework calls self.set_result(self.current_ret) which stores the value and makes it available via command() or command.await_done().

A complete minimal example

Here is a custom command for uptime that extracts the system load averages:

import re
from moler.cmd.unix.genericunix import GenericUnixCommand
from moler.exceptions import ParsingDone


class Uptime(GenericUnixCommand):
    """
    Parse uptime output to extract load averages.

    Example output:
     14:32:15 up 10 days,  3:21,  2 users,  load average: 0.15, 0.12, 0.09
    """

    def __init__(self, connection, prompt=None, newline_chars=None, runner=None):
        super(Uptime, self).__init__(
            connection=connection,
            prompt=prompt,
            newline_chars=newline_chars,
            runner=runner,
        )

    def build_command_string(self):
        return "uptime"

    # load average: 0.15, 0.12, 0.09
    _re_load = re.compile(
        r"load average:\s+(?P<load1>[\d.]+),\s+(?P<load5>[\d.]+),\s+(?P<load15>[\d.]+)"
    )

    def on_new_line(self, line, is_full_line):
        if is_full_line:
            try:
                self._parse_load_averages(line)
            except ParsingDone:
                pass
        return super(Uptime, self).on_new_line(line, is_full_line)

    def _parse_load_averages(self, line):
        if self._regex_helper.search_compiled(Uptime._re_load, line):
            self.current_ret['load_1min']  = float(self._regex_helper.group('load1'))
            self.current_ret['load_5min']  = float(self._regex_helper.group('load5'))
            self.current_ret['load_15min'] = float(self._regex_helper.group('load15'))
            raise ParsingDone


# At the bottom of each command module, Moler expects these test fixtures:
COMMAND_OUTPUT = """ 14:32:15 up 10 days,  3:21,  2 users,  load average: 0.15, 0.12, 0.09
user@host:~$"""

COMMAND_KWARGS = {}

COMMAND_RESULT = {
    'load_1min':  0.15,
    'load_5min':  0.12,
    'load_15min': 0.09,
}

self._regex_helper is provided by CommandTextualGeneric. It wraps re search operations and stores the last match object, so you can call self._regex_helper.group('name') without re-running the regex.

Using the command with a device

Via device (get_cmd)

If the command file is in a package that the device scans (e.g., moler/cmd/unix/), you can retrieve it by name:

from moler.config import load_config
from moler.device import DeviceFactory

load_config(config='my_devices.yml')
my_unix = DeviceFactory.get_device(name='MyMachine')

# Device discovers commands in moler.cmd.unix automatically
uptime_cmd = my_unix.get_cmd(cmd_name="uptime")
result = uptime_cmd()
print("1-min load:", result['load_1min'])

Direct instantiation

You can always instantiate the command directly with a connection:

from moler.connection_factory import get_connection
from myproject.commands.uptime import Uptime

terminal = get_connection(io_type='terminal', variant='threaded')
with terminal.open():
    uptime_cmd = Uptime(connection=terminal.moler_connection)
    result = uptime_cmd()
    print("1-min load:", result['load_1min'])

Adding failure detection

GenericUnixCommand already sets self.re_fail to a pattern matching common Unix error messages (not found, No such file or directory, etc.). If your command has additional failure indicators, add them:

def __init__(self, connection, prompt=None, newline_chars=None, runner=None):
    super(Uptime, self).__init__(
        connection=connection, prompt=prompt,
        newline_chars=newline_chars, runner=runner,
    )
    # add a command-specific failure pattern
    self.add_failure_indication(r"uptime: command not found")

When any line in the output matches re_fail, the command raises CommandFailure and marks itself as done with an exception.

Handling commands that produce no result

Some commands (like touch or rm) produce no parseable output. Set self.ret_required = False so the command does not wait for a non-empty current_ret before completing:

def __init__(self, connection, ...):
    super().__init__(connection=connection, ...)
    self.ret_required = False  # command completes on prompt alone

Real source reference

The patterns above come directly from these source files:

• moler/cmd/unix/ping.py — regex parsing with multiple parse helpers and ParsingDone
• moler/cmd/unix/ps.py — list result (self.current_ret = []), dynamic header parsing
• moler/cmd/unix/ls.py — structured dict result with helper methods (get_files(), get_dirs())
• moler/cmd/commandtextualgeneric.py — base class with on_new_line, build_command_string, lifecycle callbacks