def update_configuration()

in deepracer_systems_pkg/deepracer_systems_pkg/network_monitor_module/network_monitor_node.py [0:0]

• 40 lines of code
• 7 McCabe index (conditional complexity)

    def update_configuration(self, keyworded_args):
        """Main function to parse the WiFi configuration file, connect to WiFi,
           update WiFi LED status during the process and publish the final network
           connection status.

        Args:
            keyworded_args (dict): Keyworded arguments passed to the function while scheduling.
        """
        self.get_logger().info(f"Update Wifi Configuration: {keyworded_args}")
        connected = False
        connection_report = list()

        # Enable WiFi LED effect.
        self.led_update_pause.set()
        self.led_blink_request.led_index = constants.LEDIndex.WIFI_LED
        self.led_blink_request.color1 = constants.LEDColor.BLUE
        self.led_blink_request.color2 = constants.LEDColor.NO_COLOR
        self.led_blink_request.delay = 0.2
        self.led_blink_service.call_async(self.led_blink_request)

        base_path = keyworded_args.get("path", "")
        config_name = keyworded_args.get("name", "")
        node_name = keyworded_args.get("node_name", None)

        full_name = os.path.join(base_path, config_name)

        # Store and use base path as the target path to write status file to.
        self.state_dir = base_path

        # Read the configuration file.
        self.get_logger().info(f"Reading WiFi configuration {full_name}")
        config_dict = wifi_config_utils.parse_wifi_config(full_name)
        self.get_logger().info(f"Config dictionary read from file: {config_dict}")
        if config_dict is None:
            connection_report.append(f"{config_name}: failed to read WiFi configuration file.")

        if network_config.WiFiConfigKeys.SSID not in config_dict:
            connection_report.append(f"{config_name}: No SSID found in WiFi configuration file.")
        else:
            connected = self.connect_wifi_using_config(config_dict, connection_report)
        self.report_state(connection_report)

        # Unmount the media.
        if node_name is not None:
            mount_point_mgr_request = USBMountPointManagerSrv.Request()
            mount_point_mgr_request.node_name = node_name
            mount_point_mgr_request.action = 0
            self.usb_mount_point_manager_client.call_async(mount_point_mgr_request)
            self.state_dir = ""

        # Turn on alarm effect if failed.
        if not connected:
            self.led_solid_request.led_index = constants.LEDIndex.WIFI_LED
            self.led_solid_request.color = constants.LEDColor.RED
            self.led_solid_request.hold = 2.0
            self.led_solid_service.call_async(self.led_solid_request)

        # Set the correct LED color.
        self.update_status_LED()
        self.led_update_pause.clear()

        # Publish the network connection status
        self.publish_network_connection_status()
        self.get_logger().info("Updating WiFi configuration completed")