魔のkyo的BLOG

# -*- coding: utf-8 -*-
import multiprocessing
import queue
import threading


def call_helper(func, common_data, respective_data):
    args = []
    kw = {}
    if common_data:
        assert(isinstance(common_data, list) or isinstance(common_data, tuple) or isinstance(common_data, dict))
        if isinstance(common_data, list) or isinstance(common_data, tuple):
            args += common_data
        elif isinstance(common_data, dict):
            kw.update(common_data)
    if respective_data:
        assert(isinstance(respective_data, list) or isinstance(respective_data, tuple) or isinstance(respective_data, dict))
        if isinstance(respective_data, list) or isinstance(respective_data, tuple):
            args += respective_data
        else:
            kw.update(respective_data)
    return func(*args, **kw)


def _func(func, common_data, queue_in, queue_out):
    while True:
        try:
            (i, respective_data) = queue_in.get(False)
            r = call_helper(func, common_data, respective_data)
            if r is not None:
                queue_out.put((i, r))
        except queue.Empty as e:
            break

"""
用多进程执行func函数
Args:
    process_num: 并发的进程数量
    func: 回调函数，任务的函数体，需要接收通过common_data和respective_data_list传递的参数
    common_data: 所有func共用的数据，应为list/tuple/dict类型之一或None，为dict类型时作为具名参数传递
    respective_data_list: 每个func各自的数据，应为list/tuple/dict类型的list，为dict类型的list时作为具名参数传递

Returns:
    list: func返回值的列表，长度等于len(respective_data_list)，顺序和respective_data_list中的参数顺序保持一致
"""
def multi_process(process_num, func, common_data, respective_data_list):
    manager = multiprocessing.Manager()
    queue_in = manager.Queue()
    for i, e in enumerate(respective_data_list):
        queue_in.put((i,e))
    queue_out = manager.Queue()
    processes = []
    for i in range(0, process_num):
        proc = multiprocessing.Process(target=_func, args=(func, common_data, queue_in, queue_out))
        processes.append(proc)
    for proc in processes:
        proc.start()
    for proc in processes:
        proc.join()
    if not queue_out.empty():
        ret = [None] * len(respective_data_list)
        while True:
            try:
                (i, ret_value) = queue_out.get(False)
                ret[i] = ret_value
            except queue.Empty as e:
                break
        return ret

"""
用多线程执行func函数
Args:
    thread_num: 并发的线程数量
    func: 回调函数，任务的函数体，需要接收通过common_data和respective_data_list传递的参数
    common_data: 所有func共用的数据，应为list/tuple/dict类型之一或None，为dict类型时作为具名参数传递
    respective_data_list: 每个func各自的数据，应为list/tuple/dict类型的list，为dict类型的list时作为具名参数传递

Returns:
    list: func返回值的列表，长度等于len(respective_data_list)，顺序和respective_data_list中的参数顺序保持一致
"""
def multi_thread(thread_num, func, common_data, respective_data_list):
    queue_in = queue.Queue()
    for i, e in enumerate(respective_data_list):
        queue_in.put((i,e))
    queue_out = queue.Queue()
    threads = []
    for i in range(0, thread_num):
        thread = threading.Thread(target=_func, args=(func, common_data, queue_in, queue_out))
        threads.append(thread)
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    if not queue_out.empty():
        ret = [None] * len(respective_data_list)
        while True:
            try:
                (i, ret_value) = queue_out.get(False)
                ret[i] = ret_value
            except queue.Empty as e:
                break
        return ret

 
"""
用单线程执行func函数
Args:
    _dummy_num: 无用的参数，仅为了和其他同类函数统一调用格式
    func: 回调函数，任务的函数体，需要接收通过common_data和respective_data_list传递的参数
    common_data: 所有func共用的数据，应为list/tuple/dict类型之一或None，为dict类型时作为具名参数传递
    respective_data_list: 每个func各自的数据，应为list/tuple/dict类型的list，为dict类型的list时作为具名参数传递

Returns:
    list: func返回值的列表，长度等于len(respective_data_list)，顺序和respective_data_list中的参数顺序保持一致
"""
def single_thread(_dummy_num, func, common_data, respective_data_list):
    ret = []
    for i, respective_data in enumerate(respective_data_list):
        r = call_helper(func, common_data, respective_data)
        if r is not None:
            if not ret:
                ret = [None] * len(respective_data_list)
            ret[i] = r
    return ret

from concurrent import futures

import grpc
import sys
sys.path.append("../protos_gen")
from calc_pb2 import AddReply # protoc生成的代码
from calc_pb2_grpc import CalcServicer, add_CalcServicer_to_server # protoc生成的代码


class Calc(CalcServicer):
    def Add(self, request, context):
        return AddReply(sum=request.a + request.b)


def main():
    port = '50051'
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    add_CalcServicer_to_server(Calc(), server)
    server.add_insecure_port('0.0.0.0:' + port)
    server.start()
    print("Server started, listening on " + port)
    server.wait_for_termination()


if __name__ == '__main__':
    main()

import grpc
import sys
sys.path.append("../protos_gen")
from calc_pb2 import AddRequest   # protoc生成的代码
from calc_pb2_grpc import CalcStub # protoc生成的代码


class Calc:
    def __init__(self, server):
        self.channel = grpc.insecure_channel(server)
        self.stub = CalcStub(self.channel)

    def __del__(self):
        del self.stub
        self.channel.close()

    def Add(self, **kw):
        response = self.stub.Add(AddRequest(**kw))
        return response.sum


def main():
    client = Calc(server='localhost:50051')
    s = client.Add(a=1, b=2)
    print(f"1+2={s}")


if __name__ == '__main__':
    main()

import sys
import os
import re

# 将传入的路径根据当前平台做规范化，以便让路径在Windows和Linux上兼容
# 对于共享或挂载路径，需要共享已经按命名规则挂在到/mnt目录下（在Linux上） 或 已经通过net use建立了连接（在Windows上）
# Windows和Linux上共享或挂载点对应规则：
#   \\192.168.1.107\e <=> /mnt/107e
#   X: <=> /mnt/X
#   \\192.168.1.99\database <=> /mnt/99database
# 用法：
#   将 open(r"\\192.168.1.107\e\NAV\abc.txt")
#   改成 open(to_os_path(r"\\192.168.1.107\e\NAV\abc.txt"))
#   即可做到平台兼容
def to_os_path(path):
    if sys.platform == 'linux':
        return to_linux_path(path)
    else:
        return to_windows_path(path)


def to_linux_path(path):
    m = re.match(r"\\\\\d+\.\d+\.\d+\.(\d+)\\([^\\])", path)
    if m: # 形似一个Windows网络共享地址
        return '/mnt/' + m.group(1) + m.group(2) + path[m.span()[1] : ].replace('\\', '/')
    m = re.match(r"([a-zA-Z]):\\", path)
    if m: # 形似一个Windows本地绝对路径
        return '/mnt/' + m.group(1) + '/' + path[m.span()[1] : ].replace('\\', '/')

    # 已经是一个Linux的据对路径，或者是Windows或Linux的相对路径
    return path.replace('\\', '/')


def to_windows_path(path):
    m = re.match(r"/mnt/(\d+)([^\\])", path)
    if m:
        return r"\\192.168.1.{}\{}{}".format(m.group(1), m.group(2), path[m.span()[1] : ].replace('/', '\\'))
    m = re.match(r"/mnt/([a-zA-Z])\b", path)
    if m:
        return m.group(1) + ":" + path[m.span()[1] : ].replace('/', '\\')
    return path.replace('/', '\\')