py 简单记录

def print_tools():
    print("hello")
    print(sys)

    logf = open("cat.txt", "a")
    print("log to file", file=logf)
    logf.close()

    print("a", "b", 2000, sep=",", flush=True)
    print("error out", file=sys.stderr)

    csv_file = open("python.csv", "a")
    print("a", "b", 2000, sep=",", file=csv_file)
    print("z", "b", 456, sep=",", file=csv_file)
    print("x", "b", 33, sep=",", file=csv_file)
    csv_file.close()

    print("aaa"
          "bbbb")

    num = 2 + 3j
   # num=complex(2,3)
    print(type(num.imag),type(num.real))

def fmt_paramster(p1,*fmt, **kw) :
    # fmt 可变参数，kw关键字参数
    print(p1,fmt) # this is param ('a', 'b')
    print("fmt type is",type(fmt))  #  type is <class 'tuple'>
    print("kw type is",type(kw))    # kw type is <class 'dict'>

fmt_paramster("this is param","a","b",name="william")

gx = 100
def outfun(a):
    c = 1000
    global gx;
    gx +=1

    def infun(b):
        nonlocal  c
        return a + b + c
    return infun;

# def lamda_test():
#     return lambda x,y:x + y
# lm=lamda_test();
# print(lm(1,2))

fun = outfun(1)
ve = fun(2)
print(ve)

def import_module():
    import sys, os
    path = os.getcwd() + os.sep + "module"
    sys.path.insert(0, path)
    print(sys.path)
    import module
    module.pymodule()

import packet
packet.init()

# 引入包的命名空间
import packet.video
packet.video.play_video()

# 引入包和模块的命名空间,不导入包中的模块，如要加入写__all__=["***","***"]
from packet import *
packet.opengl.draw()

from packet import audio
audio.play_audio()

from packet.audio import play_audio
play_audio()

from multiprocessing import Process
import time,os
def proc_fun(name):
    for i in range(10):
        print(i, "child proc",name,os.getpid())
        time.sleep(1)
       
if __name__ == "__main__":
    proc = Process(target=proc_fun,args=("name001",))
    proc.start()
    for i in range(10):
        print(i, "parent proc",os.getpid())
        time.sleep(1)

from threading import Thread,Lock
from queue import Queue
import time

is_running = True
mux = Lock()
products = Queue()

class Producer(Thread):
    def __int__(self):
        Thread.__init__(self)
        print("Producer Created!")
    def run(self):
        product_id = 1000
        while is_running:
            mux.acquire()
            print("produce a product",product_id)
            product_id = product_id +1
            products.put(product_id)
            mux.release()
            time.sleep(1)
        print("Producer thread exited")

class Consumer(Thread):
    def __int__(self):
        Thread.__init__(self)
        print("Consumer Created!")
    def run(self):
        while is_running:
            mux.acquire()
            if products.empty():
                mux.release()
                continue

            print("Consumer a product",products.get())
            mux.release()
            time.sleep(1)
        print("Consumer thread exited")

producer = Producer()
producer.start()
consumer = Consumer()
consumer.start()

ths=[producer,consumer]
time.sleep(6)
is_running = False
#the main thead wait for the child thread exit
for th in ths:
    th.join() # wait for the thread exit
print("main thread exited")

class XError(Exception):
    def __init__(self,value=""):
        self.value= value
    def __str__(self):
        return "XError:" + str(self.value)

try:
    raise XError("Xerror for testing")
except XError as e:
    print(e)

try:
    pass
except:
    pass
else:
    print("no error")