Generate Correlated Synthetic Data
October 9, 2024About 2 min
The nature of Async is an Event Loop.
The result is a coroutine object after invoke a async
function. The function can only be executed after becoming a task.
import asyncio
import time
async def say_after(delay, what):
await asyncio.sleep(delay)
return f"{what} - {delay}"
async def main():
print(f"started at {time.strftime('%X')}")
await say_after(1, "hello")
await say_after(2, "world")
print(f"finished at {time.strftime('%X')}")
asyncio.run(main())
import hashlib
import os
def getmd5(filename):
# Open,close, read file and calculate MD5 on its contents
with open(filename, 'rb') as file_to_check:
# read contents of the file
data = file_to_check.read()
# pipe contents of the file through
return hashlib.md5(data).hexdigest()
def fsmd5(fpath):
ret = dict()
for p, folders, fs in os.walk(fpath):
for i in fs:
fullpath = os.path.join(p,i)
relative_f = fullpath[len(fpath)+1:]
ret[relative_f] = getmd5(fullpath)
return ret
__init__.py
from .base import base
import os
import importlib.util
import sys
import inspect
import threading
import copy
class class_loader:
_instance_lock = threading.Lock()
def __init__(self):
if hasattr(self, '_tag_'):
return
self._tag_ = dict()
self.__load__()
def __load__(self):
ret = dict()
filepath, _ = os.path.split(__file__)
_, module_name = os.path.split(filepath)
files = filter(lambda x: x.endswith('.py'), os.listdir(filepath))
files = filter(lambda x: not x.startswith('__'), files)
files = filter(lambda x: not x.startswith('base'), files)
#print(list(files))
for f in files:
fullpath = os.path.join(filepath, f)
name, __ = os.path.splitext(f)
spec = importlib.util.spec_from_file_location(f'{module_name}.{name}', fullpath)
mo = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mo)
# get all classes in module mo
clses = filter(lambda x: inspect.isclass(getattr(mo, x)), dir(mo))
# Get all classes derived from base
clses = filter(lambda x: issubclass(getattr(mo, x), base) and not x=='base', clses)
# Get all classes has class attribute TASK_NAME
clses = filter(lambda x: hasattr(getattr(mo, x), 'TASK_NAME'), clses)
for cls in clses:
class_type = getattr(mo, cls)
ret[class_type.TASK_NAME] = class_type
self._tag_.update(ret)
def getTaskMap(self):
return copy.copy(self._tag_)
#Single Class
def __new__(cls, *args, **kwargs):
if not hasattr(eval(cls.__name__), "_instance"):
with eval(cls.__name__)._instance_lock:
if not hasattr(eval(cls.__name__), "_instance"):
eval(cls.__name__)._instance = object.__new__(cls)
return eval(cls.__name__)._instance
def __del__(self):
pass
def getTaskIns(task_name):
tasks = class_loader().getTaskMap()
if tasks.__contains__(task_name):
return tasks[task_name]()
return None
ins = getTaskIns('Task-A')
ins.say()
pip install jupyter
jupyter notebook
jupyter notebook --generate-config
in cmd, generate config file ~\.jupyter\jupyter_notebook_config.py
#c.NotebookApp.notebook_dir = ''
in this filec.NotebookApp.notebook_dir = '{Folder path}'
from base64 import b64encode
# Authorization token is using base 64 to encode
def basic_auth(username, password):
token = b64encode(f"{username}:{password}".encode('utf-8')).decode("ascii")
return f'Basic {token}'