Demo-1

• 有如下值集合 [11,22,33,44,55,66,77,88,99,90]，将所有大于 66 的值保存至字典的第一个key中，将小于 66 的值保存至第二个key的值中。
• 即： {'k1': 大于66的所有值, 'k2': 小于66的所有值}

L=[11,22,33,44,55,66,77,88,99,90]
{"k1":list(filter(lambda x:x>66,L)),"k2":list(filter(lambda x:x<66,L))}

Demo-2

• 查找列表中元素，移除每个元素的空格，并查找以 a或A开头 并且以 c 结尾的所有元素。

li = ["alec", " aric", "Alex", "Tony", "rain"]
tu = ("alec", "aric", "Alex", "Tony", "rain")
dic = {'k1': "alex", 'k2': ' aric', "k3": "Alex", "k4": "Tony"}

from typing import Iterable,List,Tuple,Dict
import logging
li = ["alec", " aric", "Alex", "Tony", "rain"]
tu = ("alec", " aric", "Alex", "Tony", "rain")
dic = {'k1': "alex", 'k2': ' aric', "k3": "Alex", "k4": "Tony"}

def remove_strip(x:Iterable):
    '''
    input different data structure
    '''
    if isinstance(x,List):
        return list(map(str.strip,li))
    elif isinstance(x,Tuple):
        return tuple(map(str.strip,li))
    elif isinstance(x,Dict):
        return dict(map(str.strip,li))
    else:
        logging.warning("notice input data structure")


def p(x:Iterable):
    return list(filter(lambda item:item.endswith('c') and item.capitalize().startswith('A'),x))

def main(x:Iterable):
    if isinstance(x,Dict):
        return p(x.values())
    else:
        return p(x)
if __name__=="__main__":
    print(main(li))
    print(remove_strip(li))

Demo-3

• 输出商品列表，用户输入序号，显示用户选中的商品

li = ["手机", "电脑", '鼠标垫', '游艇']

li = ["手机", "电脑", '鼠标垫', '游艇']
for index, text in enumerate(li,1):
    print(index, text)

inp = input("请选择要购买的商品：")
inp_num = int(inp)
print(li[inp_num-1])

Demo-10

• 写函数，检查用户传入的对象（字符串、列表、元组）的每一个元素是否含有空内容。

f=lambda i:True if i == None or i.isspace() == True or i == '' else False
if any(map(f,li))==True:
    print("有空内容")

Demo-9

• 写函数，判断用户传入的对象（字符串、列表、元组）长度是否大于5

def is_length(obj):
    obj_len = 0
    for i in obj:
        obj_len += 1
    if obj_len > 5:
        print(obj, "长度大于5")
    else:
        print(obj, "长度为", obj_len)

Demo-22

• python日志模块

import logging

logging.basicConfig(
    #filename="base.log",
    format="[%(lineno)d]%(asctime)s - %(pathname)s - %(levelname)-8s - %(module)s : %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    level=logging.DEBUG
)

logging.debug('debug')
logging.info('info')
logging.warning('warning')
logging.error('error')
logging.critical('critical')
logging.log(10, 'log')

Demo-19

• 实现命令行进度条，带百分比

import sys
import time

for i in range(101):
    # 清除屏幕
    sys.stdout.write("\r")

    # 显示百分比和进度条
    sys.stdout.write("[%s%%|%-100s]" % (i, i * ('█')))

    # 从缓存刷入到屏幕
    sys.stdout.flush()

    # 延时0.3秒
    time.sleep(0.3)

Demo-16

• 实现多层装饰器及任意传参

import time
def decorator_2(func):
    """
    最外层装饰器
    :param func:
    :return:
    """
    def warp(*args, **kwargs):
        print("最外层装饰器1".center(20, "="))
        ret = func(*args, **kwargs)
        print("最外层装饰器2".center(20, "="))
        return ret

    return warp
@decorator_2
def s(x:int)->int:
    '''
    :param x : number
    :return :
    '''
    return x*2

print(s(5))

#Demo-14
import random

code = ""
for i in range(4):

    num = random.randrange(0, 2)

    # 当前位显示字母或者数字各有50%几率
    if num == 0:
        # 随机显示数字
        r1 = random.randrange(0, 10)
        code += str(r1)
    else:
        # 随机显示字母
        i = random.randrange(65, 91)
        c = chr(i)
        code += c

print(code)

Demo-24 并行运行

并行实例

• import joblib
• Parallel(n_job=4,backend="multiprocess")(delayed(function)(*args) for i,j in iteratable)

Parallel(n_jobs=4,backend="threading")(delayed(f_compute) (j) for j in range(5))
Parallel(n_jobs=30)(delayed(mapping)(sample, fqs, ref, 2)
                        for sample, fqs in data.items())

import psutil
from math import sqrt
%timeit [sqrt(i ** 2) for i in range(1000)]

#
print("CPU number :{}".format(psutil.cpu_count()))
from joblib import Parallel, delayed
%timeit Parallel(n_jobs=2)(delayed(sqrt)(i ** 2) for i in range(1000))

计算时间比较

• 计算密集型任务适合多进程，IO 密集型任务适合多线程
• 计算密集型任务中速度：多进程 > 单进程/线程 > 多线程， IO 密集型任务速度： 多线程 > 多进程 > 单进程/线程。

def f_IO(a):  # IO 密集型
    time.sleep(5)

def f_compute(a):  # 计算密集型
    for _ in range(int(1e7)):
        math.sin(40) + math.cos(40)
    return


def joblib_process(sub_f):
    with parallel_backend("multiprocessing", n_jobs=6):
        res = Parallel()(delayed(sub_f)(j) for j in range(6))
    return


def joblib_thread(sub_f):
    with parallel_backend('threading', n_jobs=6):
        res = Parallel()(delayed(sub_f)(j) for j in range(6))
    return

def showtime(f, sub_f, name):
    start_time = time.time()
    f(sub_f)
    print("{} time: {:.4f}s".format(name, time.time() - start_time))
   
sub_f = f_compute
showtime(joblib_process, sub_f, "joblib multiprocess")
showtime(joblib_thread, sub_f, "joblib thread")