Update app.py

app.py

@@ -1,333 +1,2181 @@
-#!/usr/bin/env python
-# coding: utf-8
-import json
-import flask
-import os
-import re
-import time
-from random import random
+import array
+import asyncio
+import concurrent.futures
+import math
 import socket
-
+import sys
+from asyncio.base_events import _run_until_complete_cb  # type: ignore[attr-defined]
+from collections import OrderedDict, deque
+from concurrent.futures import Future
+from contextvars import Context, copy_context
+from dataclasses import dataclass
+from functools import partial, wraps
+from inspect import (
+    CORO_RUNNING,
+    CORO_SUSPENDED,
+    GEN_RUNNING,
+    GEN_SUSPENDED,
+    getcoroutinestate,
+    getgeneratorstate,
+)
+from io import IOBase
+from os import PathLike
+from queue import Queue
+from socket import AddressFamily, SocketKind
 from threading import Thread
-from time import sleep
-
-test_html = '''
-<!-- Header -->
-<header class="w3-display-container w3-content w3-wide" style="max-width:1500px;" id="home">
-  <img class="w3-image" src="https://cdn.pixabay.com/photo/2018/12/10/16/22/city-3867295_960_720.png" alt="Architecture" width="1500" height="800">
-  <div class="w3-display-middle w3-margin-top w3-center">
-    <h1 class="w3-xxlarge w3-text-white"><span class="w3-padding w3-black w3-opacity-min"><b>WATCH</b></span> <span class="w3-hide-small w3-text-dark-grey">Tower</span></h1>
-  </div>
-</header>
-
-<!-- Container (About Section) -->
-<div class="w3-content w3-container w3-padding-64" id="about">
-  <h3 class="w3-center">Block Violent Content Before It Reaches Your Feed</h3>
-  <p class="w3-center"><em>WatchTower identifies, blocks, and filters out violent and radical content before it reaches your Twitter feed.
-</em></p>
-<br>
-  <p>WatchTower works to protect you from violent, misinformation, hate speech and other malicious communication by using a suite of machine learning models to identify user accounts that post content that commonly falls into these categories. WatchTower is broken down into two components, the first utilises the Twitter streaming API and applies a suite of machine learning models to identify users that commonly post malicious information, while the second element provides a web UI where users can authenticaate with Twitter and tailor the types and thresholds for the accounts they block.  </p>
-  <br>
-  <p> WatchTower was developed solely by James Stevenson and primarily uses Pinpoint, a machine learning model also developed by James. The future roadmap sees WatchTower incoperate other models for identifying contrent such as misinformation and hate speech. More on Pinpoint and the model WatchTower uses to identify violent extremism can be seen below.</p>
-
-  <p class="w3-large w3-center w3-padding-16">Model Accuracy:</p>
-  <p class="w3-center"><em>Machine learning models can be validated based on several statistics. These statistics for Pinpoint the main ML model used by WatchTower can be seen below. </p>
-  <br>
-  <p class="w3-wide"><i class="fa fa-camera"></i>Accuracy</p>
-  <div class="w3-light-grey">
-    <div class="w3-container w3-padding-small w3-dark-grey w3-center" style="width:73%">73%</div>
-  </div>
-  <p class="w3-wide"><i class="fa fa-laptop"></i>Recall</p>
-  <div class="w3-light-grey">
-    <div class="w3-container w3-padding-small w3-dark-grey w3-center" style="width:62%">62%</div>
-  </div>
-  <p class="w3-wide"><i class="fa fa-photo"></i>Precision</p>
-  <div class="w3-light-grey">
-    <div class="w3-container w3-padding-small w3-dark-grey w3-center" style="width:78%">78%</div>
-  </div>
-  <p class="w3-wide"><i class="fa fa-photo"></i>F-Measure</p>
-  <div class="w3-light-grey">
-    <div class="w3-container w3-padding-small w3-dark-grey w3-center" style="width:69%">69%</div>
-  </div>
-</div>
-
-<div class="w3-row w3-center w3-dark-grey w3-padding-16">
-  <div class="w3-quarter w3-section">
-    <span class="w3-xlarge">14+</span><br>
-    Partners
-  </div>
-  <div class="w3-quarter w3-section">
-    <span class="w3-xlarge">55+</span><br>
-    Projects Done
-  </div>
-  <div class="w3-quarter w3-section">
-    <span class="w3-xlarge">89+</span><br>
-    Happy Clients
-  </div>
-  <div class="w3-quarter w3-section">
-    <span class="w3-xlarge">150+</span><br>
-    Meetings
-  </div>
-</div>
-<br>
-<!-- Container (Portfolio Section) -->
-<div class="w3-content w3-container w3-padding-64" id="portfolio">
-  <h3 class="w3-center">Chirp Development Challenge 2022</h3>
-  <p class="w3-center"><em>WatchTower was developed for the Chirp 2022 Twitter API Developer Challenge</em></p>
-</div><p> Watchtower was developed solely by James Stevenson for the  Chirp 2022 Twitter API Developer Challenge. More infomration of this can be found below.</p>
-<br>
-<img class="w3-image" src="https://cdn.cms-twdigitalassets.com/content/dam/developer-twitter/redesign-2021-images/blog2022/chirp/Chirp-Hero-Banner.jpg.twimg.1920.jpg" alt="Architecture" width="1500" height="800">
-<br>
-<!-- Modal for full size images on click-->
-<div id="modal01" class="w3-modal w3-black" onclick="this.style.display='none'">
-  <span class="w3-button w3-large w3-black w3-display-topright" title="Close Modal Image"><i class="fa fa-remove"></i></span>
-  <div class="w3-modal-content w3-animate-zoom w3-center w3-transparent w3-padding-64">
-    <img id="img01" class="w3-image">
-    <p id="caption" class="w3-opacity w3-large"></p>
-  </div>
-</div>
- 
-<script>
-// Modal Image Gallery
-function onClick(element) {
-  document.getElementById("img01").src = element.src;
-  document.getElementById("modal01").style.display = "block";
-  var captionText = document.getElementById("caption");
-  captionText.innerHTML = element.alt;
-}
-
-// Change style of navbar on scroll
-window.onscroll = function() {myFunction()};
-function myFunction() {
-    var navbar = document.getElementById("myNavbar");
-    if (document.body.scrollTop > 100 || document.documentElement.scrollTop > 100) {
-        navbar.className = "w3-bar" + " w3-card" + " w3-animate-top" + " w3-white";
-    } else {
-        navbar.className = navbar.className.replace(" w3-card w3-animate-top w3-white", "");
-    }
-}
-
-// Used to toggle the menu on small screens when clicking on the menu button
-function toggleFunction() {
-    var x = document.getElementById("navDemo");
-    if (x.className.indexOf("w3-show") == -1) {
-        x.className += " w3-show";
-    } else {
-        x.className = x.className.replace(" w3-show", "");
-    }
-}
-</script>
-
-</body>
-</html>
-
-
-
-'''
-
-import gradio as gr
-import tweepy
-from fastapi import FastAPI, Request
-
-consumer_token = os.getenv('CONSUMER_TOKEN')
-consumer_secret = os.getenv('CONSUMER_SECRET')
-my_access_token = os.getenv('ACCESS_TOKEN')
-my_access_secret = os.getenv('ACCESS_SECRET')
-bearer = os.getenv('BEARER')
-client_id = os.getenv('CLIENT_ID')
-client_secret = os.getenv('CLIENT_SECRET')
-
-oauth1_user_handler = tweepy.OAuth2UserHandler(client_id=client_id,
-                                               client_secret=client_secret,
-                                               redirect_uri="https://hf.space/embed/User1342/WatchTower/",
-                                               scope=["block.write"]
+from types import TracebackType
+from typing import (
+    IO,
+    Any,
+    AsyncGenerator,
+    Awaitable,
+    Callable,
+    Collection,
+    Coroutine,
+    Deque,
+    Dict,
+    Generator,
+    Iterable,
+    List,
+    Mapping,
+    Optional,
+    Sequence,
+    Set,
+    Tuple,
+    Type,
+    TypeVar,
+    Union,
+    cast,
+)
+from weakref import WeakKeyDictionary
+
+import sniffio
 
+from .. import CapacityLimiterStatistics, EventStatistics, TaskInfo, abc
+from .._core._compat import DeprecatedAsyncContextManager, DeprecatedAwaitable
+from .._core._eventloop import claim_worker_thread, threadlocals
+from .._core._exceptions import (
+    BrokenResourceError,
+    BusyResourceError,
+    ClosedResourceError,
+    EndOfStream,
 )
+from .._core._exceptions import ExceptionGroup as BaseExceptionGroup
+from .._core._exceptions import WouldBlock
+from .._core._sockets import GetAddrInfoReturnType, convert_ipv6_sockaddr
+from .._core._synchronization import CapacityLimiter as BaseCapacityLimiter
+from .._core._synchronization import Event as BaseEvent
+from .._core._synchronization import ResourceGuard
+from .._core._tasks import CancelScope as BaseCancelScope
+from ..abc import IPSockAddrType, UDPPacketType
+from ..lowlevel import RunVar
+
+if sys.version_info >= (3, 8):
+
+    def get_coro(task: asyncio.Task) -> Union[Generator, Awaitable[Any]]:
+        return task.get_coro()
+
+else:
+
+    def get_coro(task: asyncio.Task) -> Union[Generator, Awaitable[Any]]:
+        return task._coro
+
+
+if sys.version_info >= (3, 7):
+    from asyncio import all_tasks, create_task, current_task, get_running_loop
+    from asyncio import run as native_run
+
+    def _get_task_callbacks(task: asyncio.Task) -> Iterable[Callable]:
+        return [cb for cb, context in task._callbacks]  # type: ignore[attr-defined]
+
+else:
+    _T = TypeVar("_T")
+
+    def _get_task_callbacks(task: asyncio.Task) -> Iterable[Callable]:
+        return task._callbacks
+
+    def native_run(main, *, debug=False):
+        # Snatched from Python 3.7
+        from asyncio import coroutines, events, tasks
+
+        def _cancel_all_tasks(loop):
+            to_cancel = all_tasks(loop)
+            if not to_cancel:
+                return
+
+            for task in to_cancel:
+                task.cancel()
+
+            loop.run_until_complete(
+                tasks.gather(*to_cancel, loop=loop, return_exceptions=True)
+            )
+
+            for task in to_cancel:
+                if task.cancelled():
+                    continue
+                if task.exception() is not None:
+                    loop.call_exception_handler(
+                        {
+                            "message": "unhandled exception during asyncio.run() shutdown",
+                            "exception": task.exception(),
+                            "task": task,
+                        }
+                    )
+
+        if events._get_running_loop() is not None:
+            raise RuntimeError(
+                "asyncio.run() cannot be called from a running event loop"
+            )
+
+        if not coroutines.iscoroutine(main):
+            raise ValueError(f"a coroutine was expected, got {main!r}")
+
+        loop = events.new_event_loop()
+        try:
+            events.set_event_loop(loop)
+            loop.set_debug(debug)
+            return loop.run_until_complete(main)
+        finally:
+            try:
+                _cancel_all_tasks(loop)
+                loop.run_until_complete(loop.shutdown_asyncgens())
+            finally:
+                events.set_event_loop(None)
+                loop.close()
+
+    def create_task(
+        coro: Union[Generator[Any, None, _T], Awaitable[_T]], *, name: object = None
+    ) -> asyncio.Task:
+        return get_running_loop().create_task(coro)
+
+    def get_running_loop() -> asyncio.AbstractEventLoop:
+        loop = asyncio._get_running_loop()
+        if loop is not None:
+            return loop
+        else:
+            raise RuntimeError("no running event loop")
+
+    def all_tasks(
+        loop: Optional[asyncio.AbstractEventLoop] = None,
+    ) -> Set[asyncio.Task]:
+        """Return a set of all tasks for the loop."""
+        from asyncio import Task
+
+        if loop is None:
+            loop = get_running_loop()
+
+        return {t for t in Task.all_tasks(loop) if not t.done()}
+
+    def current_task(
+        loop: Optional[asyncio.AbstractEventLoop] = None,
+    ) -> Optional[asyncio.Task]:
+        if loop is None:
+            loop = get_running_loop()
+
+        return asyncio.Task.current_task(loop)
+
+
+T_Retval = TypeVar("T_Retval")
+
+# Check whether there is native support for task names in asyncio (3.8+)
+_native_task_names = hasattr(asyncio.Task, "get_name")
+
+
+_root_task: RunVar[Optional[asyncio.Task]] = RunVar("_root_task")
+
+
+def find_root_task() -> asyncio.Task:
+    root_task = _root_task.get(None)
+    if root_task is not None and not root_task.done():
+        return root_task
+
+    # Look for a task that has been started via run_until_complete()
+    for task in all_tasks():
+        if task._callbacks and not task.done():
+            for cb in _get_task_callbacks(task):
+                if (
+                    cb is _run_until_complete_cb
+                    or getattr(cb, "__module__", None) == "uvloop.loop"
+                ):
+                    _root_task.set(task)
+                    return task
+
+    # Look up the topmost task in the AnyIO task tree, if possible
+    task = cast(asyncio.Task, current_task())
+    state = _task_states.get(task)
+    if state:
+        cancel_scope = state.cancel_scope
+        while cancel_scope and cancel_scope._parent_scope is not None:
+            cancel_scope = cancel_scope._parent_scope
+
+        if cancel_scope is not None:
+            return cast(asyncio.Task, cancel_scope._host_task)
+
+    return task
+
+
+def get_callable_name(func: Callable) -> str:
+    module = getattr(func, "__module__", None)
+    qualname = getattr(func, "__qualname__", None)
+    return ".".join([x for x in (module, qualname) if x])
+
+
+#
+# Event loop
+#
+
+_run_vars = (
+    WeakKeyDictionary()
+)  # type: WeakKeyDictionary[asyncio.AbstractEventLoop, Any]
+
+current_token = get_running_loop
+
+
+def _task_started(task: asyncio.Task) -> bool:
+    """Return ``True`` if the task has been started and has not finished."""
+    coro = cast(Coroutine[Any, Any, Any], get_coro(task))
+    try:
+        return getcoroutinestate(coro) in (CORO_RUNNING, CORO_SUSPENDED)
+    except AttributeError:
+        try:
+            return getgeneratorstate(cast(Generator, coro)) in (
+                GEN_RUNNING,
+                GEN_SUSPENDED,
+            )
+        except AttributeError:
+            # task coro is async_genenerator_asend https://bugs.python.org/issue37771
+            raise Exception(f"Cannot determine if task {task} has started or not")
+
+
+def _maybe_set_event_loop_policy(
+    policy: Optional[asyncio.AbstractEventLoopPolicy], use_uvloop: bool
+) -> None:
+    # On CPython, use uvloop when possible if no other policy has been given and if not
+    # explicitly disabled
+    if policy is None and use_uvloop and sys.implementation.name == "cpython":
+        try:
+            import uvloop
+        except ImportError:
+            pass
+        else:
+            # Test for missing shutdown_default_executor() (uvloop 0.14.0 and earlier)
+            if not hasattr(
+                asyncio.AbstractEventLoop, "shutdown_default_executor"
+            ) or hasattr(uvloop.loop.Loop, "shutdown_default_executor"):
+                policy = uvloop.EventLoopPolicy()
+
+    if policy is not None:
+        asyncio.set_event_loop_policy(policy)
+
+
+def run(
+    func: Callable[..., Awaitable[T_Retval]],
+    *args: object,
+    debug: bool = False,
+    use_uvloop: bool = False,
+    policy: Optional[asyncio.AbstractEventLoopPolicy] = None,
+) -> T_Retval:
+    @wraps(func)
+    async def wrapper() -> T_Retval:
+        task = cast(asyncio.Task, current_task())
+        task_state = TaskState(None, get_callable_name(func), None)
+        _task_states[task] = task_state
+        if _native_task_names:
+            task.set_name(task_state.name)
+
+        try:
+            return await func(*args)
+        finally:
+            del _task_states[task]
+
+    _maybe_set_event_loop_policy(policy, use_uvloop)
+    return native_run(wrapper(), debug=debug)
+
+
+#
+# Miscellaneous
+#
+
+sleep = asyncio.sleep
+
+
+#
+# Timeouts and cancellation
+#
+
+CancelledError = asyncio.CancelledError
+
+
+class CancelScope(BaseCancelScope):
+    def __new__(
+        cls, *, deadline: float = math.inf, shield: bool = False
+    ) -> "CancelScope":
+        return object.__new__(cls)
+
+    def __init__(self, deadline: float = math.inf, shield: bool = False):
+        self._deadline = deadline
+        self._shield = shield
+        self._parent_scope: Optional[CancelScope] = None
+        self._cancel_called = False
+        self._active = False
+        self._timeout_handle: Optional[asyncio.TimerHandle] = None
+        self._cancel_handle: Optional[asyncio.Handle] = None
+        self._tasks: Set[asyncio.Task] = set()
+        self._host_task: Optional[asyncio.Task] = None
+        self._timeout_expired = False
+
+    def __enter__(self) -> "CancelScope":
+        if self._active:
+            raise RuntimeError(
+                "Each CancelScope may only be used for a single 'with' block"
+            )
+
+        self._host_task = host_task = cast(asyncio.Task, current_task())
+        self._tasks.add(host_task)
+        try:
+            task_state = _task_states[host_task]
+        except KeyError:
+            task_name = host_task.get_name() if _native_task_names else None
+            task_state = TaskState(None, task_name, self)
+            _task_states[host_task] = task_state
+        else:
+            self._parent_scope = task_state.cancel_scope
+            task_state.cancel_scope = self
+
+        self._timeout()
+        self._active = True
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> Optional[bool]:
+        if not self._active:
+            raise RuntimeError("This cancel scope is not active")
+        if current_task() is not self._host_task:
+            raise RuntimeError(
+                "Attempted to exit cancel scope in a different task than it was "
+                "entered in"
+            )
+
+        assert self._host_task is not None
+        host_task_state = _task_states.get(self._host_task)
+        if host_task_state is None or host_task_state.cancel_scope is not self:
+            raise RuntimeError(
+                "Attempted to exit a cancel scope that isn't the current tasks's "
+                "current cancel scope"
+            )
+
+        self._active = False
+        if self._timeout_handle:
+            self._timeout_handle.cancel()
+            self._timeout_handle = None
+
+        self._tasks.remove(self._host_task)
+
+        host_task_state.cancel_scope = self._parent_scope
+
+        # Restart the cancellation effort in the farthest directly cancelled parent scope if this
+        # one was shielded
+        if self._shield:
+            self._deliver_cancellation_to_parent()
+
+        if exc_val is not None:
+            exceptions = (
+                exc_val.exceptions if isinstance(exc_val, ExceptionGroup) else [exc_val]
+            )
+            if all(isinstance(exc, CancelledError) for exc in exceptions):
+                if self._timeout_expired:
+                    return True
+                elif not self._cancel_called:
+                    # Task was cancelled natively
+                    return None
+                elif not self._parent_cancelled():
+                    # This scope was directly cancelled
+                    return True
+
+        return None
+
+    def _timeout(self) -> None:
+        if self._deadline != math.inf:
+            loop = get_running_loop()
+            if loop.time() >= self._deadline:
+                self._timeout_expired = True
+                self.cancel()
+            else:
+                self._timeout_handle = loop.call_at(self._deadline, self._timeout)
+
+    def _deliver_cancellation(self) -> None:
+        """
+        Deliver cancellation to directly contained tasks and nested cancel scopes.
+
+        Schedule another run at the end if we still have tasks eligible for cancellation.
+        """
+        should_retry = False
+        current = current_task()
+        for task in self._tasks:
+            if task._must_cancel:  # type: ignore[attr-defined]
+                continue
+
+            # The task is eligible for cancellation if it has started and is not in a cancel
+            # scope shielded from this one
+            cancel_scope = _task_states[task].cancel_scope
+            while cancel_scope is not self:
+                if cancel_scope is None or cancel_scope._shield:
+                    break
+                else:
+                    cancel_scope = cancel_scope._parent_scope
+            else:
+                should_retry = True
+                if task is not current and (
+                    task is self._host_task or _task_started(task)
+                ):
+                    task.cancel()
+
+        # Schedule another callback if there are still tasks left
+        if should_retry:
+            self._cancel_handle = get_running_loop().call_soon(
+                self._deliver_cancellation
+            )
+        else:
+            self._cancel_handle = None
+
+    def _deliver_cancellation_to_parent(self) -> None:
+        """Start cancellation effort in the farthest directly cancelled parent scope"""
+        scope = self._parent_scope
+        scope_to_cancel: Optional[CancelScope] = None
+        while scope is not None:
+            if scope._cancel_called and scope._cancel_handle is None:
+                scope_to_cancel = scope
+
+            # No point in looking beyond any shielded scope
+            if scope._shield:
+                break
+
+            scope = scope._parent_scope
+
+        if scope_to_cancel is not None:
+            scope_to_cancel._deliver_cancellation()
+
+    def _parent_cancelled(self) -> bool:
+        # Check whether any parent has been cancelled
+        cancel_scope = self._parent_scope
+        while cancel_scope is not None and not cancel_scope._shield:
+            if cancel_scope._cancel_called:
+                return True
+            else:
+                cancel_scope = cancel_scope._parent_scope
+
+        return False
+
+    def cancel(self) -> DeprecatedAwaitable:
+        if not self._cancel_called:
+            if self._timeout_handle:
+                self._timeout_handle.cancel()
+                self._timeout_handle = None
+
+            self._cancel_called = True
+            self._deliver_cancellation()
+
+        return DeprecatedAwaitable(self.cancel)
+
+    @property
+    def deadline(self) -> float:
+        return self._deadline
+
+    @deadline.setter
+    def deadline(self, value: float) -> None:
+        self._deadline = float(value)
+        if self._timeout_handle is not None:
+            self._timeout_handle.cancel()
+            self._timeout_handle = None
+
+        if self._active and not self._cancel_called:
+            self._timeout()
+
+    @property
+    def cancel_called(self) -> bool:
+        return self._cancel_called
+
+    @property
+    def shield(self) -> bool:
+        return self._shield
+
+    @shield.setter
+    def shield(self, value: bool) -> None:
+        if self._shield != value:
+            self._shield = value
+            if not value:
+                self._deliver_cancellation_to_parent()
+
+
+async def checkpoint() -> None:
+    await sleep(0)
+
+
+async def checkpoint_if_cancelled() -> None:
+    task = current_task()
+    if task is None:
+        return
+
+    try:
+        cancel_scope = _task_states[task].cancel_scope
+    except KeyError:
+        return
+
+    while cancel_scope:
+        if cancel_scope.cancel_called:
+            await sleep(0)
+        elif cancel_scope.shield:
+            break
+        else:
+            cancel_scope = cancel_scope._parent_scope
+
+
+async def cancel_shielded_checkpoint() -> None:
+    with CancelScope(shield=True):
+        await sleep(0)
+
+
+def current_effective_deadline() -> float:
+    try:
+        cancel_scope = _task_states[current_task()].cancel_scope  # type: ignore[index]
+    except KeyError:
+        return math.inf
+
+    deadline = math.inf
+    while cancel_scope:
+        deadline = min(deadline, cancel_scope.deadline)
+        if cancel_scope.shield:
+            break
+        else:
+            cancel_scope = cancel_scope._parent_scope
+
+    return deadline
+
+
+def current_time() -> float:
+    return get_running_loop().time()
+
+
+#
+# Task states
+#
+
+
+class TaskState:
+    """
+    Encapsulates auxiliary task information that cannot be added to the Task instance itself
+    because there are no guarantees about its implementation.
+    """
+
+    __slots__ = "parent_id", "name", "cancel_scope"
+
+    def __init__(
+        self,
+        parent_id: Optional[int],
+        name: Optional[str],
+        cancel_scope: Optional[CancelScope],
+    ):
+        self.parent_id = parent_id
+        self.name = name
+        self.cancel_scope = cancel_scope
+
+
+_task_states = WeakKeyDictionary()  # type: WeakKeyDictionary[asyncio.Task, TaskState]
+
+
+#
+# Task groups
+#
+
+
+class ExceptionGroup(BaseExceptionGroup):
+    def __init__(self, exceptions: List[BaseException]):
+        super().__init__()
+        self.exceptions = exceptions
+
+
+class _AsyncioTaskStatus(abc.TaskStatus):
+    def __init__(self, future: asyncio.Future, parent_id: int):
+        self._future = future
+        self._parent_id = parent_id
+
+    def started(self, value: object = None) -> None:
+        try:
+            self._future.set_result(value)
+        except asyncio.InvalidStateError:
+            raise RuntimeError(
+                "called 'started' twice on the same task status"
+            ) from None
+
+        task = cast(asyncio.Task, current_task())
+        _task_states[task].parent_id = self._parent_id
+
+
+class TaskGroup(abc.TaskGroup):
+    def __init__(self) -> None:
+        self.cancel_scope: CancelScope = CancelScope()
+        self._active = False
+        self._exceptions: List[BaseException] = []
+
+    async def __aenter__(self) -> "TaskGroup":
+        self.cancel_scope.__enter__()
+        self._active = True
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> Optional[bool]:
+        ignore_exception = self.cancel_scope.__exit__(exc_type, exc_val, exc_tb)
+        if exc_val is not None:
+            self.cancel_scope.cancel()
+            self._exceptions.append(exc_val)
+
+        while self.cancel_scope._tasks:
+            try:
+                await asyncio.wait(self.cancel_scope._tasks)
+            except asyncio.CancelledError:
+                self.cancel_scope.cancel()
+
+        self._active = False
+        if not self.cancel_scope._parent_cancelled():
+            exceptions = self._filter_cancellation_errors(self._exceptions)
+        else:
+            exceptions = self._exceptions
+
+        try:
+            if len(exceptions) > 1:
+                if all(
+                    isinstance(e, CancelledError) and not e.args for e in exceptions
+                ):
+                    # Tasks were cancelled natively, without a cancellation message
+                    raise CancelledError
+                else:
+                    raise ExceptionGroup(exceptions)
+            elif exceptions and exceptions[0] is not exc_val:
+                raise exceptions[0]
+        except BaseException as exc:
+            # Clear the context here, as it can only be done in-flight.
+            # If the context is not cleared, it can result in recursive tracebacks (see #145).
+            exc.__context__ = None
+            raise
+
+        return ignore_exception
+
+    @staticmethod
+    def _filter_cancellation_errors(
+        exceptions: Sequence[BaseException],
+    ) -> List[BaseException]:
+        filtered_exceptions: List[BaseException] = []
+        for exc in exceptions:
+            if isinstance(exc, ExceptionGroup):
+                new_exceptions = TaskGroup._filter_cancellation_errors(exc.exceptions)
+                if len(new_exceptions) > 1:
+                    filtered_exceptions.append(exc)
+                elif len(new_exceptions) == 1:
+                    filtered_exceptions.append(new_exceptions[0])
+                elif new_exceptions:
+                    new_exc = ExceptionGroup(new_exceptions)
+                    new_exc.__cause__ = exc.__cause__
+                    new_exc.__context__ = exc.__context__
+                    new_exc.__traceback__ = exc.__traceback__
+                    filtered_exceptions.append(new_exc)
+            elif not isinstance(exc, CancelledError) or exc.args:
+                filtered_exceptions.append(exc)
+
+        return filtered_exceptions
+
+    async def _run_wrapped_task(
+        self, coro: Coroutine, task_status_future: Optional[asyncio.Future]
+    ) -> None:
+        # This is the code path for Python 3.6 and 3.7 on which asyncio freaks out if a task raises
+        # a BaseException.
+        __traceback_hide__ = __tracebackhide__ = True  # noqa: F841
+        task = cast(asyncio.Task, current_task())
+        try:
+            await coro
+        except BaseException as exc:
+            if task_status_future is None or task_status_future.done():
+                self._exceptions.append(exc)
+                self.cancel_scope.cancel()
+            else:
+                task_status_future.set_exception(exc)
+        else:
+            if task_status_future is not None and not task_status_future.done():
+                task_status_future.set_exception(
+                    RuntimeError("Child exited without calling task_status.started()")
+                )
+        finally:
+            if task in self.cancel_scope._tasks:
+                self.cancel_scope._tasks.remove(task)
+                del _task_states[task]
+
+    def _spawn(
+        self,
+        func: Callable[..., Coroutine],
+        args: tuple,
+        name: object,
+        task_status_future: Optional[asyncio.Future] = None,
+    ) -> asyncio.Task:
+        def task_done(_task: asyncio.Task) -> None:
+            # This is the code path for Python 3.8+
+            assert _task in self.cancel_scope._tasks
+            self.cancel_scope._tasks.remove(_task)
+            del _task_states[_task]
+
+            try:
+                exc = _task.exception()
+            except CancelledError as e:
+                while isinstance(e.__context__, CancelledError):
+                    e = e.__context__
+
+                exc = e
+
+            if exc is not None:
+                if task_status_future is None or task_status_future.done():
+                    self._exceptions.append(exc)
+                    self.cancel_scope.cancel()
+                else:
+                    task_status_future.set_exception(exc)
+            elif task_status_future is not None and not task_status_future.done():
+                task_status_future.set_exception(
+                    RuntimeError("Child exited without calling task_status.started()")
+                )
+
+        if not self._active:
+            raise RuntimeError(
+                "This task group is not active; no new tasks can be started."
+            )
+
+        options = {}
+        name = get_callable_name(func) if name is None else str(name)
+        if _native_task_names:
+            options["name"] = name
+
+        kwargs = {}
+        if task_status_future:
+            parent_id = id(current_task())
+            kwargs["task_status"] = _AsyncioTaskStatus(
+                task_status_future, id(self.cancel_scope._host_task)
+            )
+        else:
+            parent_id = id(self.cancel_scope._host_task)
+
+        coro = func(*args, **kwargs)
+        if not asyncio.iscoroutine(coro):
+            raise TypeError(
+                f"Expected an async function, but {func} appears to be synchronous"
+            )
+
+        foreign_coro = not hasattr(coro, "cr_frame") and not hasattr(coro, "gi_frame")
+        if foreign_coro or sys.version_info < (3, 8):
+            coro = self._run_wrapped_task(coro, task_status_future)
+
+        task = create_task(coro, **options)
+        if not foreign_coro and sys.version_info >= (3, 8):
+            task.add_done_callback(task_done)
+
+        # Make the spawned task inherit the task group's cancel scope
+        _task_states[task] = TaskState(
+            parent_id=parent_id, name=name, cancel_scope=self.cancel_scope
+        )
+        self.cancel_scope._tasks.add(task)
+        return task
+
+    def start_soon(
+        self, func: Callable[..., Coroutine], *args: object, name: object = None
+    ) -> None:
+        self._spawn(func, args, name)
+
+    async def start(
+        self, func: Callable[..., Coroutine], *args: object, name: object = None
+    ) -> None:
+        future: asyncio.Future = asyncio.Future()
+        task = self._spawn(func, args, name, future)
+
+        # If the task raises an exception after sending a start value without a switch point
+        # between, the task group is cancelled and this method never proceeds to process the
+        # completed future. That's why we have to have a shielded cancel scope here.
+        with CancelScope(shield=True):
+            try:
+                return await future
+            except CancelledError:
+                task.cancel()
+                raise
+
+
+#
+# Threads
+#
+
+_Retval_Queue_Type = Tuple[Optional[T_Retval], Optional[BaseException]]
+
+
+class WorkerThread(Thread):
+    MAX_IDLE_TIME = 10  # seconds
+
+    def __init__(
+        self,
+        root_task: asyncio.Task,
+        workers: Set["WorkerThread"],
+        idle_workers: Deque["WorkerThread"],
+    ):
+        super().__init__(name="AnyIO worker thread")
+        self.root_task = root_task
+        self.workers = workers
+        self.idle_workers = idle_workers
+        self.loop = root_task._loop
+        self.queue: Queue[
+            Union[Tuple[Context, Callable, tuple, asyncio.Future], None]
+        ] = Queue(2)
+        self.idle_since = current_time()
+        self.stopping = False
+
+    def _report_result(
+        self, future: asyncio.Future, result: Any, exc: Optional[BaseException]
+    ) -> None:
+        self.idle_since = current_time()
+        if not self.stopping:
+            self.idle_workers.append(self)
+
+        if not future.cancelled():
+            if exc is not None:
+                future.set_exception(exc)
+            else:
+                future.set_result(result)
+
+    def run(self) -> None:
+        with claim_worker_thread("asyncio"):
+            threadlocals.loop = self.loop
+            while True:
+                item = self.queue.get()
+                if item is None:
+                    # Shutdown command received
+                    return
+
+                context, func, args, future = item
+                if not future.cancelled():
+                    result = None
+                    exception: Optional[BaseException] = None
+                    try:
+                        result = context.run(func, *args)
+                    except BaseException as exc:
+                        exception = exc
+
+                    if not self.loop.is_closed():
+                        self.loop.call_soon_threadsafe(
+                            self._report_result, future, result, exception
+                        )
+
+                self.queue.task_done()
+
+    def stop(self, f: Optional[asyncio.Task] = None) -> None:
+        self.stopping = True
+        self.queue.put_nowait(None)
+        self.workers.discard(self)
+        try:
+            self.idle_workers.remove(self)
+        except ValueError:
+            pass
+
+
+_threadpool_idle_workers: RunVar[Deque[WorkerThread]] = RunVar(
+    "_threadpool_idle_workers"
+)
+_threadpool_workers: RunVar[Set[WorkerThread]] = RunVar("_threadpool_workers")
+
+
+async def run_sync_in_worker_thread(
+    func: Callable[..., T_Retval],
+    *args: object,
+    cancellable: bool = False,
+    limiter: Optional["CapacityLimiter"] = None,
+) -> T_Retval:
+    await checkpoint()
+
+    # If this is the first run in this event loop thread, set up the necessary variables
+    try:
+        idle_workers = _threadpool_idle_workers.get()
+        workers = _threadpool_workers.get()
+    except LookupError:
+        idle_workers = deque()
+        workers = set()
+        _threadpool_idle_workers.set(idle_workers)
+        _threadpool_workers.set(workers)
+
+    async with (limiter or current_default_thread_limiter()):
+        with CancelScope(shield=not cancellable):
+            future: asyncio.Future = asyncio.Future()
+            root_task = find_root_task()
+            if not idle_workers:
+                worker = WorkerThread(root_task, workers, idle_workers)
+                worker.start()
+                workers.add(worker)
+                root_task.add_done_callback(worker.stop)
+            else:
+                worker = idle_workers.pop()
+
+                # Prune any other workers that have been idle for MAX_IDLE_TIME seconds or longer
+                now = current_time()
+                while idle_workers:
+                    if now - idle_workers[0].idle_since < WorkerThread.MAX_IDLE_TIME:
+                        break
+
+                    expired_worker = idle_workers.popleft()
+                    expired_worker.root_task.remove_done_callback(expired_worker.stop)
+                    expired_worker.stop()
+
+            context = copy_context()
+            context.run(sniffio.current_async_library_cvar.set, None)
+            worker.queue.put_nowait((context, func, args, future))
+            return await future
+
+
+def run_sync_from_thread(
+    func: Callable[..., T_Retval],
+    *args: object,
+    loop: Optional[asyncio.AbstractEventLoop] = None,
+) -> T_Retval:
+    @wraps(func)
+    def wrapper() -> None:
+        try:
+            f.set_result(func(*args))
+        except BaseException as exc:
+            f.set_exception(exc)
+            if not isinstance(exc, Exception):
+                raise
+
+    f: concurrent.futures.Future[T_Retval] = Future()
+    loop = loop or threadlocals.loop
+    if sys.version_info < (3, 7):
+        loop.call_soon_threadsafe(copy_context().run, wrapper)
+    else:
+        loop.call_soon_threadsafe(wrapper)
+
+    return f.result()
+
+
+def run_async_from_thread(
+    func: Callable[..., Coroutine[Any, Any, T_Retval]], *args: object
+) -> T_Retval:
+    f: concurrent.futures.Future[T_Retval] = asyncio.run_coroutine_threadsafe(
+        func(*args), threadlocals.loop
+    )
+    return f.result()
+
+
+class BlockingPortal(abc.BlockingPortal):
+    def __new__(cls) -> "BlockingPortal":
+        return object.__new__(cls)
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._loop = get_running_loop()
+
+    def _spawn_task_from_thread(
+        self,
+        func: Callable,
+        args: tuple,
+        kwargs: Dict[str, Any],
+        name: object,
+        future: Future,
+    ) -> None:
+        run_sync_from_thread(
+            partial(self._task_group.start_soon, name=name),
+            self._call_func,
+            func,
+            args,
+            kwargs,
+            future,
+            loop=self._loop,
+        )
+
+
+#
+# Subprocesses
+#
+
+
+@dataclass(eq=False)
+class StreamReaderWrapper(abc.ByteReceiveStream):
+    _stream: asyncio.StreamReader
+
+    async def receive(self, max_bytes: int = 65536) -> bytes:
+        data = await self._stream.read(max_bytes)
+        if data:
+            return data
+        else:
+            raise EndOfStream
+
+    async def aclose(self) -> None:
+        self._stream.feed_eof()
+
+
+@dataclass(eq=False)
+class StreamWriterWrapper(abc.ByteSendStream):
+    _stream: asyncio.StreamWriter
+
+    async def send(self, item: bytes) -> None:
+        self._stream.write(item)
+        await self._stream.drain()
+
+    async def aclose(self) -> None:
+        self._stream.close()
+
+
+@dataclass(eq=False)
+class Process(abc.Process):
+    _process: asyncio.subprocess.Process
+    _stdin: Optional[StreamWriterWrapper]
+    _stdout: Optional[StreamReaderWrapper]
+    _stderr: Optional[StreamReaderWrapper]
+
+    async def aclose(self) -> None:
+        if self._stdin:
+            await self._stdin.aclose()
+        if self._stdout:
+            await self._stdout.aclose()
+        if self._stderr:
+            await self._stderr.aclose()
+
+        await self.wait()
+
+    async def wait(self) -> int:
+        return await self._process.wait()
+
+    def terminate(self) -> None:
+        self._process.terminate()
+
+    def kill(self) -> None:
+        self._process.kill()
+
+    def send_signal(self, signal: int) -> None:
+        self._process.send_signal(signal)
+
+    @property
+    def pid(self) -> int:
+        return self._process.pid
+
+    @property
+    def returncode(self) -> Optional[int]:
+        return self._process.returncode
+
+    @property
+    def stdin(self) -> Optional[abc.ByteSendStream]:
+        return self._stdin
 
-target_website = oauth1_user_handler.get_authorization_url()
-print(target_website)
-
-block = gr.Blocks(css=".container { max-width: 800px; margin: auto; }")
-
-chat_history = []
-
-
-def get_client_from_tokens():
-    oauth1_user_handler = tweepy.OAuth2UserHandler(client_id=client_id,
-                                                   client_secret=client_secret,
-                                                   redirect_uri="https://hf.space/embed/User1342/WatchTower/",
-                                                   scope=["block.write"])
-
-    for connection in block.server.server_state.connections:
-        # connection_app_id = connection.app.app.blocks.app_id
-        # if active_app_id == connection_app_id:
-        #    print("Its a match")
-        url = None
-        if connection.headers != None:
-            for header in connection.headers:
-                header = header[1].decode()
-                if "code" in header:
-                    url = header
-                print(header)
-    print("urls {}".format(url))
-    their_client = None
-    if url != None:
-        access_token = oauth1_user_handler.fetch_token(
-            url
+    @property
+    def stdout(self) -> Optional[abc.ByteReceiveStream]:
+        return self._stdout
+
+    @property
+    def stderr(self) -> Optional[abc.ByteReceiveStream]:
+        return self._stderr
+
+
+async def open_process(
+    command: Union[str, bytes, Sequence[Union[str, bytes]]],
+    *,
+    shell: bool,
+    stdin: Union[int, IO[Any], None],
+    stdout: Union[int, IO[Any], None],
+    stderr: Union[int, IO[Any], None],
+    cwd: Union[str, bytes, PathLike, None] = None,
+    env: Optional[Mapping[str, str]] = None,
+    start_new_session: bool = False,
+) -> Process:
+    await checkpoint()
+    if shell:
+        process = await asyncio.create_subprocess_shell(
+            cast(Union[str, bytes], command),
+            stdin=stdin,
+            stdout=stdout,
+            stderr=stderr,
+            cwd=cwd,
+            env=env,
+            start_new_session=start_new_session,
+        )
+    else:
+        process = await asyncio.create_subprocess_exec(
+            *command,
+            stdin=stdin,
+            stdout=stdout,
+            stderr=stderr,
+            cwd=cwd,
+            env=env,
+            start_new_session=start_new_session,
         )
-        their_client = tweepy.Client(access_token)
 
-    return their_client
+    stdin_stream = StreamWriterWrapper(process.stdin) if process.stdin else None
+    stdout_stream = StreamReaderWrapper(process.stdout) if process.stdout else None
+    stderr_stream = StreamReaderWrapper(process.stderr) if process.stderr else None
+    return Process(process, stdin_stream, stdout_stream, stderr_stream)
+
+
+def _forcibly_shutdown_process_pool_on_exit(
+    workers: Set[Process], _task: object
+) -> None:
+    """
+    Forcibly shuts down worker processes belonging to this event loop."""
+    child_watcher: Optional[asyncio.AbstractChildWatcher]
+    try:
+        child_watcher = asyncio.get_event_loop_policy().get_child_watcher()
+    except NotImplementedError:
+        child_watcher = None
+
+    # Close as much as possible (w/o async/await) to avoid warnings
+    for process in workers:
+        if process.returncode is None:
+            continue
+
+        process._stdin._stream._transport.close()  # type: ignore[union-attr]
+        process._stdout._stream._transport.close()  # type: ignore[union-attr]
+        process._stderr._stream._transport.close()  # type: ignore[union-attr]
+        process.kill()
+        if child_watcher:
+            child_watcher.remove_child_handler(process.pid)
+
+
+async def _shutdown_process_pool_on_exit(workers: Set[Process]) -> None:
+    """
+    Shuts down worker processes belonging to this event loop.
+
+    NOTE: this only works when the event loop was started using asyncio.run() or anyio.run().
+
+    """
+    process: Process
+    try:
+        await sleep(math.inf)
+    except asyncio.CancelledError:
+        for process in workers:
+            if process.returncode is None:
+                process.kill()
+
+        for process in workers:
+            await process.aclose()
+
+
+def setup_process_pool_exit_at_shutdown(workers: Set[Process]) -> None:
+    kwargs = {"name": "AnyIO process pool shutdown task"} if _native_task_names else {}
+    create_task(_shutdown_process_pool_on_exit(workers), **kwargs)
+    find_root_task().add_done_callback(
+        partial(_forcibly_shutdown_process_pool_on_exit, workers)
+    )
+
+
+#
+# Sockets and networking
+#
+
+
+class StreamProtocol(asyncio.Protocol):
+    read_queue: Deque[bytes]
+    read_event: asyncio.Event
+    write_event: asyncio.Event
+    exception: Optional[Exception] = None
+
+    def connection_made(self, transport: asyncio.BaseTransport) -> None:
+        self.read_queue = deque()
+        self.read_event = asyncio.Event()
+        self.write_event = asyncio.Event()
+        self.write_event.set()
+        cast(asyncio.Transport, transport).set_write_buffer_limits(0)
+
+    def connection_lost(self, exc: Optional[Exception]) -> None:
+        if exc:
+            self.exception = BrokenResourceError()
+            self.exception.__cause__ = exc
+
+        self.read_event.set()
+        self.write_event.set()
+
+    def data_received(self, data: bytes) -> None:
+        self.read_queue.append(data)
+        self.read_event.set()
+
+    def eof_received(self) -> Optional[bool]:
+        self.read_event.set()
+        return True
+
+    def pause_writing(self) -> None:
+        self.write_event = asyncio.Event()
+
+    def resume_writing(self) -> None:
+        self.write_event.set()
+
+
+class DatagramProtocol(asyncio.DatagramProtocol):
+    read_queue: Deque[Tuple[bytes, IPSockAddrType]]
+    read_event: asyncio.Event
+    write_event: asyncio.Event
+    exception: Optional[Exception] = None
+
+    def connection_made(self, transport: asyncio.BaseTransport) -> None:
+        self.read_queue = deque(maxlen=100)  # arbitrary value
+        self.read_event = asyncio.Event()
+        self.write_event = asyncio.Event()
+        self.write_event.set()
+
+    def connection_lost(self, exc: Optional[Exception]) -> None:
+        self.read_event.set()
+        self.write_event.set()
+
+    def datagram_received(self, data: bytes, addr: IPSockAddrType) -> None:
+        addr = convert_ipv6_sockaddr(addr)
+        self.read_queue.append((data, addr))
+        self.read_event.set()
+
+    def error_received(self, exc: Exception) -> None:
+        self.exception = exc
+
+    def pause_writing(self) -> None:
+        self.write_event.clear()
+
+    def resume_writing(self) -> None:
+        self.write_event.set()
+
+
+class SocketStream(abc.SocketStream):
+    def __init__(self, transport: asyncio.Transport, protocol: StreamProtocol):
+        self._transport = transport
+        self._protocol = protocol
+        self._receive_guard = ResourceGuard("reading from")
+        self._send_guard = ResourceGuard("writing to")
+        self._closed = False
+
+    @property
+    def _raw_socket(self) -> socket.socket:
+        return self._transport.get_extra_info("socket")
+
+    async def receive(self, max_bytes: int = 65536) -> bytes:
+        with self._receive_guard:
+            await checkpoint()
+
+            if (
+                not self._protocol.read_event.is_set()
+                and not self._transport.is_closing()
+            ):
+                self._transport.resume_reading()
+                await self._protocol.read_event.wait()
+                self._transport.pause_reading()
+
+            try:
+                chunk = self._protocol.read_queue.popleft()
+            except IndexError:
+                if self._closed:
+                    raise ClosedResourceError from None
+                elif self._protocol.exception:
+                    raise self._protocol.exception
+                else:
+                    raise EndOfStream from None
+
+            if len(chunk) > max_bytes:
+                # Split the oversized chunk
+                chunk, leftover = chunk[:max_bytes], chunk[max_bytes:]
+                self._protocol.read_queue.appendleft(leftover)
+
+            # If the read queue is empty, clear the flag so that the next call will block until
+            # data is available
+            if not self._protocol.read_queue:
+                self._protocol.read_event.clear()
+
+        return chunk
+
+    async def send(self, item: bytes) -> None:
+        with self._send_guard:
+            await checkpoint()
+
+            if self._closed:
+                raise ClosedResourceError
+            elif self._protocol.exception is not None:
+                raise self._protocol.exception
+
+            try:
+                self._transport.write(item)
+            except RuntimeError as exc:
+                if self._transport.is_closing():
+                    raise BrokenResourceError from exc
+                else:
+                    raise
+
+            await self._protocol.write_event.wait()
+
+    async def send_eof(self) -> None:
+        try:
+            self._transport.write_eof()
+        except OSError:
+            pass
+
+    async def aclose(self) -> None:
+        if not self._transport.is_closing():
+            self._closed = True
+            try:
+                self._transport.write_eof()
+            except OSError:
+                pass
+
+            self._transport.close()
+            await sleep(0)
+            self._transport.abort()
+
 
-def block_users(client, threshold, dataset):
-    num_users_blocked = 0
+class UNIXSocketStream(abc.SocketStream):
+    _receive_future: Optional[asyncio.Future] = None
+    _send_future: Optional[asyncio.Future] = None
+    _closing = False
 
-    for filename in os.listdir("users"):
-        filename = os.path.join("users", filename)
+    def __init__(self, raw_socket: socket.socket):
+        self.__raw_socket = raw_socket
+        self._loop = get_running_loop()
+        self._receive_guard = ResourceGuard("reading from")
+        self._send_guard = ResourceGuard("writing to")
 
-        user_file = open(filename, "r")
-        users = json.load(user_file)
+    @property
+    def _raw_socket(self) -> socket.socket:
+        return self.__raw_socket
 
-        for user in users:
-            if threshold >= user["threshold"]:
+    def _wait_until_readable(self, loop: asyncio.AbstractEventLoop) -> asyncio.Future:
+        def callback(f: object) -> None:
+            del self._receive_future
+            loop.remove_reader(self.__raw_socket)
 
-                user_id = str(user["username"])
+        f = self._receive_future = asyncio.Future()
+        self._loop.add_reader(self.__raw_socket, f.set_result, None)
+        f.add_done_callback(callback)
+        return f
 
-                finished = False
-                while not finished:
+    def _wait_until_writable(self, loop: asyncio.AbstractEventLoop) -> asyncio.Future:
+        def callback(f: object) -> None:
+            del self._send_future
+            loop.remove_writer(self.__raw_socket)
+
+        f = self._send_future = asyncio.Future()
+        self._loop.add_writer(self.__raw_socket, f.set_result, None)
+        f.add_done_callback(callback)
+        return f
+
+    async def send_eof(self) -> None:
+        with self._send_guard:
+            self._raw_socket.shutdown(socket.SHUT_WR)
+
+    async def receive(self, max_bytes: int = 65536) -> bytes:
+        loop = get_running_loop()
+        await checkpoint()
+        with self._receive_guard:
+            while True:
+                try:
+                    data = self.__raw_socket.recv(max_bytes)
+                except BlockingIOError:
+                    await self._wait_until_readable(loop)
+                except OSError as exc:
+                    if self._closing:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+                else:
+                    if not data:
+                        raise EndOfStream
+
+                    return data
+
+    async def send(self, item: bytes) -> None:
+        loop = get_running_loop()
+        await checkpoint()
+        with self._send_guard:
+            view = memoryview(item)
+            while view:
+                try:
+                    bytes_sent = self.__raw_socket.send(item)
+                except BlockingIOError:
+                    await self._wait_until_writable(loop)
+                except OSError as exc:
+                    if self._closing:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+                else:
+                    view = view[bytes_sent:]
+
+    async def receive_fds(self, msglen: int, maxfds: int) -> Tuple[bytes, List[int]]:
+        if not isinstance(msglen, int) or msglen < 0:
+            raise ValueError("msglen must be a non-negative integer")
+        if not isinstance(maxfds, int) or maxfds < 1:
+            raise ValueError("maxfds must be a positive integer")
+
+        loop = get_running_loop()
+        fds = array.array("i")
+        await checkpoint()
+        with self._receive_guard:
+            while True:
+                try:
+                    message, ancdata, flags, addr = self.__raw_socket.recvmsg(
+                        msglen, socket.CMSG_LEN(maxfds * fds.itemsize)
+                    )
+                except BlockingIOError:
+                    await self._wait_until_readable(loop)
+                except OSError as exc:
+                    if self._closing:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+                else:
+                    if not message and not ancdata:
+                        raise EndOfStream
+
+                    break
+
+        for cmsg_level, cmsg_type, cmsg_data in ancdata:
+            if cmsg_level != socket.SOL_SOCKET or cmsg_type != socket.SCM_RIGHTS:
+                raise RuntimeError(
+                    f"Received unexpected ancillary data; message = {message!r}, "
+                    f"cmsg_level = {cmsg_level}, cmsg_type = {cmsg_type}"
+                )
+
+            fds.frombytes(cmsg_data[: len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
+
+        return message, list(fds)
+
+    async def send_fds(
+        self, message: bytes, fds: Collection[Union[int, IOBase]]
+    ) -> None:
+        if not message:
+            raise ValueError("message must not be empty")
+        if not fds:
+            raise ValueError("fds must not be empty")
+
+        loop = get_running_loop()
+        filenos: List[int] = []
+        for fd in fds:
+            if isinstance(fd, int):
+                filenos.append(fd)
+            elif isinstance(fd, IOBase):
+                filenos.append(fd.fileno())
+
+        fdarray = array.array("i", filenos)
+        await checkpoint()
+        with self._send_guard:
+            while True:
+                try:
+                    # The ignore can be removed after mypy picks up
+                    # https://github.com/python/typeshed/pull/5545
+                    self.__raw_socket.sendmsg(
+                        [message], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fdarray)]
+                    )
+                    break
+                except BlockingIOError:
+                    await self._wait_until_writable(loop)
+                except OSError as exc:
+                    if self._closing:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+
+    async def aclose(self) -> None:
+        if not self._closing:
+            self._closing = True
+            if self.__raw_socket.fileno() != -1:
+                self.__raw_socket.close()
+
+            if self._receive_future:
+                self._receive_future.set_result(None)
+            if self._send_future:
+                self._send_future.set_result(None)
+
+
+class TCPSocketListener(abc.SocketListener):
+    _accept_scope: Optional[CancelScope] = None
+    _closed = False
+
+    def __init__(self, raw_socket: socket.socket):
+        self.__raw_socket = raw_socket
+        self._loop = cast(asyncio.BaseEventLoop, get_running_loop())
+        self._accept_guard = ResourceGuard("accepting connections from")
+
+    @property
+    def _raw_socket(self) -> socket.socket:
+        return self.__raw_socket
+
+    async def accept(self) -> abc.SocketStream:
+        if self._closed:
+            raise ClosedResourceError
+
+        with self._accept_guard:
+            await checkpoint()
+            with CancelScope() as self._accept_scope:
+                try:
+                    client_sock, _addr = await self._loop.sock_accept(self._raw_socket)
+                except asyncio.CancelledError:
+                    # Workaround for https://bugs.python.org/issue41317
                     try:
-                        client.block(target_user_id=user_id)
-                    except tweepy.errors.TooManyRequests as e:
-                        print(e)
-                        time.sleep(240)
-                        continue
-                    finished = True
-                    me = client.get_me()
-                    print("{} blocked {}".format(me.data["username"], user))
-                    num_users_blocked = num_users_blocked + 1
-
-    return num_users_blocked
-
-username_populated = False
-
-
-def chat(radio_score=None, selected_option=None):
-    client = get_client_from_tokens()
-    history = []
-
-    if radio_score != None and selected_option != None:
-        response = "no blocking"
-        if client != None:
-            chat_history.append(
-                ["Model tuned to a '{}%' threshold and is using the '{}' dataset.".format(radio_score, selected_option),
-                 "{} Account blocking initialised".format(selected_option.capitalize())])
-            num_users_blocked = block_users(client, radio_score, selected_option)
-            chat_history.append(
-                ["Blocked {} user account(s).".format(num_users_blocked), "Thank you for using Watchtower."])
-    elif radio_score != None or selected_option != None:
-        chat_history.append(["Initialisation error!", "Please tune the model by using the above options"])
-
-    return chat_history
-
-
-def infer(prompt):
-    pass
-
-
-have_initialised = False
-client = None
-name = None
-
-
-def changed_tab():
-    global have_initialised
-    global chatbot
-    global chat_history
-    global client
-    global name
-
-    name = "no username"
-
-    chat_history = [
-        ["Welcome to Watchtower.".format(name), "Log in via Twitter and configure your blocking options above."]]
-
-    if client != None and name != "no username":
-        chat_history = [["Welcome {}".format(name), "Initialising WatchTower"]]
-
-        print("changed tabs - {}".format(name))
-        chatbot.value = chat_history
-        chatbot.update(value=chat_history)
-
-
-        client = get_client_from_tokens()
-        name = client.get_me().data.name
-        have_initialised = True
-        chat_history = [["Welcome {}".format(name), "Initialising WatchTower"]]
-
-        chatbot.value = chat_history
-        chatbot.update(value=chat_history)
-
-with block:
-    gr.HTML('''
-
-<meta name="viewport" content="width=device-width, initial-scale=1">
-<link rel="stylesheet" href="https://www.w3schools.com/w3css/4/w3.css">
-    <!-- Navbar (sit on top) -->
-<div class="w3-top">
-  <div class="w3-bar w3-white w3-wide w3-padding w3-card">
-    <p class="w3-bar-item w3-button"><b>WATCH</b> Tower</p>
-  </div>
-</div>
-''')
-    gr.HTML("<center><p><br></p></center>")
-
-    # todo check if user signed in
-
-    user_message = "Log in via Twitter and configure your blocking options above."
-
-    chat_history.append(["Welcome to Watchtower.", user_message])
-    tabs = gr.Tabs()
-    with tabs:
-        intro_tab = gr.TabItem("Introduction")
-        with intro_tab:
-            gr.HTML(test_html)
-
-        prediction_tab = gr.TabItem("Getting Started")
-        with prediction_tab:
-            gr.HTML('''
-            <header class="w3-display-container w3-content w3-wide" style="max-height:250px;" id="home">
-          <img class="w3-image" src="https://cdn.pixabay.com/photo/2018/12/10/16/22/city-3867295_960_720.png" alt="Architecture" width="1500" height="800">
-          <div class="w3-display-middle w3-margin-top w3-center">
-            <h1 class="w3-xxlarge w3-text-white"><span class="w3-padding w3-black w3-opacity-min"><b>WATCH</b></span> <span class="w3-hide-small w3-text-dark-grey">Tower</span></h1>
-          </div>
-        </header>
-''')
-            with gr.Group():
-                with gr.Box():
-                    with gr.Row().style(mobile_collapse=False, equal_height=True):
-                        gr.HTML(
-                            value='<a href={}><img src="https://cdn.cms-twdigitalassets.com/content/dam/developer-twitter/auth-docs/sign-in-with-twitter-gray.png.twimg.1920.png" alt="Log In With Twitter"></a><br>'.format(
-                                target_website))
-                    with gr.Row().style(mobile_collapse=False, equal_height=True):
-                        radio = gr.CheckboxGroup(value="Violent", choices=["Violent", "Hate Speech", "Misinformation"],
-                                                 interactive=False, label="Behaviour To Block")
-
-                        slider = gr.Slider(value=80, label="Threshold Certainty Tolerance")
-
-                    chatbot = gr.Chatbot(value=chat_history, label="Watchtower Output").style()
-                    btn = gr.Button("Run WatchTower").style(full_width=True)
-
-        
-        btn.click(fn=chat, inputs=[slider, radio], outputs=chatbot)
-        tabs.change(fn=changed_tab, inputs=None, outputs=None)
-
-    gr.Markdown(
-        """___
-   <p style='text-align: center'>
-   Created by <a href="https://twitter.com/borisdayma" target="_blank"James Stevenson</a> et al. 2021-2022
-   <br/>
-   <a href="https://github.com/CartographerLabs/Pinpoint" target="_blank">GitHub</a>
-   </p>"""
+                        self._loop.remove_reader(self._raw_socket)
+                    except (ValueError, NotImplementedError):
+                        pass
+
+                    if self._closed:
+                        raise ClosedResourceError from None
+
+                    raise
+                finally:
+                    self._accept_scope = None
+
+        client_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+        transport, protocol = await self._loop.connect_accepted_socket(
+            StreamProtocol, client_sock
+        )
+        return SocketStream(cast(asyncio.Transport, transport), protocol)
+
+    async def aclose(self) -> None:
+        if self._closed:
+            return
+
+        self._closed = True
+        if self._accept_scope:
+            # Workaround for https://bugs.python.org/issue41317
+            try:
+                self._loop.remove_reader(self._raw_socket)
+            except (ValueError, NotImplementedError):
+                pass
+
+            self._accept_scope.cancel()
+            await sleep(0)
+
+        self._raw_socket.close()
+
+
+class UNIXSocketListener(abc.SocketListener):
+    def __init__(self, raw_socket: socket.socket):
+        self.__raw_socket = raw_socket
+        self._loop = get_running_loop()
+        self._accept_guard = ResourceGuard("accepting connections from")
+        self._closed = False
+
+    async def accept(self) -> abc.SocketStream:
+        await checkpoint()
+        with self._accept_guard:
+            while True:
+                try:
+                    client_sock, _ = self.__raw_socket.accept()
+                    client_sock.setblocking(False)
+                    return UNIXSocketStream(client_sock)
+                except BlockingIOError:
+                    f: asyncio.Future = asyncio.Future()
+                    self._loop.add_reader(self.__raw_socket, f.set_result, None)
+                    f.add_done_callback(
+                        lambda _: self._loop.remove_reader(self.__raw_socket)
+                    )
+                    await f
+                except OSError as exc:
+                    if self._closed:
+                        raise ClosedResourceError from None
+                    else:
+                        raise BrokenResourceError from exc
+
+    async def aclose(self) -> None:
+        self._closed = True
+        self.__raw_socket.close()
+
+    @property
+    def _raw_socket(self) -> socket.socket:
+        return self.__raw_socket
+
+
+class UDPSocket(abc.UDPSocket):
+    def __init__(
+        self, transport: asyncio.DatagramTransport, protocol: DatagramProtocol
+    ):
+        self._transport = transport
+        self._protocol = protocol
+        self._receive_guard = ResourceGuard("reading from")
+        self._send_guard = ResourceGuard("writing to")
+        self._closed = False
+
+    @property
+    def _raw_socket(self) -> socket.socket:
+        return self._transport.get_extra_info("socket")
+
+    async def aclose(self) -> None:
+        if not self._transport.is_closing():
+            self._closed = True
+            self._transport.close()
+
+    async def receive(self) -> Tuple[bytes, IPSockAddrType]:
+        with self._receive_guard:
+            await checkpoint()
+
+            # If the buffer is empty, ask for more data
+            if not self._protocol.read_queue and not self._transport.is_closing():
+                self._protocol.read_event.clear()
+                await self._protocol.read_event.wait()
+
+            try:
+                return self._protocol.read_queue.popleft()
+            except IndexError:
+                if self._closed:
+                    raise ClosedResourceError from None
+                else:
+                    raise BrokenResourceError from None
+
+    async def send(self, item: UDPPacketType) -> None:
+        with self._send_guard:
+            await checkpoint()
+            await self._protocol.write_event.wait()
+            if self._closed:
+                raise ClosedResourceError
+            elif self._transport.is_closing():
+                raise BrokenResourceError
+            else:
+                self._transport.sendto(*item)
+
+
+class ConnectedUDPSocket(abc.ConnectedUDPSocket):
+    def __init__(
+        self, transport: asyncio.DatagramTransport, protocol: DatagramProtocol
+    ):
+        self._transport = transport
+        self._protocol = protocol
+        self._receive_guard = ResourceGuard("reading from")
+        self._send_guard = ResourceGuard("writing to")
+        self._closed = False
+
+    @property
+    def _raw_socket(self) -> socket.socket:
+        return self._transport.get_extra_info("socket")
+
+    async def aclose(self) -> None:
+        if not self._transport.is_closing():
+            self._closed = True
+            self._transport.close()
+
+    async def receive(self) -> bytes:
+        with self._receive_guard:
+            await checkpoint()
+
+            # If the buffer is empty, ask for more data
+            if not self._protocol.read_queue and not self._transport.is_closing():
+                self._protocol.read_event.clear()
+                await self._protocol.read_event.wait()
+
+            try:
+                packet = self._protocol.read_queue.popleft()
+            except IndexError:
+                if self._closed:
+                    raise ClosedResourceError from None
+                else:
+                    raise BrokenResourceError from None
+
+            return packet[0]
+
+    async def send(self, item: bytes) -> None:
+        with self._send_guard:
+            await checkpoint()
+            await self._protocol.write_event.wait()
+            if self._closed:
+                raise ClosedResourceError
+            elif self._transport.is_closing():
+                raise BrokenResourceError
+            else:
+                self._transport.sendto(item)
+
+
+async def connect_tcp(
+    host: str, port: int, local_addr: Optional[Tuple[str, int]] = None
+) -> SocketStream:
+    transport, protocol = cast(
+        Tuple[asyncio.Transport, StreamProtocol],
+        await get_running_loop().create_connection(
+            StreamProtocol, host, port, local_addr=local_addr
+        ),
+    )
+    transport.pause_reading()
+    return SocketStream(transport, protocol)
+
+
+async def connect_unix(path: str) -> UNIXSocketStream:
+    await checkpoint()
+    loop = get_running_loop()
+    raw_socket = socket.socket(socket.AF_UNIX)
+    raw_socket.setblocking(False)
+    while True:
+        try:
+            raw_socket.connect(path)
+        except BlockingIOError:
+            f: asyncio.Future = asyncio.Future()
+            loop.add_writer(raw_socket, f.set_result, None)
+            f.add_done_callback(lambda _: loop.remove_writer(raw_socket))
+            await f
+        except BaseException:
+            raw_socket.close()
+            raise
+        else:
+            return UNIXSocketStream(raw_socket)
+
+
+async def create_udp_socket(
+    family: socket.AddressFamily,
+    local_address: Optional[IPSockAddrType],
+    remote_address: Optional[IPSockAddrType],
+    reuse_port: bool,
+) -> Union[UDPSocket, ConnectedUDPSocket]:
+    result = await get_running_loop().create_datagram_endpoint(
+        DatagramProtocol,
+        local_addr=local_address,
+        remote_addr=remote_address,
+        family=family,
+        reuse_port=reuse_port,
     )
+    transport = cast(asyncio.DatagramTransport, result[0])
+    protocol = result[1]
+    if protocol.exception:
+        transport.close()
+        raise protocol.exception
+
+    if not remote_address:
+        return UDPSocket(transport, protocol)
+    else:
+        return ConnectedUDPSocket(transport, protocol)
+
+
+async def getaddrinfo(
+    host: Union[bytes, str],
+    port: Union[str, int, None],
+    *,
+    family: Union[int, AddressFamily] = 0,
+    type: Union[int, SocketKind] = 0,
+    proto: int = 0,
+    flags: int = 0,
+) -> GetAddrInfoReturnType:
+    # https://github.com/python/typeshed/pull/4304
+    result = await get_running_loop().getaddrinfo(
+        host, port, family=family, type=type, proto=proto, flags=flags
+    )
+    return cast(GetAddrInfoReturnType, result)
+
+
+async def getnameinfo(sockaddr: IPSockAddrType, flags: int = 0) -> Tuple[str, str]:
+    return await get_running_loop().getnameinfo(sockaddr, flags)
+
+
+_read_events: RunVar[Dict[Any, asyncio.Event]] = RunVar("read_events")
+_write_events: RunVar[Dict[Any, asyncio.Event]] = RunVar("write_events")
+
+
+async def wait_socket_readable(sock: socket.socket) -> None:
+    await checkpoint()
+    try:
+        read_events = _read_events.get()
+    except LookupError:
+        read_events = {}
+        _read_events.set(read_events)
+
+    if read_events.get(sock):
+        raise BusyResourceError("reading from") from None
+
+    loop = get_running_loop()
+    event = read_events[sock] = asyncio.Event()
+    loop.add_reader(sock, event.set)
+    try:
+        await event.wait()
+    finally:
+        if read_events.pop(sock, None) is not None:
+            loop.remove_reader(sock)
+            readable = True
+        else:
+            readable = False
+
+    if not readable:
+        raise ClosedResourceError
+
+
+async def wait_socket_writable(sock: socket.socket) -> None:
+    await checkpoint()
+    try:
+        write_events = _write_events.get()
+    except LookupError:
+        write_events = {}
+        _write_events.set(write_events)
+
+    if write_events.get(sock):
+        raise BusyResourceError("writing to") from None
+
+    loop = get_running_loop()
+    event = write_events[sock] = asyncio.Event()
+    loop.add_writer(sock.fileno(), event.set)
+    try:
+        await event.wait()
+    finally:
+        if write_events.pop(sock, None) is not None:
+            loop.remove_writer(sock)
+            writable = True
+        else:
+            writable = False
+
+    if not writable:
+        raise ClosedResourceError
+
+
+#
+# Synchronization
+#
+
+
+class Event(BaseEvent):
+    def __new__(cls) -> "Event":
+        return object.__new__(cls)
+
+    def __init__(self) -> None:
+        self._event = asyncio.Event()
+
+    def set(self) -> DeprecatedAwaitable:
+        self._event.set()
+        return DeprecatedAwaitable(self.set)
+
+    def is_set(self) -> bool:
+        return self._event.is_set()
+
+    async def wait(self) -> None:
+        if await self._event.wait():
+            await checkpoint()
+
+    def statistics(self) -> EventStatistics:
+        return EventStatistics(len(self._event._waiters))  # type: ignore[attr-defined]
+
+
+class CapacityLimiter(BaseCapacityLimiter):
+    _total_tokens: float = 0
+
+    def __new__(cls, total_tokens: float) -> "CapacityLimiter":
+        return object.__new__(cls)
+
+    def __init__(self, total_tokens: float):
+        self._borrowers: Set[Any] = set()
+        self._wait_queue: Dict[Any, asyncio.Event] = OrderedDict()
+        self.total_tokens = total_tokens
+
+    async def __aenter__(self) -> None:
+        await self.acquire()
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> None:
+        self.release()
+
+    @property
+    def total_tokens(self) -> float:
+        return self._total_tokens
+
+    @total_tokens.setter
+    def total_tokens(self, value: float) -> None:
+        if not isinstance(value, int) and not math.isinf(value):
+            raise TypeError("total_tokens must be an int or math.inf")
+        if value < 1:
+            raise ValueError("total_tokens must be >= 1")
+
+        old_value = self._total_tokens
+        self._total_tokens = value
+        events = []
+        for event in self._wait_queue.values():
+            if value <= old_value:
+                break
+
+            if not event.is_set():
+                events.append(event)
+                old_value += 1
+
+        for event in events:
+            event.set()
+
+    @property
+    def borrowed_tokens(self) -> int:
+        return len(self._borrowers)
+
+    @property
+    def available_tokens(self) -> float:
+        return self._total_tokens - len(self._borrowers)
+
+    def acquire_nowait(self) -> DeprecatedAwaitable:
+        self.acquire_on_behalf_of_nowait(current_task())
+        return DeprecatedAwaitable(self.acquire_nowait)
+
+    def acquire_on_behalf_of_nowait(self, borrower: object) -> DeprecatedAwaitable:
+        if borrower in self._borrowers:
+            raise RuntimeError(
+                "this borrower is already holding one of this CapacityLimiter's "
+                "tokens"
+            )
+
+        if self._wait_queue or len(self._borrowers) >= self._total_tokens:
+            raise WouldBlock
+
+        self._borrowers.add(borrower)
+        return DeprecatedAwaitable(self.acquire_on_behalf_of_nowait)
+
+    async def acquire(self) -> None:
+        return await self.acquire_on_behalf_of(current_task())
+
+    async def acquire_on_behalf_of(self, borrower: object) -> None:
+        await checkpoint_if_cancelled()
+        try:
+            self.acquire_on_behalf_of_nowait(borrower)
+        except WouldBlock:
+            event = asyncio.Event()
+            self._wait_queue[borrower] = event
+            try:
+                await event.wait()
+            except BaseException:
+                self._wait_queue.pop(borrower, None)
+                raise
+
+            self._borrowers.add(borrower)
+        else:
+            try:
+                await cancel_shielded_checkpoint()
+            except BaseException:
+                self.release()
+                raise
+
+    def release(self) -> None:
+        self.release_on_behalf_of(current_task())
+
+    def release_on_behalf_of(self, borrower: object) -> None:
+        try:
+            self._borrowers.remove(borrower)
+        except KeyError:
+            raise RuntimeError(
+                "this borrower isn't holding any of this CapacityLimiter's " "tokens"
+            ) from None
+
+        # Notify the next task in line if this limiter has free capacity now
+        if self._wait_queue and len(self._borrowers) < self._total_tokens:
+            event = self._wait_queue.popitem()[1]
+            event.set()
+
+    def statistics(self) -> CapacityLimiterStatistics:
+        return CapacityLimiterStatistics(
+            self.borrowed_tokens,
+            self.total_tokens,
+            tuple(self._borrowers),
+            len(self._wait_queue),
+        )
+
+
+_default_thread_limiter: RunVar[CapacityLimiter] = RunVar("_default_thread_limiter")
+
+
+def current_default_thread_limiter() -> CapacityLimiter:
+    try:
+        return _default_thread_limiter.get()
+    except LookupError:
+        limiter = CapacityLimiter(40)
+        _default_thread_limiter.set(limiter)
+        return limiter
+
+
+#
+# Operating system signals
+#
+
+
+class _SignalReceiver(DeprecatedAsyncContextManager["_SignalReceiver"]):
+    def __init__(self, signals: Tuple[int, ...]):
+        self._signals = signals
+        self._loop = get_running_loop()
+        self._signal_queue: Deque[int] = deque()
+        self._future: asyncio.Future = asyncio.Future()
+        self._handled_signals: Set[int] = set()
+
+    def _deliver(self, signum: int) -> None:
+        self._signal_queue.append(signum)
+        if not self._future.done():
+            self._future.set_result(None)
+
+    def __enter__(self) -> "_SignalReceiver":
+        for sig in set(self._signals):
+            self._loop.add_signal_handler(sig, self._deliver, sig)
+            self._handled_signals.add(sig)
+
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> Optional[bool]:
+        for sig in self._handled_signals:
+            self._loop.remove_signal_handler(sig)
+        return None
+
+    def __aiter__(self) -> "_SignalReceiver":
+        return self
+
+    async def __anext__(self) -> int:
+        await checkpoint()
+        if not self._signal_queue:
+            self._future = asyncio.Future()
+            await self._future
+
+        return self._signal_queue.popleft()
+
+
+def open_signal_receiver(*signals: int) -> _SignalReceiver:
+    return _SignalReceiver(signals)
+
+
+#
+# Testing and debugging
+#
+
+
+def _create_task_info(task: asyncio.Task) -> TaskInfo:
+    task_state = _task_states.get(task)
+    if task_state is None:
+        name = task.get_name() if _native_task_names else None
+        parent_id = None
+    else:
+        name = task_state.name
+        parent_id = task_state.parent_id
+
+    return TaskInfo(id(task), parent_id, name, get_coro(task))
+
+
+def get_current_task() -> TaskInfo:
+    return _create_task_info(current_task())  # type: ignore[arg-type]
+
+
+def get_running_tasks() -> List[TaskInfo]:
+    return [_create_task_info(task) for task in all_tasks() if not task.done()]
+
+
+async def wait_all_tasks_blocked() -> None:
+    await checkpoint()
+    this_task = current_task()
+    while True:
+        for task in all_tasks():
+            if task is this_task:
+                continue
+
+            if task._fut_waiter is None or task._fut_waiter.done():  # type: ignore[attr-defined]
+                await sleep(0.1)
+                break
+        else:
+            return
+
+
+class TestRunner(abc.TestRunner):
+    def __init__(
+        self,
+        debug: bool = False,
+        use_uvloop: bool = False,
+        policy: Optional[asyncio.AbstractEventLoopPolicy] = None,
+    ):
+        self._exceptions: List[BaseException] = []
+        _maybe_set_event_loop_policy(policy, use_uvloop)
+        self._loop = asyncio.new_event_loop()
+        self._loop.set_debug(debug)
+        self._loop.set_exception_handler(self._exception_handler)
+        asyncio.set_event_loop(self._loop)
+
+    def _cancel_all_tasks(self) -> None:
+        to_cancel = all_tasks(self._loop)
+        if not to_cancel:
+            return
+
+        for task in to_cancel:
+            task.cancel()
+
+        self._loop.run_until_complete(
+            asyncio.gather(*to_cancel, return_exceptions=True)
+        )
+
+        for task in to_cancel:
+            if task.cancelled():
+                continue
+            if task.exception() is not None:
+                raise cast(BaseException, task.exception())
+
+    def _exception_handler(
+        self, loop: asyncio.AbstractEventLoop, context: Dict[str, Any]
+    ) -> None:
+        if isinstance(context.get("exception"), Exception):
+            self._exceptions.append(context["exception"])
+        else:
+            loop.default_exception_handler(context)
+
+    def _raise_async_exceptions(self) -> None:
+        # Re-raise any exceptions raised in asynchronous callbacks
+        if self._exceptions:
+            exceptions, self._exceptions = self._exceptions, []
+            if len(exceptions) == 1:
+                raise exceptions[0]
+            elif exceptions:
+                raise ExceptionGroup(exceptions)
+
+    def close(self) -> None:
+        try:
+            self._cancel_all_tasks()
+            self._loop.run_until_complete(self._loop.shutdown_asyncgens())
+        finally:
+            asyncio.set_event_loop(None)
+            self._loop.close()
+
+    def run_asyncgen_fixture(
+        self,
+        fixture_func: Callable[..., AsyncGenerator[T_Retval, Any]],
+        kwargs: Dict[str, Any],
+    ) -> Iterable[T_Retval]:
+        async def fixture_runner() -> None:
+            agen = fixture_func(**kwargs)
+            try:
+                retval = await agen.asend(None)
+                self._raise_async_exceptions()
+            except BaseException as exc:
+                f.set_exception(exc)
+                return
+            else:
+                f.set_result(retval)
+
+            await event.wait()
+            try:
+                await agen.asend(None)
+            except StopAsyncIteration:
+                pass
+            else:
+                await agen.aclose()
+                raise RuntimeError("Async generator fixture did not stop")
+
+        f = self._loop.create_future()
+        event = asyncio.Event()
+        fixture_task = self._loop.create_task(fixture_runner())
+        self._loop.run_until_complete(f)
+        yield f.result()
+        event.set()
+        self._loop.run_until_complete(fixture_task)
+        self._raise_async_exceptions()
+
+    def run_fixture(
+        self,
+        fixture_func: Callable[..., Coroutine[Any, Any, T_Retval]],
+        kwargs: Dict[str, Any],
+    ) -> T_Retval:
+        retval = self._loop.run_until_complete(fixture_func(**kwargs))
+        self._raise_async_exceptions()
+        return retval
+
+    def run_test(
+        self, test_func: Callable[..., Coroutine[Any, Any, Any]], kwargs: Dict[str, Any]
+    ) -> None:
+        try:
+            self._loop.run_until_complete(test_func(**kwargs))
+        except Exception as exc:
+            self._exceptions.append(exc)
 
-block.launch(enable_queue=False)
+        self._raise_async_exceptions()