in deepracer_systems_pkg/deepracer_systems_pkg/network_monitor_module/network_monitor_node.py [0:0]
def report_state(self, message=None):
"""Write the network connection report to the state_dir if the path is set.
Args:
message ([str, tuple, list], optional): Message that needs to be written to
the device connection status file.
Defaults to None.
"""
self.get_logger().info(f"report_state: {message}")
state_directory = ""
if self.state_dir != "":
state_directory = self.state_dir
if state_directory != "":
state_filename = os.path.join(state_directory, network_config.DEVICE_STATUS_NAME)
state_file_lines = list()
# First read the state file if exists and remove the state for this host.
try:
skipping = False
for line in open(state_filename):
line = line.rstrip()
if skipping and (line == ""):
skipping = False
continue
if line.lstrip().startswith("Hostname:"):
hostname = line.split(":")[1].strip()
skipping = (hostname == self.hostname)
if not skipping:
state_file_lines.append(line)
except Exception:
pass
try:
with open(state_filename, 'w') as state_file:
# Write back the state line for other hosts.
for line in state_file_lines:
print(line, file=state_file)
# Write the state for this host.
print(f"Hostname: {self.hostname}", file=state_file)
print(f"Timestamp: {time.strftime('%x %X')}", file=state_file)
if type(message) in (list, tuple):
for line in message:
self.get_logger().info(line)
print(line, file=state_file)
elif type(message) is str:
self.get_logger().info(message)
print(message, file=state_file)
ssid = network_utils.get_current_SSID()
if ssid == "":
ssid_message = "Not connected to WiFi network."
else:
ssid_message = f"Connected to WiFi network: {ssid}"
self.get_logger().info(ssid_message)
print(ssid_message, file=state_file)
connections = network_utils.get_net_connections()
if len(connections) != 0:
print("Active connections:", file=state_file)
for interface, ip in connections.items():
print(f" {interface}: {ip}", file=state_file)
print("", file=state_file)
except Exception as ex:
self.get_logger().error(f"Failed to write state file {state_filename}: {ex}")
file_system_utils.make_writable(state_filename)