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

malcolmgreaves

from threading import Thread, Lock
from Queue import Queue
import itertools, traceback, multiprocessing

class FixedThreadPoolExecutor(object):
    """
    Executes tasks in a fixed thread pool.
    
    Makes sure to gather all returned results and thrown exceptions in one place, in order of task
    submission.

malcolmgreaves

# Uses `strip_margin` defined here:
# https://gist.github.com/malcolmgreaves/7c9ec3407a02a3eb6d4b898bc16dc902

import subprocess


def exec_bash_script(
    script_contents: str, *, prepend_bash_shebang_if_needed: bool = True
) -> subprocess.CompletedProcess:
    script_contents = script_contents.strip()

malcolmgreaves

from typing import List


def strip_margin(long_multiline_string: str) -> str:
    bits = long_multiline_string.strip().split('\n')

    stripped: List[str] = []
    for b in bits:
        for i in range(len(b)):
            c = b[i]

malcolmgreaves

from dataclasses import *
from abc import *
from pathlib import Path
from typing import TypeVar, Optional, Union, Type, Any, Generic

ServiceClass = TypeVar("ServiceClass")

T = TypeVar("T")