jmri_logixng.actions.flow

Flow-control actions: sequence + delay.

 1"""Flow-control actions: sequence + delay."""
 2from __future__ import annotations
 3
 4import asyncio
 5from typing import Any, ClassVar
 6
 7from jmri_logixng.context import ExecutionContext
 8from jmri_logixng.registry import register
 9from jmri_logixng.tree import Action
10
11
12@register
13class Sequence(Action):
14    """Run a list of actions in order."""
15
16    type_tag: ClassVar[str] = "action:sequence"
17
18    def __init__(
19        self,
20        actions: list[Action],
21        *,
22        user_name: str = "",
23        node_id: str | None = None,
24    ) -> None:
25        super().__init__(user_name=user_name, node_id=node_id)
26        self.actions = list(actions)
27
28    async def execute(self, ctx: ExecutionContext) -> None:
29        for action in self.actions:
30            await action.execute(ctx)
31
32    def extra_fields(self) -> dict[str, Any]:
33        return {"actions": [a.to_dict() for a in self.actions]}
34
35
36@register
37class Delay(Action):
38    """Sleep for *seconds*. Useful for timed sequences (signal aspect change → hold)."""
39
40    type_tag: ClassVar[str] = "action:delay"
41
42    def __init__(
43        self,
44        *,
45        seconds: float,
46        user_name: str = "",
47        node_id: str | None = None,
48    ) -> None:
49        super().__init__(user_name=user_name, node_id=node_id)
50        if seconds < 0:
51            raise ValueError(f"delay seconds must be non-negative, got {seconds}")
52        self.seconds = seconds
53
54    async def execute(self, ctx: ExecutionContext) -> None:
55        ctx.emit_trace(f"delay {self.seconds}s")
56        await asyncio.sleep(self.seconds)
57
58    def extra_fields(self) -> dict[str, Any]:
59        return {"seconds": self.seconds}
@register
class Sequence(jmri_logixng.tree.Action): 
13@register
14class Sequence(Action):
15    """Run a list of actions in order."""
16
17    type_tag: ClassVar[str] = "action:sequence"
18
19    def __init__(
20        self,
21        actions: list[Action],
22        *,
23        user_name: str = "",
24        node_id: str | None = None,
25    ) -> None:
26        super().__init__(user_name=user_name, node_id=node_id)
27        self.actions = list(actions)
28
29    async def execute(self, ctx: ExecutionContext) -> None:
30        for action in self.actions:
31            await action.execute(ctx)
32
33    def extra_fields(self) -> dict[str, Any]:
34        return {"actions": [a.to_dict() for a in self.actions]}

Run a list of actions in order.

Sequence( actions: list[jmri_logixng.Action], *, user_name: str = '', node_id: str | None = None) 
19    def __init__(
20        self,
21        actions: list[Action],
22        *,
23        user_name: str = "",
24        node_id: str | None = None,
25    ) -> None:
26        super().__init__(user_name=user_name, node_id=node_id)
27        self.actions = list(actions)
type_tag: ClassVar[str] = 'action:sequence'
actions
async def execute(self, ctx: jmri_logixng.ExecutionContext) -> None: 
29    async def execute(self, ctx: ExecutionContext) -> None:
30        for action in self.actions:
31            await action.execute(ctx)
def extra_fields(self) -> dict[str, typing.Any]: 
33    def extra_fields(self) -> dict[str, Any]:
34        return {"actions": [a.to_dict() for a in self.actions]}

Hook for subclasses to add their own fields to to_dict.

Inherited Members
jmri_logixng.tree.Node
user_name
id
to_dict
@register
class Delay(jmri_logixng.tree.Action): 
37@register
38class Delay(Action):
39    """Sleep for *seconds*. Useful for timed sequences (signal aspect change → hold)."""
40
41    type_tag: ClassVar[str] = "action:delay"
42
43    def __init__(
44        self,
45        *,
46        seconds: float,
47        user_name: str = "",
48        node_id: str | None = None,
49    ) -> None:
50        super().__init__(user_name=user_name, node_id=node_id)
51        if seconds < 0:
52            raise ValueError(f"delay seconds must be non-negative, got {seconds}")
53        self.seconds = seconds
54
55    async def execute(self, ctx: ExecutionContext) -> None:
56        ctx.emit_trace(f"delay {self.seconds}s")
57        await asyncio.sleep(self.seconds)
58
59    def extra_fields(self) -> dict[str, Any]:
60        return {"seconds": self.seconds}

Sleep for seconds. Useful for timed sequences (signal aspect change → hold).

Delay(*, seconds: float, user_name: str = '', node_id: str | None = None) 
43    def __init__(
44        self,
45        *,
46        seconds: float,
47        user_name: str = "",
48        node_id: str | None = None,
49    ) -> None:
50        super().__init__(user_name=user_name, node_id=node_id)
51        if seconds < 0:
52            raise ValueError(f"delay seconds must be non-negative, got {seconds}")
53        self.seconds = seconds
type_tag: ClassVar[str] = 'action:delay'
seconds
async def execute(self, ctx: jmri_logixng.ExecutionContext) -> None: 
55    async def execute(self, ctx: ExecutionContext) -> None:
56        ctx.emit_trace(f"delay {self.seconds}s")
57        await asyncio.sleep(self.seconds)
def extra_fields(self) -> dict[str, typing.Any]: 
59    def extra_fields(self) -> dict[str, Any]:
60        return {"seconds": self.seconds}

Hook for subclasses to add their own fields to to_dict.

Inherited Members
jmri_logixng.tree.Node
user_name
id
to_dict