malcolmgreaves

import threading
from abc import ABC
from concurrent.futures import Executor, Future
from multiprocessing import synchronize
from typing import Optional, Sequence, Union

__all__: Sequence[str] = ("BoundedPoolExecutor",)


class BoundedPoolExecutor(ABC, Executor):

malcolmgreaves

from abc import ABC
from typing import Any, Iterable, Tuple
from dataclasses import dataclass
from functools import partial


@dataclass
class TupleDestructureSyntax(ABC):
  """Allows any @dataclass class to be decomposed into its fields & values as if it were a tuple.
  

malcolmgreaves

from enum import Enum
from typing import Callable, Optional, Type, TypeVar


class EnumStr(str, Enum):
  pass


E = TypeVar('E', bound=EnumStr)

malcolmgreaves

"""
Use as:

    REPO='ecr-repo-name' BRANCH='main' python git_branch_commits_in_ecr_repo.py

You can pipe this to `cut -f1 -s` to get the commits only. Or `-f2` to get the pushed by date only.
These values are separated by a tab (`"\t"`).
"""
import json
import os

malcolmgreaves

import traceback
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Iterator, Sequence, Union

import torch
from core_utils.common import type_name

__all__: Sequence[str] = (
    "descent_script",

malcolmgreaves

async def heartbeat(interval=5):
    while True:
        await asyncio.sleep(interval)

        
async def safe():
    loop = asyncio.get_running_loop()
    h = loop.create_task(heartbeat())

    await asyncio.sleep(1)

malcolmgreaves

import os
import sys
from multiprocessing import Process
from typing import Callable, Optional, Sequence


def make_logger():
    import logging

    log_level = logging.INFO

malcolmgreaves

import os
from collections import Counter
from pathlib import Path
from typing import Iterator, List, Sequence, Tuple


def calculate_greatest_common_subpath(files: Sequence[Path]) -> Path:
    """Find the longest common subpath amongst a collection of files."""
    if len(files) == 0:
        raise ValueError("Must input at least one file!")

malcolmgreaves

 def cache_once(func):
    """Decorator that caches the result of a unary function or method.

    REQUIREMENTS: The decorated function must be **referentially transparent**. I.e. the same arguments
                  result in the same output result, **every time**. With this guarantee, it is safe to
                  cache the value of a unary function from a single execution. If your funciton has any
                  side effects, it is not referentially transparent. Avoid any dependencies on mutable
                  state, I/O, etc.

    To use, you simply wrap an existing function:

malcolmgreaves

import logging
import threading
import time
import traceback
from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
from contextlib import ExitStack
from dataclasses import dataclass
from time import sleep
from typing import Any, Callable, ContextManager, List, NamedTuple, Optional, Sequence, Set