ktbarrett

function logerr {
  file="$1"
  shift 1
  "$@" 2> >(tee "$file" >&2)
}

ktbarrett

from typing import Optional, Tuple, Union, overload


_time_unit_map = {
    "ps": -12,
    "ns": -9,
    "us": -6,
    "ms": -3,
    "s": 0,
}

ktbarrett

from functools import lru_cache
from typing import Dict, Hashable, List, Optional, TypeVar

T = TypeVar("T", bound=Hashable)


def c3_linearize(dep_tree: Dict[T, List[T]], root: T) -> List[T]:
    """Linearize a dependency tree"""

    @lru_cache(None)

ktbarrett

from functools import cache, cached_property
from typing import (
    Collection,
    Iterable,
    Optional,
    Reversible,
    Sized,
    TypeAlias,
    TypeVar,
    overload,

ktbarrett

#include <variant>

template <typename ...Args>
struct sum_;

template <typename Last>
struct sum_<Last> {
    using type = Last;
};

ktbarrett

from functools import update_wrapper, wraps

class cached_method:

    def __init__(self, method):
        self._method = method
        update_wrapper(self, method)

    def __get__(self, instance, objtype=None):
        if instance is None:

ktbarrett

class TaskManager(Mapping[str, Task[Any]]):
    """Associative container of names to child Tasks.

    Tasks created with this object are not allowed to leak;
    child tasks will be cancelled if the TaskManager is collected.
    Typically this is used as an async context manager,
    forcing all child Tasks to finish before returning.