魔の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
from typing import List, Tuple

# 如果确信path是Windows路径则返回True，否则返回False
def must_be_windows_path(path):
    if len(path) >= 2 and path[1] == ':':
        return True
    if len(path) >= 2 and path[0] == '\\' and path[1] == '\\':
        return True
    return False

# 如果确信path是Linux路径则返回True，否则返回False
def must_be_linux_path(path):
    if len(path) >= 1 and path[0] == '/':
        return True
    return False


class PathNormalizer:
    def __init__(self, path_map_lst: List[Tuple[str, str]]):
        """
        Args:
        - path_map_lst List[Tuple[str, str]]: 通过列表指定Linux和Windows系统的路径前缀的匹配对
        """
        self.l2w = {}
        self.w2l = {}

        for e in path_map_lst:
            if must_be_linux_path(e[0]) or must_be_windows_path(e[1]):
                linux_path = e[0]
                windows_path = e[1]
            else:
                linux_path = e[1]
                windows_path = e[0]
            if linux_path[-1] != '/':
                linux_path += '/'
            if windows_path[-1] != '\\':
                windows_path += '\\'
            self.l2w[linux_path] = windows_path
            self.w2l[windows_path] = linux_path

    def __call__(self, path: str):
        """
        会自动转换传入的路径匹配到当前环境的操作系统。
        可以传入绝对路径或相对路径
        也可以传入已经符合当前操作系统规则的路径，此时则不会修改

        Args:
        - path str: 需要转换的路径

        Returns:
        - str: 匹配当前环境操作系统的路径
        """
        if sys.platform.lower() == 'linux':
            return self.to_linux_path(path)
        else:
            return self.to_windows_path(path)        

    def to_linux_path(self, path):
        if must_be_linux_path(path):
            return path

        for (windows_prefix, linux_prefix) in self.w2l.items():
            if path.startswith(windows_prefix):
                return linux_prefix + path[len(windows_prefix):].replace('\\', '/')
        return path.replace('\\', '/')

    def to_windows_path(self, path):
        if must_be_windows_path(path):
            return path

        for (linux_prefix, windows_prefix) in self.l2w.items():
            if path.startswith(linux_prefix):
                return windows_prefix + path[len(linux_prefix):].replace('/', '\\')
        return path.replace('/', '\\')


norm_path = PathNormalizer([
    (r"\\192.168.1.107\e", "/mnt/107e"),
    (r"\\192.168.1.99\database", "/mnt/99database"),
    (r"I:", "/mnt/112"),
    ])


def test():
    print(norm_path(r"I:\snapshot\abc"))


if __name__ == '__main__':
    test()