assert isinstance(str, type)
assert isinstance(list, type)
assert isinstance(dict, type)
assert isinstance(int, type)
assert isinstance(object, type)

assert issubclass(str, object)
assert issubclass(list, object)
assert issubclass(type, object)

class myClass(object):
    def __init__(self, initArgs:str):
        pass

myClassObj = myClass("initArgs")

class myClass(object):
    def __init__(self, initArgs:str):
        print(id(self))

myClassObj = myClass("initArgs")
print(id(myClassObj))
# 输出两行内存地址，结果一致

class type(object):
    def __call__(cls: Type, *args: Any, **kwargs: Any) -> Any:
        self = cls.__new__(cls)
        if hasattr(self, "__init__"):
            self.__init__(*args, **kwargs)
        return self

import abc
import torch
from torch.nn import Module
class NetMetaBase(abc.ABCMeta):
    def __call__(
        self:Type[Module], 
        inChannels:int, 
        outChannels:int, 
        *args: Any, **kwargs: Any) -> Module:
        """
        该方法创建self参数指定类的实例对象
        Args:
            self 即为NetMetaBase的实例，也就是以NetMetaBase为元类的类对象
            
        """
        assert issubclass(self, Module), "Class must be a subclass of torch.nn.Module"
        obj = self.__new__(self)
        obj.__init__(inChannels, outChannels, *args, **kwargs)
        return obj

class ModuleBase(Module, metaclass=NetMetaBase):
    def __init__(self, inChannels:int, outChannels:int, **kwargs)->None:
        super(ModuleBase, self).__init__()
        self._inChannels = inChannels
        self._outChannels = outChannels
    
    @abc.abstractmethod
    def forward(self, x:torch.Tensor)->torch.Tensor:
        raise NotImplementedError

class A(object):
    def __init__(self):
        print("A")
        print(self.__class__)

class B(object):
    def __init__(self):
        print("B")
        print(self.__class__)

class C(A, B):
    def __init__(self):
        super(C, self).__init__()
        super(A, self).__init__()
        print("C")
        print(self.__class__)

C()
assert C.__mro__ == (C, A, B, object)
# 输出如下
# A
# <class '__main__.C'>
# B
# <class '__main__.C'>
# C
# <class '__main__.C'>

super(C, self).__init__()
A.__init__(self)

import _thread as th
import time

def print_time( threadName, delay):
   count = 0
   while count < 5:
      time.sleep(delay)
      count += 1
      print ("%s: %s" % ( threadName, time.ctime(time.time()) ))

th.start_new_thread(print_time, ("Thread-1", 2))
th.start_new_thread(print_time, ("Thread-2", 2))

import threading

class timeThread(threading.Thread):
    def __init__(self, threadName):
        threading.Thread.__init__(self)
        # self.name = threadName
        self.setName(threadName)
    
    def run(self):
        print_time(self.getName(), 2)

thread1 = timeThread("Thread-1")
thread2 = timeThread("Thread-2")

thread1.start()
thread2.start()
print("Two thread created and started")
thread1.join()
thread2.join()
print("Main thread ended")

class timeLockThread(threading.Thread):
    lock = threading.Lock()
    def __init__(self, threadName):
        threading.Thread.__init__(self)
        self.setName(threadName)
    
    def run(self):
        print(self.getName(),"准备获取锁")
        self.lock.acquire()
        print(self.getName(),"获取锁成功")
        print_time(self.getName(), 2)
        print(self.getName(),"准备释放锁")
        self.lock.release()
        print(self.getName(),"释放锁成功")

lockthread1 = timeLockThread("Thread-3")
lockthread2 = timeLockThread("Thread-4")

lockthread1.start()
lockthread2.start()
print("Two thread created and started")
lockthread1.join()
lockthread2.join()
print("Main thread ended")

class Executor(object):
    """This is an abstract base class for concrete asynchronous executors."""
    def submit(*args, **kwargs)->Future:

    def map(self, fn, *iterables, timeout=None, chunksize=1):

    def shutdown(self, wait=True):

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.shutdown(wait=True)
        return False

import time
from concurrent.futures import ThreadPoolExecutor, Future
def task(a, b, sleeps = 5):
    time.sleep(sleeps)
    return a + b
pool = ThreadPoolExecutor(max_workers = 8)
task: Future = pool.submit(task, a = 5, b = 6, sleeps = 6)
print(f"a + b = {task.result}")
pool.shutdown(wait=True)

import time
import os
from concurrent.futures import ProcessPoolExecutor, Future
def task1(a, b, sleeps = 5):
    print(f"PID is {os.getpid()}, PPID is {os.getppid()}")
    time.sleep(sleeps)
    return a + b

if __name__ == "__main__":
    print(f"PID is {os.getpid()}")
    pool = ProcessPoolExecutor(max_workers = 8)
    task: Future = pool.submit(task1, a = 5, b = 6, sleeps = 6)
    print(f"a + b = {task.result()}")
    pool.shutdown(wait=True)

import time
import os
from concurrent.futures import ProcessPoolExecutor, Future
def task1(a, b, sleeps = 5):
    print(f"PID is {os.getpid()}, PPID is {os.getppid()}")
    time.sleep(sleeps)
    return a + b

print(f"Import {__name__} in PID is {os.getpid()}")
if __name__ == "__main__":
    pool = ProcessPoolExecutor(max_workers = 8)
    task: Future = pool.submit(task1, a = 5, b = 6, sleeps = 6)
    print(f"a + b = {task.result()}")
    pool.shutdown(wait=True)

import time
import os
from concurrent.futures import ProcessPoolExecutor, Future
def task1(a, b, sleeps = 5):
    print(f"PID is {os.getpid()}, PPID is {os.getppid()}")
    time.sleep(sleeps)
    return a + b

print(f"Import {__name__} in PID is {os.getpid()}")
pool = ProcessPoolExecutor(max_workers = 8)
task: Future = pool.submit(task1, a = 5, b = 6, sleeps = 6)
print(f"a + b = {task.result()}")
pool.shutdown(wait=True)

import functools
import os
import time
from concurrent.futures import ProcessPoolExecutor, Future
def parallel(func):
      @functools.wraps(func)
      def wrapper(*args, **kwargs):
            pool = ProcessPoolExecutor(max_workers=2)
            task = pool.submit(func, *args, **kwargs)
            return task
      return wrapper

@parallel
def task1(a, b, sleeps = 5):
    print(f"PID is {os.getpid()}, PPID is {os.getppid()}")
    time.sleep(sleeps)
    return a + b

if __name__ == "__main__":
    task: Future = task1(a=5, b= 6)
    print(task.result())