piroyoung

import abc
import functools
import json
import time
import uuid
from dataclasses import asdict
from dataclasses import dataclass
from logging import INFO
from logging import Logger
from logging import basicConfig

piroyoung

from typing import Callable

from numpy import array
from numpy import ndarray


def dimension(*expected: int):
    def wrap_up(func: Callable) -> Callable:
        def wrapped(*args, **kwargs) -> ndarray:
            result: ndarray = func(*args, **kwargs)

piroyoung

import tensorflow as tf


class VarLenSeries:
    """
    sess = tf.InteractiveSession()
    series = tf.constant(list(range(5)))
    sess.run(VarLenSeries(series).to_skip_gram(3))

    # # it returns below

piroyoung

import queue
import threading
from typing import Callable, Generator, Any

_TERM = threading.Event()


class Worker(threading.Thread):
    def __init__(self, source: queue.Queue, destination: queue.Queue, batch_size: int,
                 callback: Callable[[Generator], Any]):

piroyoung

import redis
import time
from flask import jsonify, Flask

class EventCounter:
    def __init__(self, seconds: float):
        pool: redis.ConnectionPool = redis.ConnectionPool(host='localhost', port=6379, db=0)
        self._r: redis.StrictRedis = redis.StrictRedis(connection_pool=pool)
        self._s: float = seconds
        

piroyoung

import threading
import contextlib


class LeakObject:
    
    def __init__(self, lifetime: int, cls: object, **kwargs):
        self._counter = 0
        self.lifetime = lifetime
        self._kwargs = kwargs

piroyoung

import threading
import tracemalloc
import time

tracemalloc.start()

class Sub(threading.Thread):
    def run(self):
        while True:
            time.sleep(3)

piroyoung

import time
from typing import Callable, Any


def print_timestamp(func: Callable) -> Callable:
    def wrap_up(**kwargs) -> Any:
        t0 = time.time()
        res = func(**kwargs)
        t1 = time.time()
        logger.debug({'elapsed_time': t1-t0})

piroyoung

def retry_if_fail(callback: Callable):
    """
    bigtable sdk auto-reconnect works only if failed.
    :param callback: some callback function
    :return:
    """

    def wrap(*args, **kwargs):
        try:
            return callback(*args, **kwargs)

piroyoung

from typing import List, Callable, TypeVar
from functools import reduce, lru_cache
T = TypeVar('T')
class MyCollection(List[T]):
    def reduce(self, f: Callable[[T, T], T]):
        return reduce(f, self)

# example usage
obj = MyCollection(range(5))
obj.reduce(lambda x, y: x + y)