def get_update_state()

in deepracer_systems_pkg/deepracer_systems_pkg/software_update_module/software_update_node.py [0:0]

• 21 lines of code
• 3 McCabe index (conditional complexity)

    def get_update_state(self, force_update_check=False):
        """Return the update status after perfoming software update check or waiting for current
           ongoing check to complete.

        Args:
            force_update_check (bool, optional): Set to True to force schedule update check immediately.
                                                 Defaults to False.

        Returns:
            int: Software update state information.
        """
        self.get_logger().info("Get software update state...")
        with utility.AutoLock(self.state_guard):

            # Force update check if it hasn"t been completed yet.
            if self.update_state == software_update_config.SoftwareUpdateState.UPDATE_UNKNOWN:
                force_update_check = True

            # Kick off update check if needed.
            if not self.check_in_progress:
                if force_update_check:
                    self.check_complete.clear()
                    self.scheduler.schedule_action(self.do_update_check)
                else:
                    self.get_logger().info("Force software update check flag is not set")
            else:
                self.get_logger().info("Previous scheduled software update check in progress...")

        # Wait for the check to complete.
        while not self.check_complete.wait(2):
            self.get_logger().info("Waiting for software update check complete...")

        self.get_logger().info("Software update check complete.")
        with utility.AutoLock(self.state_guard):
            # State here can still be unknown if it had been unknown before the last update
            # and the last update failed. Report and do the forced check next time.
            update_state = self.update_state
            self.get_logger().info("Software update status: "
                                   f"{self.get_state_description(update_state)}")
            return update_state