Unix events

Asynchronous event observers for Unix/Linux systems. Events watch connection output in real time and fire callbacks when configured patterns are detected.

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

Wait for prompt once

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")