MicroPython Forum (Archive)

Code: Select all

from machine import WDT
>>> wdt=WDT(timeout=2)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: function does not take keyword arguments

Code: Select all

#include <string.h>

#include "py/runtime.h"
#include "user_interface.h"
#include "etshal.h"
#include "ets_alt_task.h"

const mp_obj_type_t esp_wdt_type;

typedef struct _machine_wdt_obj_t {
    mp_obj_base_t base;
} machine_wdt_obj_t;

STATIC machine_wdt_obj_t wdt_default = {{&esp_wdt_type}};

STATIC mp_obj_t machine_wdt_make_new(const mp_obj_type_t *type_in, size_t n_args, size_t n_kw, const mp_obj_t *args) {
    mp_arg_check_num(n_args, n_kw, 0, 1, false);

    mp_int_t id = 0;
    if (n_args > 0) {
        id = mp_obj_get_int(args[0]);
    }

    switch (id) {
    case 0:
        ets_loop_dont_feed_sw_wdt = 1;
        system_soft_wdt_feed();
        return &wdt_default;
    default:
        mp_raise_ValueError(NULL);
    }
}

STATIC mp_obj_t machine_wdt_feed(mp_obj_t self_in) {
    (void)self_in;
    system_soft_wdt_feed();
    return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_1(machine_wdt_feed_obj, machine_wdt_feed);

STATIC mp_obj_t machine_wdt_deinit(mp_obj_t self_in) {
    (void)self_in;
    ets_wdt_disable();
    return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_1(machine_wdt_deinit_obj, machine_wdt_deinit);

STATIC const mp_rom_map_elem_t machine_wdt_locals_dict_table[] = {
    { MP_ROM_QSTR(MP_QSTR_feed), MP_ROM_PTR(&machine_wdt_feed_obj) },
    { MP_ROM_QSTR(MP_QSTR_deinit), MP_ROM_PTR(&machine_wdt_deinit_obj) },
};
STATIC MP_DEFINE_CONST_DICT(machine_wdt_locals_dict, machine_wdt_locals_dict_table);

const mp_obj_type_t esp_wdt_type = {
    { &mp_type_type },
    .name = MP_QSTR_WDT,
    .make_new = machine_wdt_make_new,
    .locals_dict = (mp_obj_dict_t*)&machine_wdt_locals_dict,
};

Code: Select all

import gc
import uasyncio as asyncio
import machine
from pysmartnode.utils import sys_vars
from sys import platform

gc.collect()
from pysmartnode import logging


class WDT:
    def __init__(self, id=0, timeout=120, use_rtc_memory=True):
        self._timeout = timeout / 10
        self._counter = 0
        self._timer = machine.Timer(id)
        self._use_rtc_memory = use_rtc_memory
        self.init()
        asyncio.get_event_loop().create_task(self._resetCounter())
        if sys_vars.hasFilesystem():
            try:
                with open("watchdog.txt", "r") as f:
                    if f.read() == "True":
                        logging.getLogger("WDT").warn("Reset reason: Watchdog")
            except Exception as e:
                print(e)  # file probably just does not exist
            try:
                with open("watchdog.txt", "w") as f:
                    f.write("False")
            except Exception as e:
                logging.getLogger("WDT").error("Error saving to file: {!s}".format(e))
        elif use_rtc_memory and platform == "esp8266":
            rtc = machine.RTC()
            if rtc.memory() == b"WDT reset":
                logging.getLogger("WDT").warn("Reset reason: Watchdog")
            rtc.memory(b"")

    def _wdt(self, t):
        self._counter += self._timeout
        if self._counter >= self._timeout * 10:
            if sys_vars.hasFilesystem():
                try:
                    with open("watchdog.txt", "w") as f:
                        f.write("True")
                except Exception as e:
                    print("Error saving to file: {!s}".format(e))
            elif self._use_rtc_memory and platform == "esp8266":
                rtc = machine.RTC()
                rtc.memory(b"WDT reset")
            machine.reset()

    def feed(self):
        self._counter = 0

    def init(self, timeout=None):
        timeout = timeout or self._timeout
        self._timeout = timeout
        self._timer.init(period=int(self._timeout * 1000), mode=machine.Timer.PERIODIC, callback=self._wdt)

    def deinit(self):  # will not stop coroutine
        self._timer.deinit()

    async def _resetCounter(self):
        while True:
            await asyncio.sleep(self._timeout)
            self.feed()