Unix events

Unix events are observer objects that run concurrently with commands, scanning connection output line-by-line and firing when a condition is met. Unlike commands, events do not send anything to the connection — they only observe. All Unix events inherit from either GenericUnixLineEvent or GenericUnixTextualEvent.

Events return a list of occurrence dicts. Each occurrence contains line, matched, groups, named_groups, and time. The till_occurs_times parameter controls how many occurrences to collect before the event is considered complete (-1 means run indefinitely).

Event list

Class	Module	What it detects
`Wait4prompt`	`moler.events.unix.wait4prompt`	A specific shell prompt regex
`Wait4prompts`	`moler.events.unix.wait4prompts`	One of multiple prompt regexes, returning state name
`PingResponse`	`moler.events.unix.ping_response`	Successful ICMP reply lines
`PingNoResponse`	`moler.events.unix.ping_no_response`	ICMP no-answer / unreachable lines
`LastLogin`	`moler.events.unix.last_login`	`Last login: ... from ...`
`LastFailedLogin`	`moler.events.unix.last_failed_login`	Last failed login messages
`FailedLoginCounter`	`moler.events.unix.failed_login_counter`	Failed login count lines
`Shutdown`	`moler.events.unix.shutdown`	System shutdown announcement
`WarningDefaultPassword`	`moler.events.unix.warning_default_password`	Default-password warnings
`AdviseToChangeYourPassword`	`moler.events.unix.advise_to_change_your_password`	Password change advisories
`PrivateSystem`	`moler.events.unix.private_system`	Private system / authorised-use notices
`UBootCrtm`	`moler.events.unix.u_boot_crtm`	U-Boot CRTM boot sequence output

Detailed reference

`Wait4prompt` — `moler.events.unix.wait4prompt`

The most commonly used Unix event. Waits for a specific prompt pattern to appear on the connection. Internally delegates to Wait4 with match='any'.

connection

object

required

Moler connection to observe.

prompt

str | re.Pattern

required

Regex pattern to match the expected prompt.

till_occurs_times

int

How many times the prompt must be seen before the event completes. Default -1 (run indefinitely).

runner

object

Runner used to execute the event.

Result list item keys

Key	Type	Description
`line`	`str`	The full line that matched
`matched`	`str`	The substring matched by the prompt regex
`groups`	`tuple`	Unnamed capture groups from the regex
`named_groups`	`dict`	Named capture groups from the regex
`time`	`datetime`	Timestamp when the match occurred

from moler.events.unix.wait4prompt import Wait4prompt

event = Wait4prompt(
    connection=conn,
    prompt=r'host:.*#',
    till_occurs_times=1,
)
result = event()
# result = [
#   {
#     'line': 'host:~ #',
#     'matched': 'host:~ #',
#     'groups': (),
#     'named_groups': {},
#     'time': datetime.datetime(2024, 1, 14, 13, 12, 48, 224929)
#   }
# ]

`Wait4prompts` — `moler.events.unix.wait4prompts`

Like Wait4prompt, but watches for multiple possible prompts simultaneously and returns the prompt’s state name alongside the match. Useful for state-machine-based connection management.

connection

object

required

Moler connection to observe.

prompts

dict

required

Mapping of prompt regex (str or compiled) → state name string. E.g. {r'host:.*#': 'UNIX_ROOT', r'host:.*\$': 'UNIX_USER'}.

till_occurs_times

int

Number of occurrences before completion. Default -1.

Result list item keys

Key	Type	Description
`line`	`str`	Full matched line
`matched`	`str`	Matched substring
`prompt_regex`	`str`	Pattern string of the matching prompt
`state`	`str`	State name from the `prompts` dict
`time`	`datetime`	Timestamp of the match

from moler.events.unix.wait4prompts import Wait4prompts

event = Wait4prompts(
    connection=conn,
    prompts={
        r'admin@switch#': 'JUNIPER_CONFIG',
        r'admin@switch>': 'JUNIPER_CLI',
    },
    till_occurs_times=1,
)
result = event()
print(result[0]['state'])         # 'JUNIPER_CLI'
print(result[0]['prompt_regex'])  # 'admin@switch>'

You can dynamically update the prompt set while the event is running:

event.change_prompts({r'new_prompt:~\$': 'NEW_STATE'})

`PingResponse` — `moler.events.unix.ping_response`

Fires whenever a successful ICMP reply line ("N bytes from ..." ) appears in the output. Useful for monitoring connectivity while a long-running ping command executes.

connection

object

required

Moler connection to observe.

till_occurs_times

int

Number of occurrences. Default -1.

Detection pattern: r'\d+ bytes from.+'

from moler.events.unix.ping_response import PingResponse

event = PingResponse(connection=conn, till_occurs_times=3)
result = event()
# result has 3 entries, each with the icmp_seq reply line
for entry in result:
    print(entry['line'])  # e.g. '64 bytes from 192.168.1.1: icmp_seq=1 ttl=64 time=0.6 ms'

`PingNoResponse` — `moler.events.unix.ping_no_response`

Fires on no answer yet for icmp_seq=N or Destination Host Unreachable lines.

connection

object

required

Moler connection to observe.

till_occurs_times

int

Number of occurrences. Default -1.

Detection pattern: r'(no answer yet for.*)|(.*Destination Host Unreachable)'

`LastLogin` — `moler.events.unix.last_login`

Parses Last login: <date> from <host> lines into structured data. Result list item keys

Key	Type	Description
`time`	`datetime`	Time the event was detected
`host`	`str`	Source host/IP from the login message
`date_raw`	`str`	Raw date string from the message
`date`	`datetime`	Parsed `datetime` of the last login

from moler.events.unix.last_login import LastLogin

event = LastLogin(connection=conn, till_occurs_times=1)
result = event()
print(result[0]['host'])  # '127.0.0.1'
print(result[0]['date'])  # datetime(2018, 6, 12, 8, 54, 44)

`Shutdown` — `moler.events.unix.shutdown`

Detects system shutdown announcements matching system is going down for <reason> at <time>. Result list item keys: line, matched, groups (tuple of (reason, time_string)), named_groups, time.

from moler.events.unix.shutdown import Shutdown

event = Shutdown(connection=conn, till_occurs_times=1)
result = event()
reason, at_time = result[0]['groups']
print(f"System going down for {reason} at {at_time}")
# System going down for reboot at Tue 2024-03-19 12:15:16 CET!

Using events with callbacks

Events can be used either synchronously (blocking until till_occurs_times is reached) or asynchronously with callbacks:

from moler.events.unix.ping_response import PingResponse
from moler.events.unix.ping_no_response import PingNoResponse

responses = []
no_responses = []

def on_response(event_data):
    responses.append(event_data)
    print(f"Ping OK: {event_data['line']}")

def on_no_response(event_data):
    no_responses.append(event_data)
    print(f"Ping FAIL: {event_data['line']}")

# Start both events in background
ok_event = PingResponse(connection=conn)
ok_event.add_event_occurred_callback(on_response)
ok_event.start()

fail_event = PingNoResponse(connection=conn)
fail_event.add_event_occurred_callback(on_no_response)
fail_event.start()

# ... run ping command ...

ok_event.cancel()
fail_event.cancel()
print(f"Got {len(responses)} replies, {len(no_responses)} failures")

Core API

Built-in Commands

Built-in Events

Event list

Detailed reference

`Wait4prompt` — `moler.events.unix.wait4prompt`

`Wait4prompts` — `moler.events.unix.wait4prompts`

`PingResponse` — `moler.events.unix.ping_response`

`PingNoResponse` — `moler.events.unix.ping_no_response`

`LastLogin` — `moler.events.unix.last_login`

`Shutdown` — `moler.events.unix.shutdown`

Using events with callbacks

Build docs developers (and LLMs) love

Core API

Built-in Commands

Built-in Events

Documentation Index

​Event list

​Detailed reference

​Wait4prompt — moler.events.unix.wait4prompt

​Wait4prompts — moler.events.unix.wait4prompts

​PingResponse — moler.events.unix.ping_response

​PingNoResponse — moler.events.unix.ping_no_response

​LastLogin — moler.events.unix.last_login

​Shutdown — moler.events.unix.shutdown

​Using events with callbacks

Build docs developers (and LLMs) love

Event list

Detailed reference

`Wait4prompt` — `moler.events.unix.wait4prompt`

`Wait4prompts` — `moler.events.unix.wait4prompts`

`PingResponse` — `moler.events.unix.ping_response`

`PingNoResponse` — `moler.events.unix.ping_no_response`

`LastLogin` — `moler.events.unix.last_login`

`Shutdown` — `moler.events.unix.shutdown`

Using events with callbacks