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.

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

#include <variant>

template <typename ...Args>
struct sum_;

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

ktbarrett

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

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 typing import Optional, Tuple, Union, overload


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

ktbarrett

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

ktbarrett

library ieee;
  use ieee.std_logic_1164.all;

package generic_ff_pack is
  procedure FF
    generic (
      type T)
    parameter (
      signal   q         : out   T;
      constant d         : in    T;

ktbarrett

entity piecewise_function is
    generic (
        NODE_MAP            :       NodeMapType;
        NODE_ID             :       string;
        NUM_SEGMENTS        :       natural;
        THRESHOLD_SIZE_RES  :       sfixed;
        BIAS_SIZE_RES       :       sfixed;
        GAIN_SIZE_RES       :       sfixed);
    port (

ktbarrett

async def test(dut):

    with TaskManager() as tm:

        @tm.fork
        async def stimulate():
            pass # stimulate an interface

        @tm.fork