Learn how to record every EtherCAT network and node status value continuously, so that when a network fails unexpectedly the final records show what happened. The application spreads the NetworkStatus and per-node status addresses over as many Recorders as needed, keeps recording until the network leaves OPERATIONAL (or a key is pressed), then prints a change log and the newest records, and writes the last seconds of every value to a CSV file and the network firmware log to a text file.
""" Record EtherCAT network and node status continuously, to explain a network failure.
"The network died and we do not know why." This sample keeps Recorders running on
every dynamic network value (RSIControllerAddressType NETWORK_STATE and
NETWORK_STATUS_*), every sync group's status (NETWORK_SYNC_GROUP_*) and every node's
status (RSINetworkNodeAddressType). When the network leaves OPERATIONAL it keeps
recording for a few seconds so the whole shutdown is captured, then prints:
- a timeline: every value that changed, in order, timed from the first change
(normally the moment the cable was pulled),
- a summary: which nodes stopped responding, what the sync groups saw, and when
the network state changed,
- the newest records of the values that changed, side by side.
The last seconds of every recorded value are written to a time-stamped CSV file, and the
network firmware's log to a text file, for further study.
A Recorder holds at most 32 addresses, so the values are spread over as many
Recorders as needed. Every Recorder also records the controller's sample counter so
the records of different Recorders can be lined up. A Recorder's buffer holds 512
records of 32 values, so the drain interval is derived from the sample rate (about
128 ms at 1 kHz, 32 ms at 4 kHz); the buffer is never resized, since that reconfigures
the controller and invalidates every RapidCode object in every process. Every recorded
value costs one call into RapidCode when it is drained (see helpers.recorder_value_reader),
so a large network at a high sample rate may need RECORD_PERIOD_SAMPLES raised: the
sample says so if a Recorder's buffer fills before it is drained.
The sample runs until the network leaves OPERATIONAL (it may run for days), so it
keeps a bounded history and a bounded log of changes: memory does not grow with time.
It also stops when a key is pressed (Enter on Linux/macOS), or after the number of
seconds given on the command line: python recorder-network-status.py 30
Requirements: an OPERATIONAL EtherCAT network and enough Recorders on the
MotionController (RecorderCountSet). If either is missing the sample prints how to
fix that and exits successfully, so automated runs without hardware stay green.
"""
import csv
import math
import sys
import time
from collections import deque, namedtuple
from operator import itemgetter
from _imports import RapidCode, helpers, constants
print("⬤ Recorder: Network and Node Status")
RECORD_PERIOD_SAMPLES = 1
RETRIEVE_INTERVAL_MAXIMUM_SECONDS = 0.25
BUFFER_FILL_PER_RETRIEVE = 0.25
RECORDS_PER_RETRIEVE_MAXIMUM = 1024
HISTORY_SECONDS = 10.0
NOT_OPERATIONAL_GRACE_SECONDS = 4.0
NON_INTERACTIVE_RECORD_SECONDS = 5.0
CHANGES_TO_KEEP = 10000
CHANGES_TO_PRINT = 60
FINAL_RECORDS_TO_PRINT = 8
SAMPLE_COUNTER_MODULUS = 2 ** 32
ADDRESSES_PER_RECORDER_MAXIMUM = 32
NETWORK_LOG_FILE_PREFIX = "network-log-"
RECORDS_FILE_PREFIX = "network-records-"
def controller_address_type(name):
return getattr(RapidCode, "RSIControllerAddressType_RSIControllerAddressType" + name)
def node_address_type(name):
return getattr(RapidCode, "RSINetworkNodeAddressType_RSINetworkNodeAddressType" + name)
def enum_name(enum):
"""Formatter that prints a value by its RSI enum member name, e.g. OPERATIONAL."""
return lambda value: helpers.enum_to_name(value, enum)
def node_list(mask):
"""Formatter for a node bitmask, e.g. Node1,4."""
nodes = [str(index) for index in range(64) if mask & (1 << index)]
return "Node" + ",".join(nodes) if nodes else "none"
class Value(namedtuple("Value", "label address_type continuous text")):
def __new__(cls, label, address_type, continuous=False, text=str):
return super().__new__(cls, label, address_type, continuous, text)
NETWORK_VALUES = [
Value("State", controller_address_type("NETWORK_STATE"), text=enum_name("RSINetworkState")),
Value("LastStartError", controller_address_type("NETWORK_LAST_START_ERROR"), text=enum_name("RSINetworkStartError")),
Value("ActiveNodeCount", controller_address_type("NETWORK_STATUS_ACTIVE_NODE_COUNT")),
Value("AlStatus", controller_address_type("NETWORK_STATUS_AL_STATUS"), text=helpers.decode_al_status_ored),
Value("MissedFrames", controller_address_type("NETWORK_STATUS_MISSED_CYCLIC_FRAME_COUNT")),
Value("FramePeriodUs", controller_address_type("NETWORK_STATUS_CYCLIC_FRAME_PERIOD_US"), continuous=True),
Value("SyncErrorNs", controller_address_type("NETWORK_STATUS_SYNCHRONIZATION_ERROR_NS"), continuous=True),
Value("EoeHostToNode", controller_address_type("NETWORK_STATUS_EOE_HOST_TO_NODE_FRAME_COUNT")),
Value("EoeNodeToHost", controller_address_type("NETWORK_STATUS_EOE_NODE_TO_HOST_FRAME_COUNT")),
]
SYNC_GROUP_VALUES = [
Value("State", controller_address_type("NETWORK_SYNC_GROUP_STATE"), text=enum_name("RSINetworkSyncGroupState")),
Value("ExpectedWkc", controller_address_type("NETWORK_SYNC_GROUP_EXPECTED_WKC")),
Value("ActualWkc", controller_address_type("NETWORK_SYNC_GROUP_ACTUAL_WKC")),
Value("DataValid", controller_address_type("NETWORK_SYNC_GROUP_DATA_VALID")),
Value("ToleratedWkcCount", controller_address_type("NETWORK_SYNC_GROUP_TOLERATED_WKC_COUNT")),
Value("OperationalNodeCount", controller_address_type("NETWORK_SYNC_GROUP_OPERATIONAL_NODE_COUNT")),
Value("NonOperationalNodes", controller_address_type("NETWORK_SYNC_GROUP_NON_OPERATIONAL_NODE_MASK"), text=node_list),
]
NODE_VALUES = [
Value("AlStatus", node_address_type("AL_STATUS"), text=helpers.decode_al_status),
Value("AlStatusCode", node_address_type("AL_STATUS_CODE")),
Value("CoeEmergency", node_address_type("COE_EMERGENCY_MESSAGE")),
Value("CoeEmergencyCounter", node_address_type("COE_EMERGENCY_MESSAGE_NETWORK_COUNTER")),
Value("EoeHostToNode", node_address_type("EOE_HOST_TO_NODE_FRAME_COUNT")),
Value("EoeNodeToHost", node_address_type("EOE_NODE_TO_HOST_FRAME_COUNT")),
Value("Present", node_address_type("PRESENT")),
Value("InitState", node_address_type("INITIALIZATION_STATE"), text=enum_name("RSINetworkNodeInitializationState")),
]
SAMPLE_COUNTER = Value("SampleCounter", controller_address_type("SAMPLE_COUNTER"), continuous=True)
Point = namedtuple("Point", "label address reader continuous text")
def resolve(controller, prefix, value, rsi_object, *address_args):
"""Ask a MotionController or RapidCodeNetworkNode for the value's address and data type."""
address = rsi_object.AddressGet(value.address_type, *address_args)
data_type = rsi_object.AddressDataTypeGet(value.address_type)
return Point(f"{prefix}.{value.label}", address, helpers.recorder_value_reader(controller, data_type),
value.continuous, value.text)
def gather_points(controller):
"""Every network value, every sync group's status, and every node's status."""
points = [resolve(controller, "Network", value, controller) for value in NETWORK_VALUES]
for group_id in range(controller.NetworkSyncGroupCountGet()):
points += [resolve(controller, f"SyncGroup{group_id}", value, controller, group_id) for value in SYNC_GROUP_VALUES]
for node_index in range(controller.NetworkNodeCountGet()):
node = controller.NetworkNodeGet(node_index)
helpers.check_errors(node)
points += [resolve(controller, f"Node{node_index}", value, node) for value in NODE_VALUES]
return points
def recorders_needed(sync_group_count, node_count) -> int:
"""Each Recorder also records the sample counter, so it holds one fewer value."""
values = (len(NETWORK_VALUES) + sync_group_count * len(SYNC_GROUP_VALUES)
+ node_count * len(NODE_VALUES))
return math.ceil(values / (ADDRESSES_PER_RECORDER_MAXIMUM - 1))
Change = namedtuple("Change", "sample point old new")
def samples_between(earlier, later) -> int:
"""later - earlier, allowing for the 32-bit sample counter wrapping around.
Two samples are ordered correctly only while they are less than half the modulus apart
(2^31 samples: 24.8 days at 1 kHz, 6.2 days at 4 kHz), so the timeline is in order as long
as the retained changes span less than that; with CHANGES_TO_KEEP bounding the log, older
changes are normally long gone by then."""
half = SAMPLE_COUNTER_MODULUS // 2
return (later - earlier + half) % SAMPLE_COUNTER_MODULUS - half
class Recording:
"""One Recorder and the records drained from it.
Data index 0 is always the controller's sample counter, so records from different
Recorders can be matched up by sample. Records are kept in a bounded history, and every
change of a (non-continuous) value is appended to the shared, bounded change log as the
records are drained, so a change hours before the failure is still reported."""
def __init__(self, controller, recorder_number, points, history_length, changes):
self.controller = controller
self.number = recorder_number
self.points = [resolve(controller, "Controller", SAMPLE_COUNTER, controller)] + points
self.columns = list(enumerate(point.reader for point in self.points))
self.watched = [(index, point) for index, point in enumerate(self.points) if not point.continuous]
self.watched_values = itemgetter(*[index for index, _ in self.watched]) if self.watched else (lambda record: None)
self.history = deque(maxlen=history_length)
self.changes = changes
self.total_records = 0
self.overflows = 0
def configure(self):
controller, number = self.controller, self.number
if controller.RecorderEnabledGet(number):
controller.RecorderStop(number)
controller.RecorderReset(number)
controller.RecorderPeriodSet(number, RECORD_PERIOD_SAMPLES)
controller.RecorderCircularBufferSet(number, True)
controller.RecorderDataCountSet(number, len(self.points))
for index, point in enumerate(self.points):
controller.RecorderDataAddressSet(number, index, point.address)
self.capacity = controller.RecorderRecordMaxCountGet(number)
def start(self):
self.controller.RecorderStart(self.number)
def stop(self):
self.controller.RecorderStop(self.number)
self.controller.RecorderReset(self.number)
def drain(self):
"""Retrieve every available record into the history."""
controller, number = self.controller, self.number
available = controller.RecorderRecordCountGet(number)
if available >= self.capacity:
self.overflows += 1
print(f"Recorder {number} filled its buffer of {self.capacity} records before it was drained "
f"(overflow {self.overflows}): its records were lost and it was restarted. "
"Lower BUFFER_FILL_PER_RETRIEVE or raise RECORD_PERIOD_SAMPLES.")
self.configure()
self.start()
self.history.clear()
return
columns, watched_values = self.columns, self.watched_values
previous = self.history[-1] if self.history else None
while available > 0:
retrieved = controller.RecorderRecordDataRetrieveBulk(number, min(available, RECORDS_PER_RETRIEVE_MAXIMUM))
if retrieved <= 0:
break
for record_index in range(retrieved):
record = [read(number, record_index, data_index) for data_index, read in columns]
if previous is not None and watched_values(record) != watched_values(previous):
self.note_changes(previous, record)
self.history.append(record)
previous = record
available -= retrieved
self.total_records += retrieved
def note_changes(self, previous, record):
for index, point in self.watched:
if record[index] != previous[index]:
self.changes.append(Change(record[0], point, previous[index], record[index]))
def create_recordings(controller, points, changes):
"""Spread the points over as many Recorders as needed; they share one change log."""
points_per_recorder = ADDRESSES_PER_RECORDER_MAXIMUM - 1
history_length = int(HISTORY_SECONDS * controller.SampleRateGet() / RECORD_PERIOD_SAMPLES)
return [Recording(controller, recorder_number, points[start:start + points_per_recorder], history_length, changes)
for recorder_number, start in enumerate(range(0, len(points), points_per_recorder))]
def retrieve_interval(controller, recordings) -> float:
"""Seconds between drains: the fullest Recorder's buffer is at most BUFFER_FILL_PER_RETRIEVE full."""
records_per_second = controller.SampleRateGet() / RECORD_PERIOD_SAMPLES
capacity = min(recording.capacity for recording in recordings)
return min(RETRIEVE_INTERVAL_MAXIMUM_SECONDS, capacity * BUFFER_FILL_PER_RETRIEVE / records_per_second)
def network_is_operational(controller) -> bool:
return controller.NetworkStateGet() == RapidCode.RSINetworkState_RSINetworkStateOPERATIONAL
def network_state_name(controller) -> str:
return helpers.enum_to_name(controller.NetworkStateGet(), "RSINetworkState")
def record_until_stopped(controller, recordings, record_seconds):
"""Record continuously; stop on a key press (record_seconds None), after record_seconds,
or a few seconds after the network leaves OPERATIONAL, so the whole shutdown is captured."""
interactive = record_seconds is None
interval = retrieve_interval(controller, recordings)
for recording in recordings:
recording.start()
started = time.monotonic()
not_operational_since = None
print(f"Recording {sum(len(recording.points) - 1 for recording in recordings)} values on "
f"{len(recordings)} Recorder(s), every {RECORD_PERIOD_SAMPLES} sample(s) at "
f"{controller.SampleRateGet():g} Hz, draining every {interval * 1000:.0f} ms, "
"until the network leaves OPERATIONAL.")
print("Press a key to stop (Enter on Linux terminals)." if interactive
else f"Recording for {record_seconds} seconds.")
next_drain = time.monotonic()
while True:
next_drain += interval
time.sleep(max(0.0, next_drain - time.monotonic()))
for recording in recordings:
recording.drain()
if interactive and helpers.key_pressed():
print("Key pressed, stopping.")
break
if not interactive and time.monotonic() - started >= record_seconds:
break
if network_is_operational(controller):
not_operational_since = None
elif not_operational_since is None:
not_operational_since = time.monotonic()
print(f"Network left OPERATIONAL (now {network_state_name(controller)}), "
f"recording the shutdown for {NOT_OPERATIONAL_GRACE_SECONDS} more seconds.")
elif time.monotonic() - not_operational_since >= NOT_OPERATIONAL_GRACE_SECONDS:
break
for recording in recordings:
recording.drain()
recording.stop()
Records = namedtuple("Records", "points samples values newest")
def merge_records(recordings):
"""Line up the records of every Recorder by sample counter, in recording order.
With RECORD_PERIOD_SAMPLES above 1 each Recorder records on its own phase of the period
(the sample its RecorderStart landed in), so each Recorder's counters are shifted down by
that phase before matching and the samples reported are the first Recorder's. Only samples
present in every Recorder's history are kept, so if one Recorder was restarted after an
overflow or drifted, samples (and every values list) can be empty; the report then falls
back to the change log and newest."""
points = [point for recording in recordings for point in recording.points[1:]]
phases = [recording.history[0][0] % RECORD_PERIOD_SAMPLES if recording.history else 0 for recording in recordings]
by_sample = [{record[0] - phase: record[1:] for record in recording.history}
for recording, phase in zip(recordings, phases)]
keys = [key for key in by_sample[0] if all(key in records for records in by_sample)]
samples = [key + phases[0] for key in keys]
values = {}
newest = {}
for records, recording in zip(by_sample, recordings):
for column, point in enumerate(recording.points[1:]):
values[point.label] = [records[key][column] for key in keys]
if recording.history:
newest[point.label] = recording.history[-1][1 + column]
return Records(points, samples, values, newest)
class Report:
"""The narrative of what changed, timed in seconds from the first change."""
def __init__(self, records, changes, node_names, sample_rate):
self.records = records
self.changes = changes
self.node_names = node_names
self.sample_rate = sample_rate
self.first_sample = changes[0].sample
self.latest = {}
for change in changes:
self.latest[change.point.label] = change
def seconds(self, sample) -> str:
return f"t {samples_between(self.first_sample, sample) / self.sample_rate:+.3f} s"
def transition(self, change) -> str:
return f"{change.point.text(change.old)} -> {change.point.text(change.new)}"
def print_timeline(self):
shown = self.changes[-CHANGES_TO_PRINT:]
print(f"\nTimeline: {len(self.changes)} change(s) while recording"
+ (f", the newest {len(shown)} shown" if len(shown) < len(self.changes) else "")
+ f". t = 0 is sample {self.first_sample}, the first change.")
for change in shown:
print(f" {self.seconds(change.sample):>14} {change.point.label:<32} {self.transition(change)}")
def print_summary(self):
print("\nSummary:")
self.print_node_summary()
self.print_sync_group_summary()
self.print_network_summary()
def print_node_summary(self):
node_count = len(self.node_names)
lost = [index for index in range(node_count)
if f"Node{index}.Present" in self.latest and self.latest[f"Node{index}.Present"].new == 0]
for index in lost:
change = self.latest[f"Node{index}.Present"]
print(f" Node{index} ({self.node_names[index]}) stopped responding at {self.seconds(change.sample)}.")
if lost and lost == list(range(lost[0], node_count)) and lost[0] > 0:
print(f" Every node from Node{lost[0]} on is gone while Node{lost[0] - 1} still answers: "
f"the break is between Node{lost[0] - 1} and Node{lost[0]}.")
if not lost:
print(" No node stopped responding (Present stayed 1 for every node).")
for index in range(node_count):
change = self.latest.get(f"Node{index}.AlStatus")
if change:
print(f" Node{index} ({self.node_names[index]}) AL state {self.transition(change)} "
f"at {self.seconds(change.sample)}.")
def print_sync_group_summary(self):
for label, change in sorted(self.latest.items()):
if label.endswith(".NonOperationalNodes"):
group = label.split(".")[0]
expected = self.records.newest.get(f"{group}.ExpectedWkc", "?")
actual = self.records.newest.get(f"{group}.ActualWkc", "?")
print(f" {group}: non-operational nodes {change.point.text(change.new)} at {self.seconds(change.sample)}; "
f"working counter {actual} of {expected} expected.")
def print_network_summary(self):
states = [change for change in self.changes if change.point.label == "Network.State"]
if states:
print(" Network state: " + ", ".join(f"{self.transition(change)} at {self.seconds(change.sample)}"
for change in states) + ".")
missed = self.latest.get("Network.MissedFrames")
if missed and missed.new > 0:
print(f" Missed cyclic frames rose to {missed.new}.")
def print_newest_records(self):
"""The newest records, side by side, for the values that changed (the rest would be noise)."""
if not self.records.samples:
print("\nThe Recorders' histories share no sample counter (a Recorder was restarted after "
"an overflow), so the newest records cannot be shown side by side.")
return
newest = slice(-FINAL_RECORDS_TO_PRINT, None)
print(f"\nNewest {len(self.records.samples[newest])} records of the values that changed (one column per record):")
print(f"{'sample':>32}: " + " ".join(f"{sample:>14}" for sample in self.records.samples[newest]))
for point in self.records.points:
if point.label in self.latest:
print(f"{point.label:>32}: " + " ".join(f"{point.text(value):>14}"
for value in self.records.values[point.label][newest]))
def write_records_csv(records, recordings):
"""Write the last HISTORY_SECONDS of every recorded value to a time-stamped CSV file:
one row per record, a sample counter column, then one column per value. When the
Recorders' histories cannot be lined up, each Recorder's history goes to its own file."""
stamp = time.strftime("%Y%m%d-%H%M%S")
if records.samples:
file_name = f"{RECORDS_FILE_PREFIX}{stamp}.csv"
with open(file_name, "w", newline="", encoding="utf-8") as csv_file:
writer = csv.writer(csv_file)
writer.writerow(["SampleCounter"] + [point.label for point in records.points])
for row, sample in enumerate(records.samples):
writer.writerow([sample] + [records.values[point.label][row] for point in records.points])
print(f"Wrote {len(records.samples)} records of {len(records.points)} values to {file_name}.")
return
for recording in recordings:
if not recording.history:
continue
file_name = f"{RECORDS_FILE_PREFIX}{stamp}-recorder{recording.number}.csv"
with open(file_name, "w", newline="", encoding="utf-8") as csv_file:
writer = csv.writer(csv_file)
writer.writerow([point.label for point in recording.points])
writer.writerows(recording.history)
print(f"Wrote Recorder {recording.number}: {len(recording.history)} records to {file_name}.")
def write_network_log(controller):
"""Write the network firmware's log to a time-stamped file; it may hold further clues.
The firmware publishes its log when the network shuts down, so there is nothing to
write while the network is still running."""
count = controller.NetworkLogMessageCountGet()
if count == 0:
return
file_name = NETWORK_LOG_FILE_PREFIX + time.strftime("%Y%m%d-%H%M%S") + ".txt"
with open(file_name, "w", encoding="utf-8") as log_file:
for index in range(count):
log_file.write(controller.NetworkLogMessageGet(index) + "\n")
print(f"Wrote {count} network firmware log messages to {file_name}.")
def print_results(controller, recordings, changes, node_names):
records = merge_records(recordings)
changes = sorted(changes, key=lambda change: samples_between(changes[0].sample, change.sample))
total = sum(recording.total_records for recording in recordings)
print(f"\nRetrieved {total} records; {len(records.samples)} lined-up records kept in history. "
f"Network state now: {network_state_name(controller)}.")
if changes:
report = Report(records, changes, node_names, controller.SampleRateGet())
report.print_timeline()
report.print_summary()
report.print_newest_records()
else:
print("No status value changed while recording.")
print()
write_records_csv(records, recordings)
write_network_log(controller)
exit_code = constants.EXIT_FAILURE
if len(sys.argv) > 1:
record_seconds = float(sys.argv[1])
else:
record_seconds = None if helpers.stdin_is_interactive else NON_INTERACTIVE_RECORD_SECONDS
creation_params: RapidCode.CreationParameters = helpers.get_creation_parameters()
controller: RapidCode.MotionController = RapidCode.MotionController.Create(creation_params)
try:
helpers.check_errors(controller)
node_count = controller.NetworkNodeCountGet()
needed = recorders_needed(controller.NetworkSyncGroupCountGet(), node_count)
if not network_is_operational(controller):
print(f"The network state is {network_state_name(controller)}, but this sample needs an OPERATIONAL "
"EtherCAT network so it has network and node status to record.\n"
"Start the network first (RapidSetup, or MotionController.NetworkStart()) "
"and run this sample again. Exiting successfully so automated test runs "
"without EtherCAT hardware stay green.")
exit_code = constants.EXIT_SUCCESS
elif controller.RecorderCountGet() < needed:
print(f"This controller has {controller.RecorderCountGet()} Recorder(s) configured, but recording the "
f"status of {node_count} node(s) needs {needed} "
f"(a Recorder holds at most {ADDRESSES_PER_RECORDER_MAXIMUM} addresses).\n"
f"Configure the recorder count before other objects are created (call "
f"MotionController.RecorderCountSet({needed}) early in your application, or set the "
"recorder count in RapidSetup) and run this sample again. "
"Exiting successfully so automated test runs stay green.")
exit_code = constants.EXIT_SUCCESS
else:
node_names = [controller.NetworkNodeGet(index).NameGet() for index in range(node_count)]
print("Nodes:")
for index, name in enumerate(node_names):
print(f" Node{index}: {name}")
changes = deque(maxlen=CHANGES_TO_KEEP)
recordings = create_recordings(controller, gather_points(controller), changes)
for recording in recordings:
recording.configure()
record_until_stopped(controller, recordings, record_seconds)
print_results(controller, recordings, changes, node_names)
exit_code = constants.EXIT_SUCCESS
except Exception as e:
print(f"❌ Error: {e}")
exit_code = constants.EXIT_FAILURE
finally:
controller.Delete()
sys.exit(exit_code)
1.15.0