53def run(coro: Coroutine[Any, Any, None]) -> None:
54    """`finsy.run` provides a useful wrapper around `asyncio.run`.
55
56    This function implements common boilerplate for running a Finsy application.
57
58    - Set up basic logging to stderr at the INFO log level.
59    - Set up a signal handler for SIGTERM that shuts down gracefully.
60    - Set up caching of P4Info data so common definitions are re-used.
61
62    Example:
63
64    ```python
65    import finsy as fy
66
67    async def main():
68        async with fy.Switch("sw", "127.0.0.1:50001") as sw:
69            print(sw.p4info)
70
71    if __name__ == "__main__":
72        fy.run(main())
73    ```
74
75    If you choose to use `asyncio.run` instead, your P4Schema/P4Info objects
76    will not be eligible for sharing. You can create your own `P4SchemaCache`
77    context manager to implement this.
78    """
79    try:
80        asyncio.run(_finsy_boilerplate(coro))
81    except (KeyboardInterrupt, asyncio.CancelledError):
82        pass

 30@final
 31class Controller:
 32    """Represents a collection of P4Runtime switches.
 33
 34    Each `Switch` in the Controller is identified by its name. Each name must
 35    be unique.
 36
 37    ```
 38    switches = [
 39        fy.Switch("sw1", "10.0.0.1:50000"),
 40        fy.Switch("sw2", "10.0.0.2:50000"),
 41    ]
 42    controller = fy.Controller(switches)
 43    await controller.run()
 44    ```
 45
 46    A Controller can be running or stopped. There are two ways to run a
 47    Controller. You can use the `Controller.run()` method, or you can use
 48    a Controller as a context manager.
 49
 50    ```
 51    controller = fy.Controller(switches)
 52    async with controller:
 53        # Let controller run for 60 seconds.
 54        await asyncio.sleep(60)
 55    ```
 56
 57    You can `add` or `remove` switches regardless of whether a Controller is
 58    running or not. If the Controller is not running, adding or removing a Switch is
 59    instantaneous. If the Controller is running, adding a Switch will start
 60    it running asynchronously. Removing a Switch will schedule the Switch to stop,
 61    but defer actual removal until the Switch has stopped asynchronously.
 62
 63    When a switch starts inside a Controller, it fires the `CONTROLLER_ENTER`
 64    event. When a switch stops inside a Controller, it fires the `CONTROLLER_LEAVE`
 65    event.
 66
 67    A Controller supports these methods to access its contents:
 68
 69    - len(controller): Return number of switches in the controller.
 70    - controller[name]: Return the switch with the given name.
 71    - controller.get(name): Return the switch with the given name, or None if not found.
 72
 73    You can iterate over the switches in a Controller using a for loop:
 74
 75    ```
 76    for switch in controller:
 77        print(switch.name)
 78    ```
 79
 80    Any task or sub-task running inside a controller can retrieve its
 81    Controller object using the `Controller.current()` method.
 82    """
 83
 84    _switches: dict[str, Switch]
 85    _pending_removal: set[Switch]
 86    _task_count: CountdownFuture
 87
 88    _control_task: asyncio.Task[Any] | None = None
 89    "Keep track of controller's main task."
 90
 91    def __init__(self, switches: Iterable[Switch] = ()):
 92        """Initialize Controller object with an initial list of switches.
 93
 94        Args:
 95            switches: a collection or other iterable of Switch objects.
 96        """
 97        self._switches = {}
 98        self._pending_removal = set()
 99        self._task_count = CountdownFuture()
100
101        for switch in switches:
102            if switch.name in self._switches:
103                raise ValueError(f"Switch named {switch.name!r} already exists")
104            self._switches[switch.name] = switch
105
106    @property
107    def running(self) -> bool:
108        "True if Controller is running."
109        return self._control_task is not None
110
111    async def run(self) -> None:
112        "Run the controller."
113        async with self:
114            await wait_for_cancel()
115
116    def stop(self) -> None:
117        "Stop the controller if it is running."
118        if self._control_task is not None:
119            self._control_task.cancel()
120
121    async def __aenter__(self) -> Self:
122        "Run the controller as a context manager (see also run())."
123        assert not self.running, "Controller.__aenter__ is not re-entrant"
124        assert self._task_count.value() == 0
125        assert not self._pending_removal
126
127        self._control_task = asyncio.current_task()
128        _CONTROLLER.set(self)
129
130        try:
131            # Start each switch running.
132            for switch in self:
133                self._start_switch(switch)
134        except Exception:
135            self._control_task = None
136            _CONTROLLER.set(None)
137            raise
138
139        return self
140
141    async def __aexit__(
142        self,
143        _exc_type: type[BaseException] | None,
144        _exc_val: BaseException | None,
145        _exc_tb: TracebackType | None,
146    ) -> bool | None:
147        "Run the controller as a context manager (see also run())."
148        assert self.running
149
150        try:
151            # Stop all the switches.
152            for switch in self:
153                self._stop_switch(switch)
154
155            # Wait for switch tasks to finish.
156            await self._task_count.wait()
157
158        finally:
159            self._control_task = None
160            _CONTROLLER.set(None)
161
162    def add(self, switch: Switch) -> None:
163        """Add a switch to the controller.
164
165        If the controller is running, tell the switch to start.
166
167        Args:
168            switch: the Switch object.
169        """
170        if switch.name in self._switches:
171            raise ValueError(f"Switch named {switch.name!r} already exists")
172
173        self._switches[switch.name] = switch
174        if self.running:
175            self._start_switch(switch)
176
177    def remove(self, switch: Switch) -> asyncio.Event:
178        """Remove a switch from the controller.
179
180        If the controller is running, tell the switch to stop and schedule it
181        for removal when it fully stops.
182
183        Args:
184            switch: the Switch object.
185        """
186        name = switch.name
187        if self._switches.get(name, None) is not switch:
188            raise ValueError(f"Switch named {name!r} not found")
189
190        del self._switches[name]
191
192        event = asyncio.Event()
193        if self.running:
194            # When controller is running, event will complete when switch
195            # is actually stopped.
196            self._stop_switch(switch)
197            self._pending_removal.add(switch)
198
199            def _controller_leave(sw: Switch):
200                self._pending_removal.discard(sw)
201                event.set()
202
203            switch.ee.once(SwitchEvent.CONTROLLER_LEAVE, _controller_leave)  # type: ignore
204        else:
205            # When controller is not running, event completes immediately.
206            event.set()
207
208        return event
209
210    def _start_switch(self, switch: Switch):
211        "Start the switch's control task."
212        LOGGER.debug("Controller._start_switch: %r", switch)
213        assert switch._control_task is None  # pyright: ignore[reportPrivateUsage]
214
215        switch.ee.emit(SwitchEvent.CONTROLLER_ENTER, switch)
216
217        task = asyncio.create_task(switch.run(), name=f"fy:{switch.name}")
218        switch._control_task = task  # pyright: ignore[reportPrivateUsage]
219        self._task_count.increment()
220
221        def _switch_done(done: asyncio.Task[Any]):
222            switch._control_task = None  # pyright: ignore[reportPrivateUsage]
223            switch.ee.emit(SwitchEvent.CONTROLLER_LEAVE, switch)
224            self._task_count.decrement()
225
226            if not done.cancelled():
227                ex = done.exception()
228                if ex is not None:
229                    if not isinstance(ex, SwitchFailFastError):
230                        # The `fail_fast` error has already been logged. If
231                        # it's any other error, log it. (There shouldn't be
232                        # any other error.)
233                        LOGGER.critical(
234                            "Controller task %r failed",
235                            done.get_name(),
236                            exc_info=ex,
237                        )
238                    # Shutdown the program cleanly due to switch failure.
239                    raise SystemExit(99)
240
241        task.add_done_callback(_switch_done)
242
243    def _stop_switch(self, switch: Switch):
244        "Stop the switch's control task."
245        LOGGER.debug("Controller._stop_switch: %r", switch)
246
247        if switch._control_task is not None:  # pyright: ignore[reportPrivateUsage]
248            switch._control_task.cancel()  # pyright: ignore[reportPrivateUsage]
249
250    def __len__(self) -> int:
251        "Return the number of switches."
252        return len(self._switches)
253
254    def __iter__(self) -> Iterator[Switch]:
255        "Iterate over the switches."
256        return iter(self._switches.values())
257
258    def __getitem__(self, name: str) -> Switch:
259        "Retrieve a switch by name."
260        return self._switches[name]
261
262    def get(self, name: str) -> Switch | None:
263        "Retrieve a switch by name, or return None if not found."
264        return self._switches.get(name)
265
266    @staticmethod
267    def current() -> "Controller":
268        "Return the current Controller object."
269        result = _CONTROLLER.get()
270        if result is None:
271            raise RuntimeError("controller does not exist")
272        return result

68class LoggerAdapter(_BaseLoggerAdapter):
69    """Custom log adapter to include the name of the current task."""
70
71    def process(
72        self,
73        msg: Any,
74        kwargs: MutableMapping[str, Any],
75    ) -> tuple[Any, MutableMapping[str, Any]]:
76        """Process the logging message and keyword arguments passed in to a
77        logging call to insert contextual information.
78        """
79        task_name = _get_current_task_name()
80        return f"[{task_name}] {msg}", kwargs
81
82    def info(self, msg: Any, *args: Any, **kwargs: Any) -> None:
83        """INFO level uses a concise task name representation for readability."""
84        if self.logger.isEnabledFor(logging.INFO):
85            task_name = _get_current_task_name(True)
86            self.logger.info(f"[{task_name}] {msg}", *args, **kwargs)

 24@functools.total_ordering
 25class MACAddress:
 26    """Concrete class for a MAC address."""
 27
 28    __slots__ = ("__weakref__", "_mac")
 29    _mac: int
 30
 31    def __init__(self, value: object) -> None:
 32        "Construct a MAC address from a string, integer or bytes object."
 33        match value:
 34            case int():
 35                self._mac = _from_int(value)
 36            case bytes():
 37                self._mac = _from_bytes(value)
 38            case MACAddress():
 39                self._mac = value._mac
 40            case _:
 41                # Input argument is a MAC string, or an object that is
 42                # formatted as MAC string (behave like ipaddress module).
 43                self._mac = _from_string(str(value))
 44
 45    @property
 46    def packed(self) -> bytes:
 47        "Return the MAC address a byte string."
 48        return _to_bytes(self._mac)
 49
 50    @property
 51    def max_prefixlen(self) -> int:
 52        "Return the maximum prefix length (48 bits)."
 53        return _BIT_WIDTH
 54
 55    @property
 56    def is_multicast(self) -> bool:
 57        "Return true if MAC address has the multicast bit set."
 58        return self._mac & (1 << 40) != 0
 59
 60    @property
 61    def is_private(self) -> bool:
 62        "Return true if MAC address has the locally administered bit set."
 63        return self._mac & (1 << 41) != 0
 64
 65    @property
 66    def is_global(self) -> bool:
 67        "Return true if the locally administered bit is not set."
 68        return not self.is_private
 69
 70    @property
 71    def is_unspecified(self) -> bool:
 72        "Return true if MAC address is all zeros."
 73        return self._mac == 0
 74
 75    @property
 76    def is_broadcast(self) -> bool:
 77        "Return true if MAC address is the broadcast address."
 78        return self._mac == (1 << _BIT_WIDTH) - 1
 79
 80    def __int__(self) -> int:
 81        return self._mac
 82
 83    def __eq__(self, rhs: object) -> bool:
 84        if not isinstance(rhs, MACAddress):
 85            return NotImplemented
 86        return self._mac == rhs._mac
 87
 88    def __lt__(self, rhs: object) -> bool:
 89        if not isinstance(rhs, MACAddress):
 90            return NotImplemented
 91        return self._mac < rhs._mac
 92
 93    def __repr__(self) -> str:
 94        return f"MACAddress({_to_string(self._mac)!r})"
 95
 96    def __str__(self) -> str:
 97        return _to_string(self._mac)
 98
 99    def __hash__(self) -> int:
100        return hash(hex(self._mac))

1510@decodable("action_profile_group")
1511@dataclass(slots=True)
1512class P4ActionProfileGroup(_P4Writable):
1513    "Represents a P4Runtime ActionProfileGroup."
1514
1515    action_profile_id: str = ""
1516    _: KW_ONLY
1517    group_id: int = 0
1518    max_size: int = 0
1519    members: Sequence[P4Member] | None = None
1520
1521    def encode(self, schema: P4Schema) -> p4r.Entity:
1522        "Encode P4ActionProfileGroup as protobuf."
1523        if not self.action_profile_id:
1524            return p4r.Entity(action_profile_group=p4r.ActionProfileGroup())
1525
1526        profile = schema.action_profiles[self.action_profile_id]
1527
1528        if self.members is not None:
1529            members = [member.encode() for member in self.members]
1530        else:
1531            members = None
1532
1533        entry = p4r.ActionProfileGroup(
1534            action_profile_id=profile.id,
1535            group_id=self.group_id,
1536            members=members,
1537            max_size=self.max_size,
1538        )
1539        return p4r.Entity(action_profile_group=entry)
1540
1541    @classmethod
1542    def decode(cls, msg: p4r.Entity, schema: P4Schema) -> Self:
1543        "Decode protobuf to ActionProfileGroup data."
1544        entry = msg.action_profile_group
1545        if entry.action_profile_id == 0:
1546            return cls()
1547
1548        profile = schema.action_profiles[entry.action_profile_id]
1549
1550        if entry.members:
1551            members = [P4Member.decode(member) for member in entry.members]
1552        else:
1553            members = None
1554
1555        return cls(
1556            action_profile_id=profile.alias,
1557            group_id=entry.group_id,
1558            max_size=entry.max_size,
1559            members=members,
1560        )
1561
1562    def action_str(self, _schema: P4Schema) -> str:
1563        "Return string representation of the weighted members."
1564        if not self.members:
1565            return ""
1566
1567        return " ".join(
1568            [f"{member.weight}*{member.member_id:#x}" for member in self.members]
1569        )

1408@decodable("action_profile_member")
1409@dataclass(slots=True)
1410class P4ActionProfileMember(_P4Writable):
1411    "Represents a P4Runtime ActionProfileMember."
1412
1413    action_profile_id: str = ""
1414    _: KW_ONLY
1415    member_id: int = 0
1416    action: P4TableAction | None = None
1417
1418    def encode(self, schema: P4Schema) -> p4r.Entity:
1419        "Encode P4ActionProfileMember as protobuf."
1420        if not self.action_profile_id:
1421            return p4r.Entity(action_profile_member=p4r.ActionProfileMember())
1422
1423        profile = schema.action_profiles[self.action_profile_id]
1424
1425        if self.action:
1426            action = self.action.encode_action(schema)
1427        else:
1428            action = None
1429
1430        entry = p4r.ActionProfileMember(
1431            action_profile_id=profile.id,
1432            member_id=self.member_id,
1433            action=action,
1434        )
1435        return p4r.Entity(action_profile_member=entry)
1436
1437    @classmethod
1438    def decode(cls, msg: p4r.Entity, schema: P4Schema) -> Self:
1439        "Decode protobuf to ActionProfileMember data."
1440        entry = msg.action_profile_member
1441        if entry.action_profile_id == 0:
1442            return cls()
1443
1444        profile = schema.action_profiles[entry.action_profile_id]
1445
1446        if entry.HasField("action"):
1447            action = P4TableAction.decode_action(entry.action, schema)
1448        else:
1449            action = None
1450
1451        return cls(
1452            action_profile_id=profile.alias,
1453            member_id=entry.member_id,
1454            action=action,
1455        )
1456
1457    def action_str(self, schema: P4Schema) -> str:
1458        "Format the action as a human-readable, canonical string."
1459        if self.action is None:
1460            return NOACTION_STR
1461        return self.action.format_str(schema)