twisteroidambassador

import ipaddress
import typing
import abc

IPNetworkType = typing.TypeVar('IPNetworkType', ipaddress.IPv4Network, ipaddress.IPv6Network)


class IPSubnetSequence(typing.Sequence[IPNetworkType]):
    """An indexable sequence of subnets of a given network.

twisteroidambassador

import asyncio


class CancelledFromOutside(asyncio.CancelledError):
    pass


class CancelledFromInside(asyncio.CancelledError):
    pass

twisteroidambassador

"""Accessing TCP metrics in Linux using Generic Netlink via gnlpy."""

import array
import ipaddress
import struct

from pprint import pprint
from typing import Union

import gnlpy.netlink as netlink

twisteroidambassador

import asyncio

class Barrier():
    """asyncio version of threading.Barrier (partial).
    
    Implements wait() only, without support for timeouts."""
    
    def __init__(self, parties, loop=None):
        if loop is None:
            self._loop = asyncio.get_event_loop()