Python 工具库使用 用 magic vs filetype 实现视频类型判断 magic 比filetype更好用，能判断更多的类型，能够直接从文件buffer中判断 import magic # pip install python-magic import filetype # pip install filetype src_file = "x.some" print(filetype.guess(src_file)) print(magic.from_file(src_file, mime=True)) print(magic.from_buffer(open(src_file, 'rb'), mime=True)) jupyterlab IDE 函数定义跳转工具 jupyterlab-lsp parallel_apply[依赖pandarallel库] 对于多CPU机器，加速DataFrame的apply方法 安装python包，不更新其他依赖[--no-dependencies] pip3 install torch-fidelity --no-dependencies

流畅的Python笔记 以代码和注释方式，记录《流畅的Python》的笔记，持续更新 第一章 Python的数据类型 魔术方法(magic methods)是形式为"xxx()"的函数为Python的内置函数，用于实现一些特定的功能 用魔术方法实现可迭代对象 class MyList(): """ 通过实现'__getitem__()'和'__len__()' 完成 """ def __init__(self): self.data = range(0,100) def __getitem__(self, index): """ 实现该方法之后，对象变成可迭代和切片,e.g. print(MyList()[0]) for x in MyList(): print(x) """ return self.data[index] def __len__(self): """ 实现该方法之后，可以用len求对象的长度,e.g. len(MyList()) """ return len(self.data) 用魔术方法实现运算符重载 class Vector(): def __init__(self, x, y): self.x = x self.y = y def __repr__(self): """在print 对象的时候的输出格式， 如果没有__repr__方法定义，会输出对象在内存中的地址 """ return "{}(x = {}, y = {})".format(self.__class__.__name__, self.x, self.y) def __add__(self, vector): """ 重载加法+运算符 """ x = self.x + vector.x y = self.y + vector.y return Vector(x, y) def __mul__(self, scalar): """ 重载乘法*运算符 """ return Vector(self.x * scalar, self.y * scalar) print(Vector(1,2)) #Vector(x = 1, y = 2) print(Vector(1,2) + Vector(3,4)) #Vector(x = 4, y = 6) print(Vector(1,2)*0.5) #Vector(x = 0.5, y = 1.0) 第二章 序列构成的数组 两种生成序列的方法 # 利用生成器生成的序列(元组构建生成器)可以节约内存，不会在内存中创建所有需要的元素，当列表长度过长时，优先选择生成器方式 x_generator = (i for i in range(10000000)) # 生成器，type(x_generator)=<class 'generator'> x_list = [i for i in range(10000000)] # 列表, type(x_list)=<class 'list'> 利用list[::-1]实现反转的原理 列表切片的常规用法，list[start:end:stride], list[::-1]省略了前两个，-1表示取值的步长，因此可以实现列表反转的功能 利用count实现列表/元组计数 l = [1, 1, 2, 3] t = (1, 1, 2, 3) print({x:l.count(x) for x in set(l)}) print({x:t.count(x) for x in set(t)}) 对比 list、tuple、queue、deque(collections) # TODO 第三章 字典和集合 几种不同的dict from collections import defaultdict from collections import OrderDict # {} 内置字典 # OrderDict 是有序字典 # defaultdict #当key不在字段内，默认构建初始值 dict/set vs list 内部维护了散列表，以空间换时间。代码中涉及在大数组中查找key时，可以考虑为dict实现。

Python标准库threading threading 用来管理一个进程中的并行操作(线程) 在python中简单使用线程 import threading def worker(): print('Worker') threads = [] for i in range(5): t = threading.Thread(target = worker) threads.append(t) t.start() target参数是线程调用的函数，threading.Thread还可以通过name指定线程名称，args指定调用函数传入的参数。 start()函数，开始执行线程。另外，theading 中daemon(设置线程是否为后台线程)、join(后台线程阻塞)也是常用的用法。此处就不一一举例了。 线程同步 多线程的程序面临一个重要的问题，由于线程之间的资源是共享的，当不同线程访问共享资源时可能出现访问冲突(例如消费者和生产者的执行)。另外，当线程之间可能需要等待某个事件发生后，才被激活而执行(例如行人、车辆必须等待绿灯才能通过) theading中有lock、Rlock、condtion、Semaphore、Event实现线程的同步，当然各个方法还是有不同的特点。 首先，我们通过一个简单的栗子来感受下，多线程不进行同步可能出现的问题。 import threading import time count =0 def addcount(): global count for i in range(1000): count += 1 for i in range(10): t = threading.Thread(target=addcount) t.start() time.sleep(3) print count 理论上上述代码的结果应该是1000*10,但是当我们运行程序发现每次的结果都不同，并且结果都小于10000。对全局变量的修改可以分为取值和赋值两个阶段，假设某一时刻count=100，如果按照方式①Thread1取值(temp1=100)-->Thread1赋值(count=temp1+1)-->Thread2取值(temp2=101)-->Thread2赋值(count=temp2+1)这种方式，运行结果count为102。但是如果线程的顺序变为方式②Thread1取值(temp1=100)-->Thread2取值(temp2=100)-->Thread1赋值(count=temp1+1)-->Thread2赋值(count=temp2+1)，运行结果count为101。 如果想要使得程序严格按照方式①的顺序运行，我们就需要进行线程的同步。 lock lock(锁)的工作原理： 所有的线程只有一把锁，谁抢到就执行哪个线程，当拥有锁的线程结束操作，需要释放锁让别的线程去竞争。 我们可以对上述的代码进行改写 import threading import time count =0 lock = threading.Lock() def addcount(): global count for i in range(1000): lock.acquire() count += 1 lock.release() for i in range(10): t = threading.Thread(target=addcount) t.start() time.sleep(3) print count 与原来的代码不同之处仅仅在于获取锁acquire()、释放锁release()的两个操作而已,Lock.acquire(blocking=True, timeout=-1),blocking参数表示是否阻塞当前线程等待，timeout表示阻塞时的等待时间 threading 中还有RLock与Lock功能类似，但是在同一个线程内可以多次调用RLCOK.acquire来获取锁，Lock如果多次accuire则会出现死锁的情况。 这样的RLock对比与Lock在哪些问题上有优势呢？(TO DO) GIL(Global Interpreter Lock) GIL 全局解释锁(TO DO) condtion condition 提供了比Lock更复杂的功能。线程只有满足一定条件后才能去竞争锁，不满足条件的线程处于等待池(当前线程调用condtion.wait()进入等待)，直到收到通知(其它线程调用condition.notify())才退出等待状态，再开始去竞争锁。 下面以消费者和生产者为例(始终保证生产足够,但不过剩(产品数量限制在30以内):) ) import threading import time condition = threading.Condition() class Producer(threading.Thread): def run(self): while(1): global x print 'producer accquire lock' condition.acquire() print 'producer get lock' if x > 30: print 'x>need_max producer wait，and release lock' condition.wait() else: x = x + 5 print 'producing',x condition.notify() print 'producer done, notify others' condition.release() #print 'producer release lock' time.sleep(1) #暂停线程，避免与consumer争lock class Consumer(threading.Thread): def run(self): while(1): global x print 'consumer accquire lock' condition.acquire() print 'Consumer get lock' if x == 0: print 'x==0 consumer wait，and release lock' condition.wait() else: x = x - 1 print 'consuming ',x time.sleep(1) condition.notify() print 'consumer done, notify others' condition.release() #print 'consumer release lock' time.sleep(1) x = 5 p = Producer() c = Consumer() p.start() c.start() 上述代码首先从Thread中继承得到两个子类。线程Thread类的继承只需要实现run方法即可。run方法在线程执行start启动线程被调用。Producer与Consumer的结构基本相同，只是判断进入wait或notify的条件不一样而已。当生产者当生产的量大于一定的阈值，调用wait进入等待状态，此时Consumer获得lock，消耗一个产品，并通知生产者继续生产。执行的结果如下所示，产品x的数量始终保持在30左右。 Consumer get lock consuming 22 consumer done, notify others producer get lock producing 27 producer done, notify others producer accquire lockconsumer accquire lock producer get lock producing 32 producer done, notify others Consumer get lock consuming 31 consumer done, notify others producer accquire lock producer get lock x>need_max producer wait，and release lock consumer accquire lock Consumer get lock consuming 30 consumer done, notify others consumer accquire lockproducer accquire lock Semaphore (TO DO) Event 事件对象是实现线程之间安全通通信的一种简单的方法。Event内部管理着一个内部标志，调用者可以用set()和clear()方法控制这个标志。其他线程可以使用wait()暂停直到设置这个标志，即线程调用wait()等待，调用的线程被阻塞，直到event被set或timeout，才继续执行。 以车辆等待红绿灯为例。 import threading import time class VehicleThread(threading.Thread): def __init__(self,threadName,event): threading.Thread.__init__(self,name=threadName) self.event = event def run(self): time.sleep(int(self.getName()[7:])) print self.getName(),'arrive,wait for green light' self.event.wait() print self.getName(),'passed' green_light_event = threading.Event() vehicleThreads = [] for i in xrange(10): vehicleThreads.append(VehicleThread('vehicle'+str(i),green_light_event)) for t in vehicleThreads: t.start() while threading.activeCount() > 2: #在python中除了主线程还有一个Thread-1, #IPython运行有5个线程？(Thread-5 Thread-1 MainThread Thread-4 IPythonHistorySavingThread) green_light_event.clear() time.sleep(3) print "green light appera!!! let's GO" green_light_event.set() time.sleep(1) # for t in threading.enumerate(): # print t.getName() thread与Queue的联合使用 (TO DO) 参考链接 Python Standard Library Python Standard Library-threading Python Moudle of the Week--threading Python 标准库学习 GIL Queue and Thread

python 迭代器&生成器 __iter__为python中的内建函数，如果想要定义的对象是iterable，你就需要定义__iter__ next(__next__)常与__iter__共同使用，构成迭代器(iterator) 带有yield关键字的函数为生成器(generator) 为了理解这几个关键字和迭代器、生成器。我们从一个常用的斐波那契数列(Fibinacci)开始 常规实现Fibinacci序列生成 def fib(max): n,a,b = 0,0,1 fib_list = [] while n < max : fib_list.append(b) a,b = b, a+b n = n +１ return fib_list

Pandas学习笔记 DataFrame 常用方法 Groupby的用法 Apply的用法 axis参数可用于指定按行处理还是按列处理 遍历DataFrame方法 DataFrame访问行列 # 根据row index name索引 df.loc[INDEX_NAME] # 根据column name 索引 df[COLUMN_NAME] DataFrame和Dict的相互转换 Dict to DataFrame: pandas.DataFrame.from_dict Series 常用方法 一些不常用的操作 # 对未命名column进行重命名 data.rename(columns={'Unnamed: 0':'new column name'}, inplace=True) 参考链接 Pandas 官方教程 Pandas 官方API Pandas教程 | 超好用的Groupby用法详解 Pandas|常见用法整理 10min快速入门

一分钟实现python安装包开发 1. 实现步骤： 创建一个setup.py文件。这个文件是Python安装包的核心，用于定义包的元数据、依赖关系和安装过程 使用setuptools、 wheel构建安装包 2. 举例：以实现函数计时功能为例 创建一个名为timer_decorator的Python包，它提供了一个计时器装饰器，用于测量函数的运行时间。 创建一个名为timer_decorator的目录。 在timer_decorator目录中创建一个名为__init__.py的文件，这将使其成为一个Python包。在__init__.py文件中，编写计时器装饰器的代码： import time def timer_decorator(func): def wrapper(*args, **kwargs): start_time = time.time() result = func(*args, **kwargs) end_time = time.time() print(f"{func.__name__} took {end_time - start_time:.2f} seconds to run.") return result return wrapper 创建一个setup.py文件，如下所示： from setuptools import setup setup( name='timer_decorator', version='0.1', py_modules=['timer_decorator'], install_requires=[ # 该项目没有依赖项 ], entry_points={ 'console_scripts': [ # 该项目没有可执行脚本 ], }, ) 基于setup.py创建whl安装文件 python setup.py timer_decorator 使用方法： from timer_decorator import timer_decorator @timer_decorator def my_function(): # Your code here