Wiznet makers

ruilixin6

Published August 24, 2026 ©

187 UCC

0 VAR

0 Contests

0 Followers

0 Following

Original Link

Advanced Embedded: Build Recovery‑Enabled Software Watchdog with MicroPython

Advanced embedded hands‑on tutorial, implementing software watchdog with custom recovery logic based on MicroPython for Raspberry Pi Pico.

COMPONENTS
PROJECT DESCRIPTION

【Preliminary Note】The original hardware example in this article was written based on the RP2040. The actual hardware used in this hands-on demonstration features the W55RP20 as the main controller chip. The circuit logic and UF2 flashing operation principles are universally applicable, with only the main controller model differing. The original chip model mentioned in the circuit descriptions below is provided for reference purposes only.

In the previous example, we used the Raspberry Pi Pico built‑in watchdog to monitor system operation and reboot the device. However, as documented, the on‑chip hardware watchdog has several drawbacks:

1.PNG

The main limitations fall into two categories:

Fixed timeout period Once the timeout value is configured, it cannot be adjusted during runtime. When a feed‑dog timeout occurs, hardware asserts reset immediately with no possibility of delay or alternative handling routines.

Lack of extensibility The hardware watchdog only performs monitoring and reset. It cannot record system status, write logs or attempt system recovery before triggering reset.

In real‑world embedded applications, when program runaway or exceptions occur, we normally attempt recovery first: re‑initialize key modules, restart partial functions or record diagnostic logs. The software watchdog is designed for exactly this purpose. On microcontroller systems, a software watchdog can be implemented using timers or counters. Its working principle uses a timer/counter as trigger source. The timer keeps counting down. If it is not reloaded or “fed” within the configured period, the system is regarded as hung or unresponsive, and watchdog actions such as system reset or other safety procedures are executed.

Below is a multi‑functional timer‑based software‑watchdog implementation. Its main features include:

Flexible timeout configuration and reset delay Implemented on top of timers, timeout can be configured and dynamically modified at runtime. A pre‑reset delay can be defined to reserve time for system‑recovery workflows.

Can be actively stopped The timer can be stopped by calling the stop() method, which allows temporary watchdog disablement for system maintenance or firmware‑upgrade scenarios.

Flexible parameter configuration Parameters such as maximum allowed consecutive feed failures, debug‑mode enable, and pre‑reset delay are configurable.

Flexible extension capabilities Users may pass custom callback functions for status logging, trigger‑condition judgment and recovery workflows:

Status recording and logging Register status‑recorder callback to capture runtime state, timestamps and failure counters upon faults for post‑mortem debugging.

Trigger‑condition evaluation Custom trigger‑condition callbacks decide whether reset should proceed. Under certain fault conditions, recovery may be attempted before hard reset.

Recovery‑operation mechanism When program runaway is detected, recovery routines (network reconnection, partial‑function restarts etc.) run first. System reset only occurs if recovery fails, which better matches practical project requirements.

Debug and performance monitoring A timing decorator measures execution duration of watchdog callback functions for system‑performance analysis and optimization. When debug mode is enabled, detailed runtime status and event logs are printed to help developers trace faults and response behaviour.

Available methods and properties:

2.png

3.png

Source code for the custom software‑watchdog class:

# Allocate emergency‑exception buffer (must appear before any interrupt‑related code)
micropython.alloc_emergency_exception_buf(100)

# Software watchdog class
class SoftwareWatchdog:
    """
    Software watchdog class implemented with MicroPython Timer.
    This class encapsulates timer‑driven watchdog logic, supporting timeout detection, feed‑dog operation, status recording, trigger‑condition evaluation and recovery handlers.
    Users register callbacks to implement custom status logging, trigger‑condition checks and recovery workflows.

    Attributes:
        timeout (int): Watchdog timeout in milliseconds (default 4000ms).
        debug (bool): Enable debug mode (default True).
        max_failures (int): Maximum allowed consecutive feed failures (default 1).
        reset_delay (int): Delay before performing system reset, in milliseconds (default 3000ms).
        feed_successful (bool): Flag indicating successful feed‑dog event.
        timer (Timer): Software‑timer instance for periodic watchdog checks.
        feed_count (int): Counter for total feed‑dog events.
        trigger_count (int): Counter for total watchdog trigger events.
        failure_count (int): Counter for consecutive feed failures.
        _state_recorder (callable): User‑provided status‑recorder callback.
        _trigger_condition (callable): User‑provided trigger‑condition callback.
        _recovery_handler (callable): User‑provided recovery‑operation callback.

    Methods:
        __init__(self, timeout: int = 4000, debug: bool = True, max_failures: int = 1, reset_delay: int = 3000) -> None:
            Initialize software‑watchdog instance.
        _initialize_timer(self) -> None:
            Initialize timer and attach callback function.
        register_state_recorder(self, recorder: callable[[], None]) -> None:
            Register status‑recorder callback.
        set_trigger_condition(self, condition: callable[[], bool]) -> None:
            Set trigger‑condition callback.
        register_recovery_handler(self, handler: callable[[], bool]) -> None:
            Register recovery‑operation callback.
        _watchdog_callback(self, t: Timer) -> None:
            Timer callback: check feed‑dog status, run status logging and evaluate reset trigger conditions.
        feed(self) -> None:
            Perform feed‑dog operation and reset feed flag.
        stop(self) -> None:
            Stop watchdog timer peripheral.
        __del__(self) -> None:
            Destructor to release timer resources.
    """
    def __init__(self, timeout: int = 4000, debug: bool = True, max_failures: int = 1, reset_delay: int = 3000) -> None:
        """
        Initialize software watchdog.
        Args:
            timeout (int): Watchdog timeout in milliseconds (default 4000ms).
            debug (bool): Enable debug mode (default True).
            max_failures (int): Maximum allowed consecutive feed failures (default 1).
            reset_delay (int): Pre‑reset delay in milliseconds (default 3000ms).
        Returns:
            None
        Raises:
            ValueError: timeout, max_failures or reset_delay are not positive integers; debug is not boolean.
        """
        # Validate input arguments
        if not isinstance(reset_delay, int) or reset_delay <= 0:
            raise ValueError("reset_delay must be a positive integer")
        if not isinstance(timeout, int) or timeout <= 0:
            raise ValueError("timeout must be a positive integer")
        if not isinstance(max_failures, int) or max_failures <= 0:
            raise ValueError("max_failures must be a positive integer")
        if not isinstance(debug, bool):
            raise TypeError("debug must be a boolean value")

        self.timeout = timeout
        self.debug = debug
        self.max_failures = max_failures
        self.reset_delay = reset_delay

        self.feed_successful = False
        self.timer = Timer(-1)
        self.feed_count = 0
        self.trigger_count = 0
        self.failure_count = 0

        self._state_recorder = None
        self._trigger_condition = None
        self._recovery_handler = None

        self._initialize_timer()

    def _initialize_timer(self) -> None:
        """
        Initialize timer hardware and attach callback routine.
        Args:
            None
        Returns:
            None
        """
        self.timer.init(period=self.timeout, mode=Timer.PERIODIC, callback=lambda t: schedule(self._watchdog_callback, t))

    def register_state_recorder(self, recorder: callable[[], None]) -> None:
        """
        Register status‑recorder callback function.
        Args:
            recorder (callable): Zero‑argument callback returning None for status logging.
        Returns:
            None
        Raises:
            TypeError: recorder is not callable.
        """
        if not callable(recorder):
            raise TypeError("State recorder must be callable")
        self._state_recorder = recorder

    def set_trigger_condition(self, condition: callable[[], bool]) -> None:
        """
        Assign trigger‑condition evaluation callback.
        Args:
            condition (callable): Zero‑argument callback returning boolean; True permits system reset.
        Returns:
            None
        Raises:
            TypeError: condition is not callable.
        """
        if not callable(condition):
            raise TypeError("Trigger condition must be callable")
        self._trigger_condition = condition

    def register_recovery_handler(self, handler: callable[[], bool]) -> None:
        """
        Register recovery‑operation callback function.
        Args:
            handler (callable): Zero‑argument callback returning boolean; True indicates recovery success.
        Returns:
            None
        Raises:
            TypeError: handler is not callable.
        """
        if not callable(handler):
            raise TypeError("Recovery handler must be callable")
        self._recovery_handler = handler

    @timed_function
    def _watchdog_callback(self, t: Timer) -> None:
        """
        Timer callback routine: check feed‑dog status, execute logging and decide whether to trigger reset.
        Args:
            t (Timer): Timer object passed automatically by hardware.
        Returns:
            None
        Raises:
            Exception: Errors inside status‑recorder or recovery‑handler execution.
            TypeError: Recovery handler does not return boolean value.
            Exception: Errors inside trigger‑condition callback.
        """
        # Atomic read of feed flag
        irq_state = disable_irq()
        feed_flag = self.feed_successful
        enable_irq(irq_state)

        if not feed_flag:
            self.failure_count += 1
            self.trigger_count += 1

            if self.debug:
                print("[Watchdog] Triggered ({} failures, {} total triggers)".format(
                    self.failure_count, self.trigger_count))

            # Run status‑recorder if registered
            if self._state_recorder:
                try:
                    self._state_recorder()
                except Exception as e:
                    if self.debug:
                        print("[Error] Failed to record state:", str(e))

            if self.failure_count >= self.max_failures:
                recovery_successful = False

                if self._recovery_handler:
                    try:
                        if self.debug:
                            print("[Watchdog] Attempting recovery...")
                        recovery_successful = self._recovery_handler()
                        if not isinstance(recovery_successful, bool):
                            raise TypeError("Recovery handler must return a boolean value")
                    except Exception as e:
                        if self.debug:
                            print("[Error] Recovery handler failed:", str(e))

                if recovery_successful:
                    self.failure_count = 0
                    self.should_trigger = False
                    if self.debug:
                        print("[Watchdog] Recovery successful, resetting failure count...")
                else:
                    should_trigger = True
                    if self._trigger_condition:
                        try:
                            should_trigger = self._trigger_condition()
                        except Exception as e:
                            if self.debug:
                                print("[Error] Trigger condition check failed:", str(e))
                            should_trigger = True

                    if should_trigger:
                        if self.debug:
                            print("[Watchdog] Max failures reached, resetting system after %d ms..." %(self.reset_delay))
                        self.reset_timer = Timer(-1)
                        self.reset_timer.init(period=self.reset_delay, mode=Timer.ONE_SHOT, callback=lambda t: reset())
        else:
            self.failure_count = 0
            irq_state = disable_irq()
            self.feed_successful = False
            enable_irq(irq_state)

    def feed(self) -> None:
        """
        Feed‑dog operation: set successful‑feed flag.
        Args:
            None
        Returns:
            None
        """
        irq_state = disable_irq()
        self.feed_successful = True
        self.feed_count += 1
        enable_irq(irq_state)
        if self.debug:
            print("Watchdog fed at:", time.ticks_ms())

    def stop(self) -> None:
        """
        Stop watchdog timer peripheral.
        Args:
            None
        Returns:
            None
        """
        self.timer.deinit()
        if self.debug:
            print("Watchdog stopped.")

    def __del__(self):
        """
        Destructor: release timer resources.
        Args:
            None
        Returns:
            None
        """
        self.timer.deinit()
        if self.debug:
            print("Watchdog resources released.")

Before class definition we allocate an emergency‑exception buffer. This stores exception information generated within interrupt context, allowing faults inside interrupt handlers to be captured for debugging.

The core principle of this class uses MicroPython Timer module. Inside _initialize_timer() a periodic timer invokes _watchdog_callback to poll the feed‑dog flag and realise watchdog behaviour similar to hardware watchdogs. When consecutive feed failures (system unresponsiveness) are detected, recovery procedures or system reset are triggered to prevent prolonged abnormal system state.

The timer callback _watchdog_callback implements all extended logic:

4.png

Execution flow summary:

  1. Feed‑dog detection Read feed_successful flag to check for feed‑dog events. If no feed‑dog occurred within current cycle (feed_successful is False), increment consecutive‑failure and total‑trigger counters and output debug messages.
  2. Status recording If a status‑recorder callback is registered, invoke it upon missed feed‑dog events to capture runtime state.
  3. Recovery and reset decision
  4. When consecutive failures reach max_failures, attempt recovery‑handler routine first.
  5. If recovery succeeds, reset consecutive‑failure counter. If recovery fails, evaluate trigger‑condition callback (if registered) to decide whether to perform system reset.
  6. System reset is implemented with a one‑shot timer, invoking reset() after reset_delay milliseconds.

Timing diagram:

5.png

Three registration methods attach custom callbacks:

Status‑recorder function Use register_state_recorder to register callback for logging runtime status and diagnostic information for fault analysis.

Trigger‑condition function Use set_trigger_condition to assign boolean‑returning callback evaluating whether reset should proceed after recovery failure. Logic can be customised per‑application requirements.

Recovery‑handler function Register recovery callback with register_recovery_handler. It executes upon hitting maximum consecutive feed failures to attempt system repair. Success resets failure counters; failure may lead to reset according to trigger‑condition evaluation.

Below are sample custom callback functions for class validation:

# Python env   : MicroPython v1.23.0
# -*- coding: utf-8 -*-
# @Time    : 2024/8/16 10:51 AM
# @Author  : Li Qingshui
# @File    : main.py
# @Description : WDT watchdog timer experiment, software‑timer‑based watchdog for Pico
# ======================================== Import related modules ========================================
from machine import Timer, reset, disable_irq, enable_irq
import time
from micropython import schedule
import micropython

# ======================================== Global variables ============================================
threshold = 10
current_value = 12
watchdog = None

# ======================================== Function definitions ============================================
def user_log_critical_time() -> None:
    """
    User‑defined status‑recorder function: write timestamp, current value, watchdog trigger count and consecutive‑failure count into log files.
    Automatically create new log file when line count exceeds 50 entries.
    Args:
        None
    Returns:
        None
    Raises:
        Exception: File‑write failures.
    """
    global watchdog, current_value
    timestamp = time.ticks_ms()
    log_base_name = "/log"
    log_extension = ".txt"
    log_index = 0
    log_file = f"{log_base_name}{log_index}{log_extension}"

    while True:
        try:
            with open(log_file, "r") as f:
                lines = f.readlines()
                if len(lines) < 10:
                    break
        except OSError:
            break
        log_index += 1
        log_file = f"{log_base_name}{log_index}{log_extension}"

    try:
        with open(log_file, "a") as f:
            log_entry = "Timestamp: %d ms, Current Value: %d, Triggers: %d, Failures: %d\n" % (
                timestamp, current_value, watchdog.trigger_count, watchdog.failure_count
            )
            f.write(log_entry)
            f.flush()
    except Exception as e:
        print("[Error] Failed to write log:", str(e))

@micropython.native
def user_check_threshold() -> bool:
    """
    User‑defined trigger‑condition function to evaluate threshold criteria.
    Args:
        None
    Returns:
        bool: True when threshold reached, False otherwise.
    """
    global current_value, threshold
    if current_value >= threshold:
        print("[Trigger] Threshold reached, triggering watchdog...")
        return True
    print("[Info] Current value is below threshold, no need to trigger watchdog...")
    return False

@micropython.native
def user_recovery_handler() -> bool:
    """
    User‑defined recovery‑operation handler.
    Args:
        None
    Returns:
        bool: True means recovery succeeded; False means recovery failed.
    Raises:
        Exception: Errors during recovery workflow.
    """
    global watchdog
    print("[Recovery] Attempting to recover system...")
    try:
        pass
    except Exception as e:
        print("[Error] Failed to recover system:", str(e))
        return False
    print("[Recovery] Recovery operation completed successfully.")
    return True

Explanation for each callback function:

user_log_critical_time() status‑recorder function Log key runtime information including timestamp, current value, watchdog trigger counter and consecutive‑failure counter to log files.

Log‑file management: construct filenames such as /log0.txt, /log1.txt. Detect existing files and line count; increment index and create new file once line threshold is hit to prevent oversized single log file.

Status logging: capture millisecond‑level system timestamp and assemble log entries from global variables, append to target file.

Exception handling: wrap file‑access logic inside try/except blocks, print error hints without aborting program.

user_check_threshold() trigger‑condition function Evaluate whether system state variable meets preset threshold. Return True when current_value >= threshold and print hint message; otherwise return False. This function is for demonstration purposes using static hard‑coded values.

user_recovery_handler() recovery‑handler function Simulate system‑recovery workflow invoked upon hitting consecutive‑failure limit. Return True for simulated success, return False upon exceptions. This demo assumes recovery always succeeds for illustration.

Decorator @micropython.native is applied to trigger‑condition and recovery‑handler functions to enable native‑code emitter. Native‑code execution runs approximately twice as fast as byte‑code at the cost of larger memory footprint.

Timing decorator implementation for measuring function execution duration:

@micropython.native
def timed_function(f: callable, *args: tuple, **kwargs: dict) -> callable:
    """
    Timing decorator to measure and print function execution time.
    Args:
        f (callable): Target function to wrap.
        args (tuple): Variable positional arguments.
        kwargs (dict): Variable keyword arguments.
    Returns:
        callable: Wrapped timing‑measuring function.
    """
    myname = str(f).split(' ')[1]
    def new_func(*args: tuple, **kwargs: dict) -> any:
        t: int = time.ticks_us()
        result = f(*args, **kwargs)
        delta: int = time.ticks_diff(time.ticks_us(), t)
        print('Function {} Time = {:6.3f}ms'.format(myname, delta / 1000))
        return result
    return new_func

Instantiate SoftwareWatchdog, then register callbacks by passing function references:

def register_state_recorder(self, recorder: callable[[], None]) -> None:
    """
    Register status‑recorder callback function.
    Args:
        recorder (callable): Zero‑argument callback for status logging.
    Returns:
        None
    Raises:
        TypeError: recorder is not callable.
    """
    if not callable(recorder):
        raise TypeError("State recorder must be callable")
    self._state_recorder = recorder

recorder stores function reference. Later self._state_recorder() invokes the original user‑supplied routine.

Sample registration statements:

# Initialize software watchdog: timeout 4 seconds, max 3 consecutive failures, reset delay 1 second
watchdog = SoftwareWatchdog(timeout=4000, debug=True, max_failures=3, reset_delay=1000)
# Register status‑recorder callback
watchdog.register_state_recorder(user_log_critical_time)
# Assign trigger‑condition callback
watchdog.set_trigger_condition(user_check_threshold)
# Register recovery‑operation callback
watchdog.register_recovery_handler(user_recovery_handler)

Complete project source code is located in resource package path elegance‑devkit v1\Demo\53 WDG_Timer.

Full demonstration code:

# Python env   : MicroPython v1.23.0
# -*- coding: utf-8 -*-
# @Time    : 2024/8/16 10:51 AM
# @Author  : Li Qingshui
# @File    : main.py
# @Description : WDT watchdog timer experiment, software‑timer‑based watchdog for Pico
# ======================================== Import related modules ========================================
from machine import Timer, reset, disable_irq, enable_irq
import time
from micropython import schedule
import micropython

# ======================================== Global variables ============================================
threshold = 10
current_value = 12
watchdog = None

# ======================================== Function definitions ============================================
def user_log_critical_time() -> None:
    """
    User‑defined status‑recorder function: write timestamp, current value, watchdog trigger count and consecutive‑failure count into log files.
    Automatically create new log file when line count exceeds 50 entries.
    Args:
        None
    Returns:
        None
    Raises:
        Exception: File‑write failures.
    """
    global watchdog, current_value
    timestamp = time.ticks_ms()
    log_base_name = "/log"
    log_extension = ".txt"
    log_index = 0
    log_file = f"{log_base_name}{log_index}{log_extension}"

    while True:
        try:
            with open(log_file, "r") as f:
                lines = f.readlines()
                if len(lines) < 10:
                    break
        except OSError:
            break
        log_index += 1
        log_file = f"{log_base_name}{log_index}{log_extension}"

    try:
        with open(log_file, "a") as f:
            log_entry = "Timestamp: %d ms, Current Value: %d, Triggers: %d, Failures: %d\n" % (
                timestamp, current_value, watchdog.trigger_count, watchdog.failure_count
            )
            f.write(log_entry)
            f.flush()
    except Exception as e:
        print("[Error] Failed to write log:", str(e))

@micropython.native
def user_check_threshold() -> bool:
    """
    User‑defined trigger‑condition function to evaluate threshold criteria.
    Args:
        None
    Returns:
        bool: True when threshold reached, False otherwise.
    """
    global current_value, threshold
    if current_value >= threshold:
        print("[Trigger] Threshold reached, triggering watchdog...")
        return True
    print("[Info] Current value is below threshold, no need to trigger watchdog...")
    return False

@micropython.native
def user_recovery_handler() -> bool:
    """
    User‑defined recovery‑operation handler.
    Args:
        None
    Returns:
        bool: True means recovery succeeded; False means recovery failed.
    Raises:
        Exception: Errors during recovery workflow.
    """
    global watchdog
    print("[Recovery] Attempting to recover system...")
    try:
        pass
    except Exception as e:
        print("[Error] Failed to recover system:", str(e))
        return False
    print("[Recovery] Recovery operation completed successfully.")
    return True

@micropython.native
def timed_function(f: callable, *args: tuple, **kwargs: dict) -> callable:
    """
    Timing decorator to measure and print function execution time.
    Args:
        f (callable): Target function to wrap.
        args (tuple): Variable positional arguments.
        kwargs (dict): Variable keyword arguments.
    Returns:
        callable: Wrapped timing‑measuring function.
    """
    myname = str(f).split(' ')[1]
    def new_func(*args: tuple, **kwargs: dict) -> any:
        t: int = time.ticks_us()
        result = f(*args, **kwargs)
        delta: int = time.ticks_diff(time.ticks_us(), t)
        print('Function {} Time = {:6.3f}ms'.format(myname, delta / 1000))
        return result
    return new_func

# ======================================== Custom classes ============================================
micropython.alloc_emergency_exception_buf(100)

class SoftwareWatchdog:
    """
    Software watchdog class implemented with MicroPython Timer.
    This class encapsulates timer‑driven watchdog logic, supporting timeout detection, feed‑dog operation, status recording, trigger‑condition evaluation and recovery handlers.
    Users register callbacks to implement custom status logging, trigger‑condition checks and recovery workflows.

    Attributes:
        timeout (int): Watchdog timeout in milliseconds (default 4000ms).
        debug (bool): Enable debug mode (default True).
        max_failures (int): Maximum allowed consecutive feed failures (default 1).
        reset_delay (int): Delay before performing system reset, in milliseconds (default 3000ms).
        feed_successful (bool): Flag indicating successful feed‑dog event.
        timer (Timer): Software‑timer instance for periodic watchdog checks.
        feed_count (int): Counter for total feed‑dog events.
        trigger_count (int): Counter for total watchdog trigger events.
        failure_count (int): Counter for consecutive feed failures.
        _state_recorder (callable): User‑provided status‑recorder callback.
        _trigger_condition (callable): User‑provided trigger‑condition callback.
        _recovery_handler (callable): User‑provided recovery‑operation callback.

    Methods:
        __init__(self, timeout: int = 4000, debug: bool = True, max_failures: int = 1, reset_delay: int = 3000) -> None:
            Initialize software‑watchdog instance.
        _initialize_timer(self) -> None:
            Initialize timer and attach callback function.
        register_state_recorder(self, recorder: callable[[], None]) -> None:
            Register status‑recorder callback.
        set_trigger_condition(self, condition: callable[[], bool]) -> None:
            Set trigger‑condition callback.
        register_recovery_handler(self, handler: callable[[], bool]) -> None:
            Register recovery‑operation callback.
        _watchdog_callback(self, t: Timer) -> None:
            Timer callback: check feed‑dog status, run status logging and evaluate reset trigger conditions.
        feed(self) -> None:
            Perform feed‑dog operation and reset feed flag.
        stop(self) -> None:
            Stop watchdog timer peripheral.
        __del__(self) -> None:
            Destructor to release timer resources.
    """
    def __init__(self, timeout: int = 4000, debug: bool = True, max_failures: int = 1, reset_delay: int = 3000) -> None:
        if not isinstance(reset_delay, int) or reset_delay <= 0:
            raise ValueError("reset_delay must be a positive integer")
        if not isinstance(timeout, int) or timeout <= 0:
            raise ValueError("timeout must be a positive integer")
        if not isinstance(max_failures, int) or max_failures <= 0:
            raise ValueError("max_failures must be a positive integer")
        if not isinstance(debug, bool):
            raise TypeError("debug must be a boolean value")

        self.timeout = timeout
        self.debug = debug
        self.max_failures = max_failures
        self.reset_delay = reset_delay

        self.feed_successful = False
        self.timer = Timer(-1)
        self.feed_count = 0
        self.trigger_count = 0
        self.failure_count = 0

        self._state_recorder = None
        self._trigger_condition = None
        self._recovery_handler = None

        self._initialize_timer()

    def _initialize_timer(self) -> None:
        self.timer.init(period=self.timeout, mode=Timer.PERIODIC, callback=lambda t: schedule(self._watchdog_callback, t))

    def register_state_recorder(self, recorder: callable[[], None]) -> None:
        if not callable(recorder):
            raise TypeError("State recorder must be callable")
        self._state_recorder = recorder

    def set_trigger_condition(self, condition: callable[[], bool]) -> None:
        if not callable(condition):
            raise TypeError("Trigger condition must be callable")
        self._trigger_condition = condition

    def register_recovery_handler(self, handler: callable[[], bool]) -> None:
        if not callable(handler):
            raise TypeError("Recovery handler must be callable")
        self._recovery_handler = handler

    @timed_function
    def _watchdog_callback(self, t: Timer) -> None:
        irq_state = disable_irq()
        feed_flag = self.feed_successful
        enable_irq(irq_state)

        if not feed_flag:
            self.failure_count += 1
            self.trigger_count += 1

            if self.debug:
                print("[Watchdog] Triggered ({} failures, {} total triggers)".format(
                    self.failure_count, self.trigger_count))

            if self._state_recorder:
                try:
                    self._state_recorder()
                except Exception as e:
                    if self.debug:
                        print("[Error] Failed to record state:", str(e))

            if self.failure_count >= self.max_failures:
                recovery_successful = False
                if self._recovery_handler:
                    try:
                        if self.debug:
                            print("[Watchdog] Attempting recovery...")
                        recovery_successful = self._recovery_handler()
                        if not isinstance(recovery_successful, bool):
                            raise TypeError("Recovery handler must return a boolean value")
                    except Exception as e:
                        if self.debug:
                            print("[Error] Recovery handler failed:", str(e))

                if recovery_successful:
                    self.failure_count = 0
                    self.should_trigger = False
                    if self.debug:
                        print("[Watchdog] Recovery successful, resetting failure count...")
                else:
                    should_trigger = True
                    if self._trigger_condition:
                        try:
                            should_trigger = self._trigger_condition()
                        except Exception as e:
                            if self.debug:
                                print("[Error] Trigger condition check failed:", str(e))
                            should_trigger = True
                    if should_trigger:
                        if self.debug:
                            print("[Watchdog] Max failures reached, resetting system after %d ms..." %(self.reset_delay))
                        self.reset_timer = Timer(-1)
                        self.reset_timer.init(period=self.reset_delay, mode=Timer.ONE_SHOT, callback=lambda t: reset())
        else:
            self.failure_count = 0
            irq_state = disable_irq()
            self.feed_successful = False
            enable_irq(irq_state)

    def feed(self) -> None:
        irq_state = disable_irq()
        self.feed_successful = True
        self.feed_count += 1
        enable_irq(irq_state)
        if self.debug:
            print("Watchdog fed at:", time.ticks_ms())

    def stop(self) -> None:
        self.timer.deinit()
        if self.debug:
            print("Watchdog stopped.")

    def __del__(self):
        self.timer.deinit()
        if self.debug:
            print("Watchdog resources released.")

# ======================================== Initialization ==========================================
time.sleep(3)
print("FreakStudio : Implement Watchdog Timer using a software timer Test")
watchdog = SoftwareWatchdog(timeout=4000, debug=True, max_failures=3, reset_delay=1000)
watchdog.register_state_recorder(user_log_critical_time)
watchdog.set_trigger_condition(user_check_threshold)
watchdog.register_recovery_handler(user_recovery_handler)

# ======================================== Main program ============================================
# Feed‑dog twice within timeout window, watchdog will not trigger reset
for i in range(2):
    watchdog.feed()
    time.sleep(2)

Flash firmware and connect to Raspberry Pi Pico. Terminal output:

6.png

Program execution flow:

  1. Program startup and initial feed‑dog After boot, terminal prints “FreakStudio : Implement Watchdog Timer using a software timer Test”. Two “Watchdog fed at: …” messages indicate two watchdog.feed() invocations. The output “Function _watchdog_callback Time = … ms” shows the timer has started periodic watchdog checking.

8.png

  1. Periodic checking and consecutive feed failures After main‑loop termination, feed() is no longer invoked. In each subsequent timer period the callback detects feed_successful == False and enters missed‑feed handling.
  2. Increment failure_count and trigger_count.
  3. First missed feed prints “[Watchdog] Triggered (1 failures, 1 total triggers)”.
  4. Second missed feed prints “[Watchdog] Triggered (2 failures, 2 total triggers)”.
  5. When consecutive failures reach three “(3 failures, 3 total triggers)”, recovery workflow executes.
  6. Recovery‑handler invocation

9.png

  1. “[Watchdog] Attempting recovery...” prints, then calls user_recovery_handler.
  2. Output sequence: “[Recovery] Attempting to recover system...” “[Recovery] Recovery operation completed successfully.” “[Watchdog] Recovery successful, resetting failure count...”
  3. Consecutive‑failure counter resets to zero, watchdog continues running. Total trigger counter keeps accumulating. This cycle repeats: every three consecutive missed feeds trigger one recovery pass.

Inspect log file over mpremote:

mpremote cat log0.txt

Log contains timestamp, system‑state value, watchdog trigger count and consecutive‑failure count:

10.png

Copy log file to host PC:

mpremote cp :log0.txt ./log0.txt

11.png

mmexport1782379485058.gif

Comment‑out recovery‑handler registration and run again:

# watchdog.register_recovery_handler(user_recovery_handler)

After three consecutive feed failures the device performs system reset.

Software watchdogs also have inherent limitations. They depend on system hardware and software resources. Under severe system faults the software watchdog can stop functioning. For example, fatal defects inside interrupt‑service‑routines (ISR) may paralyse the whole system. Since the software watchdog itself runs on top of system resources, it may be unable to execute reset operations, leaving the system unrecoverable.

Documents
Comments Write