The review of Evok 3.1.0: a failed step, the watchdog timeout, the alias before the write

docs/api-changes-dev-src.md

@@ -30,7 +30,9 @@ the HW definitions. Breaking changes come first. Every change names its commits
 - **Watchdog:** — `6ac4eb0`, `f298e87`, `fce8d1d`
   - its circuit is the name of the unit (`1`) instead of `1_01`, the saved aliases of watchdogs are not assigned,
   - `nv_save` returns an error and is not in `full()`,
-  - `value` is 0/1 instead of 0–3, a restart is reported only by `was_wd_reset`.
+  - `value` is 0/1 instead of 0–3, a restart is reported only by `was_wd_reset`,
+  - `timeout` must be a whole number of ms, at most 65535 (a larger one is cut to it as before); an invalid
+    `timeout`, `value` or `reset` is refused before any change; a changed timeout is an event (#280).
 - **AI and AO:** `modes` has only `unit` and `range`, without the code of the register and the `transformation`.
   — `e792e17`
 - **OwPower:** `value` is always 0/1, before `true`/`false` after a write. — `2042b3d`
@@ -116,7 +118,13 @@ the HW definitions. Breaking changes come first. Every change names its commits
 - **Sequencer** of each output (RO, DO, LED), the device `sequencer` with the circuit `<type>_<circuit>`
   (`/rest/sequencer/do_1_01`): a pulse (`value`, `pulse_duration`), a sequence of the changes of the output
   (`steps`, `final`) or `cancel`; its state has `running` and the progress, an event at the start, steps and end.
-  The RPC method `relay_set_for_time`. — `e16a3bf`, `7789635`, `a85fb14`
+  A step failed by an error of the communication (e.g. a noise on the bus) is written again after 1 s up to
+  30 s, the sequence stays running, the output does not stay in the state of a pulse; a step refused by the unit
+  ends the sequence after its final state (if it has one) is written. The RPC method `relay_set_for_time`.
+  — `e16a3bf`, `7789635`, `a85fb14`, #280
+- An invalid alias in a change (a circuit, reserved, invalid or already used) is refused before the first write,
+  the output was switched and then the request failed. The aliases of a bulk are checked in the order of the
+  changes, an alias released by an earlier change can be taken. — #280
 - The reply to POST, bulk and the WebSocket `set` has the state after the writes and the event comes at once.
   — `a3f4679`
 - **New events:** — `60e2287`, `08d9f56`, `62b37ef`, `891bd4a`
@@ -190,7 +198,7 @@ An invalid value is an error: the unit or the feature is not created and the err
   a next write renews it; then the periodic scan reads its block, a value refused by the unit is seen then. A unit
   may set a written register with a delay, e.g. xS52 the mode of its AI in 0.3–0.8 s: the read returned the old
   mode and the switch seemed to fail, it was seen by the next scan of the slow block after several seconds.
-  A unit without the periodic scan (`scan_enabled: false`) shows the value read after the change.
+  A unit without the periodic scan (`scan_enabled: false`) shows the value read after the change. — #278, #279
 - **Modbus addresses:** a serial port can be used by one `MODBUSRTU` bus only, the `slave-id` must be unique on an RTU
   bus and on a TCP server (`hostname`, `port`) over all `MODBUSTCP` buses. The second bus or device is not created,
   before both read and wrote the same unit. — `dc7b1ee`

docs/apis/sequencer.md

@@ -9,7 +9,14 @@ At most one pulse or sequence runs for an output, a new one replaces it. While i
 itself (except its alias) is refused: REST, JSON and bulk with the status `409` (`SequenceRunning`), the WebSocket
 with the same error, RPC with the code `-32002`; cancel the sequence by its sequencer first. When Evok stops,
 it writes the final state of the running pulses and sequences which have one (`final`), an output is not left
-in the state of a pulse; a sequence without `final` leaves the output in its state. After a crash of Evok
+in the state of a pulse; a sequence without `final` leaves the output in its state. A step which fails by an error of
+the communication, e.g. a noise on the bus or an unavailable unit, is written again after 1 s, the delay doubles
+up to 30 s, and the sequence stays running until it succeeds or the sequence is cancelled; the failure is logged,
+again when the error changes and every 5 minutes. Another error, e.g. a value refused by the unit (an exception
+response other than Server Device Busy, Acknowledge or of a gateway), ends the sequence and it is logged; the final
+state of the sequence, if it has one (`final`, the end of a pulse), is written before the sequence ends, written
+again as a step on an error of the communication. When Evok stops while the last step is written and it fails, the final state
+is written. After a crash of Evok
 the output stays in its state.
 
 ## State

src/evok/blocks/sequencer.py

@@ -14,12 +14,49 @@
 
 import asyncio
 import logging
+import time
 from dataclasses import dataclass
 
+from tmodbus.exceptions import (
+    AcknowledgeError,
+    GatewayPathUnavailableError,
+    GatewayTargetDeviceFailedToRespondError,
+    InvalidRequestError,
+    ModbusResponseError,
+    ServerDeviceBusyError,
+    TModbusError,
+)
+
 from ..devices import devents, to_bool, to_float
+from ..errors import UnitCommunicationError, describe_error, root_cause
 
 logger = logging.getLogger(__name__)
 
+STEP_RETRY_DELAY = 1.0  # s, a failed step is written again after it, the output does not stay in its state
+STEP_RETRY_MAX_DELAY = 30.0  # s, the delay doubles up to it, a missing unit does not load its bus
+STEP_FAILURE_LOG = 300.0  # s, a step still failing is logged again after it, also a changed error at once
+# the errors of the communication, they pass; another error of a step ends its sequence and writes its final
+# state, e.g. a value refused by the device, it would be written again forever and the output refused the changes
+STEP_RETRIED = (UnitCommunicationError, TModbusError, TimeoutError, OSError)
+# the request refused by the unit (an exception response, e.g. Illegal Data Value or Server Device Failure)
+# or invalid, it would be refused again
+STEP_REFUSED = (ModbusResponseError, InvalidRequestError)
+# the exception responses which pass, e.g. a busy unit or a gateway without its target
+STEP_TRANSIENT = (
+    ServerDeviceBusyError,
+    AcknowledgeError,
+    GatewayPathUnavailableError,
+    GatewayTargetDeviceFailedToRespondError,
+)
+
+
+def _retried(error: Exception) -> bool:
+    """An error of the communication, not a request refused by the unit"""
+    cause = root_cause(error)
+    return isinstance(error, STEP_RETRIED) and (
+        not isinstance(cause, STEP_REFUSED) or isinstance(cause, STEP_TRANSIENT)
+    )
+
 
 @dataclass
 class Step:
@@ -66,6 +103,7 @@ class Running:
     task: asyncio.Task
     sequence: Sequence
     step: int = 0  # the number of the done steps
+    finished: bool = False  # the last step was written, the sequence ended by it
 
 
 class Sequencer:
@@ -141,12 +179,12 @@ def _end(self, device, task: asyncio.Task) -> bool:
 
     async def _run(self, device, sequence: Sequence):
         task = asyncio.current_task()
-        step_number = 0
+        step_number, step = 0, None
         try:
             for step_number, step in enumerate(sequence.steps, 1):
                 await asyncio.sleep(step.delay)
                 last = step_number == len(sequence.steps)
-                if not await self._apply(device, step.params, task, step_number, last):
+                if not await self._apply_step(device, sequence, step, task, step_number, last):
                     return
                 if not last:
                     if not self._current(device, task):
@@ -156,10 +194,63 @@ async def _run(self, device, sequence: Sequence):
         except asyncio.CancelledError:
             raise
         except Exception:
-            logger.exception(f"{_name(device)}: step {step_number} of the {sequence.name} failed")
+            logger.exception(f"{_name(device)}: step {step_number} of the {sequence.name} failed, it is ended")
+            # not the refused step again, e.g. the end of a pulse is its final state
+            refused = step.params if step is not None else None
+            if sequence.final is not None and sequence.final != refused and self._current(device, task):
+                # the output does not stay in the state of the sequence: the final state is written as its last
+                # step, the sequence runs until it is written (a change of the output meanwhile is refused,
+                # finish_all() waits for it), an error of the communication is written again
+                try:
+                    await self._apply_step(
+                        device, sequence, Step(sequence.final), task, step_number - 1, True, what="final state"
+                    )
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    logger.exception(f"{_name(device)}: final state of the {sequence.name} failed")
         finally:
             self._end(device, task)
 
+    async def _apply_step(
+        self, device, sequence: Sequence, step: Step, task, step_number: int, last: bool, what: str | None = None
+    ) -> bool:
+        """Write the step until it succeeds, the sequence stays running meanwhile, e.g. a noise on the bus at
+        the end of a pulse; False if the sequence was cancelled or replaced meanwhile
+        """
+        what = what or f"step {step_number}"
+        failures, delay = 0, STEP_RETRY_DELAY
+        logged, logged_at = None, 0.0
+        while True:
+            try:
+                applied = await self._apply(device, step.params, task, step_number, last)
+            except Exception as E:
+                if not _retried(E):
+                    raise
+                if not self._current(device, task):
+                    return False
+                failures += 1
+                # the kind of the error, its text may differ by each attempt
+                now, kind = time.monotonic(), type(root_cause(E))
+                if logged is None:
+                    logger.exception(
+                        f"{_name(device)}: {what} of the {sequence.name} failed, it is written again "
+                        f"after {STEP_RETRY_DELAY}..{STEP_RETRY_MAX_DELAY} s"
+                    )
+                    logged, logged_at = kind, now
+                elif kind != logged or now - logged_at >= STEP_FAILURE_LOG:
+                    # a changed error, or a reminder that the output is still in the state of the sequence
+                    logger.error(
+                        f"{_name(device)}: {what} of the {sequence.name} failed {failures}x: {describe_error(E)}"
+                    )
+                    logged, logged_at = kind, now
+                await asyncio.sleep(delay)
+                delay = min(delay * 2, STEP_RETRY_MAX_DELAY)
+                continue
+            if failures and applied:
+                logger.info(f"{_name(device)}: {what} of the {sequence.name} written after {failures} failures")
+            return applied
+
     async def _apply(self, device, params: dict, task: asyncio.Task | None = None, step_number=0, last=False) -> bool:
         """Set the params of a step, False if the sequence was cancelled or replaced meanwhile"""
         applied = False
@@ -170,15 +261,18 @@ async def operation():
                 entry = self._running.get(device)
                 if entry is None or entry.task is not task:
                     return
-                entry.step = step_number
-                if last:
-                    # the state read after the write is not running
-                    self._end(device, task)
                 self._writing.add(task)
             try:
                 await device.set(**params)
             finally:
                 self._writing.discard(task)
+            if task is not None:
+                # done after the successful write, a failed step is written again and it is not reported as done
+                entry.step = step_number
+                if last:
+                    # the state read after the write is not running; finish_all() does not write the final state
+                    entry.finished = True
+                    self._end(device, task)
             applied = True
 
         change = getattr(getattr(device, "client", None), "change", None)
@@ -199,15 +293,18 @@ async def finish_all(self, timeout: float = 5.0):
         for device, entry in entries:
             self._stop(entry)
             self._notify(device)
-        finals = [(device, entry.sequence) for device, entry in entries if entry.sequence.final is not None]
+        candidates = [(device, entry) for device, entry in entries if entry.sequence.final is not None]
         writing = list(self._writing)
-        if not finals and not writing:
+        if not candidates and not writing:
             return
-        logger.info(f"Finishing {len(finals)} running sequences")
+        logger.info(f"Finishing {len(candidates)} running sequences")
 
         async def finish():
             if writing:
                 await asyncio.wait(writing)
+            # a sequence whose last step was written ended by it; a failed write of the last step is written
+            # by the final state
+            finals = [(device, entry.sequence) for device, entry in candidates if not entry.finished]
             await asyncio.gather(*(self._finish(device, sequence) for device, sequence in finals))
 
         try:

src/evok/devices.py

@@ -88,9 +88,10 @@ def set_force_save(self) -> None:
         if self.save_callback:
             self.save_callback()
 
-    def validate(self, alias: str, device: Device) -> None:
+    def validate(self, alias: str, device: Device, free=()) -> None:
+        """Raise ValueError for an invalid alias; free are the aliases released by the change before it"""
         # check duplicity
-        if alias in self.alias_dict:
+        if alias in self.alias_dict and alias not in free:
             raise ValueError(f"Duplicate alias {alias}")
         # an alias loaded from the file belongs to its device, even when the device is not registered yet
         rec = self.initial_dict.get(alias)
@@ -228,6 +229,20 @@ def register_device(self, devtype_name, device):
         devents.config(device)
         logger.debug(f"Registered new device '{devtype_name}' with circuit {device.circuit} \t ({device})")
 
+    def check_alias(self, alias: str | None, device: Device, free=()) -> None:
+        """Raise ValueError if set_alias() would refuse the alias, before any change of the device;
+        free are the aliases released by the changes before it, e.g. an alias moved to another device
+        """
+        if alias in (None, "") or alias == device.alias:
+            return
+        if not isinstance(alias, str):
+            # e.g. 0 or false of a request without the schema, it is not a reset of the alias
+            raise ValueError(f"Invalid alias {alias!r}")
+        # by_name() finds the circuit first, such alias would never be used
+        if alias in self[device.devtype]:
+            raise ValueError(f"Alias {alias} is a circuit of {device.devtype}")
+        self.aliases.validate(alias, device, free)
+
     def set_alias(self, alias: str | None, device: Device, file_update: bool = False) -> None:
         """Set or reset ('' or None) the alias, an invalid alias raises ValueError logged by the caller"""
         if alias == device.alias:
@@ -239,9 +254,7 @@ def set_alias(self, alias: str | None, device: Device, file_update: bool = False
             device.alias = ""
             logger.debug(f"Reset alias of {device.devtype}[{device.circuit}]")
         else:
-            # by_name() finds the circuit first, such alias would never be used
-            if alias in self[device.devtype]:
-                raise ValueError(f"Alias {alias} is a circuit of {device.devtype}")
+            self.check_alias(alias, device)
             self.aliases.add(alias, device, file_update)
             device.alias = alias
             logger.debug(f"Set alias {alias} of {device.devtype}[{device.circuit}]")

src/evok/modbus/base.py

@@ -80,6 +80,8 @@ async def set_devices(
         unit = device.client if isinstance(device, IODevice) else device
         units.setdefault(unit, []).append((index, device, kw))
 
+    _check_aliases(units)
+
     for unit, items in units.items():
         done = []
 
@@ -108,3 +110,33 @@ async def operation():
             for index, device in done:
                 states[index] = device.full()
     return [states[index] for index in range(len(assignments))]
+
+
+def _check_aliases(units: dict):
+    """Check the aliases of the assignments in the order of their changes, before the first write: an invalid
+    alias was refused after the write of the value, the output was switched and the request failed. An alias
+    released by an earlier assignment can be taken, an alias taken by an earlier one cannot.
+    """
+    planned: dict = {}  # the alias of a device after the earlier assignments
+    free: set[str] = set()
+    taken: set[str] = set()
+    for items in units.values():
+        for _index, device, kw in items:
+            # None is not given, set_alias() keeps the alias; "" resets it
+            if kw.get("alias") is None:
+                continue
+            alias = kw["alias"]
+            current = planned.get(device, device.alias)
+            if alias == current:
+                continue
+            if alias != "":
+                # first, an alias which is not a string is invalid, e.g. a list is not hashable for taken
+                Devices.check_alias(alias, device, free)
+                if alias in taken:
+                    raise ValueError(f"Duplicate alias {alias}")
+                taken.add(alias)
+                free.discard(alias)
+            if current:
+                free.add(current)
+                taken.discard(current)
+            planned[device] = alias

src/evok/modbus/client.py

@@ -439,7 +439,7 @@ def read_raw(self, client: Client) -> int:
     async def write_raw(self, client: Client, raw: int):
         """Read-modify-write, the register is read from the unit, not from the cache"""
         curr = (await client.cache.get_register_async(self.index))[0]
-        await client.write_registers(self.index, [curr | self.mask if raw else curr & ~self.mask])
+        await client.write_registers(self.index, [curr | self.mask if raw else curr & ~self.mask], pin=self.pin)
 
 
 class Accessor32(Accessor):