使用Python Multiprocessing庫處理3D數據的方法

這篇文章主要介紹“使用Python Multiprocessing庫處理3D數據的方法”，在日常操作中，相信很多人在使用Python Multiprocessing庫處理3D數據的方法問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”使用Python Multiprocessing庫處理3D數據的方法”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

我們的數據由.obj存儲在.7z存檔中的文件組成，這在存儲效率方面非常出色。但是當我們需要訪問它的確切部分時，我們應該努力。在這里，我定義了包裝 7-zip 存檔并提供底層數據接口的類。

from io import BytesIO
import py7zlib

class MeshesArchive(object):
    def __init__(self, archive_path):
        fp = open(archive_path, 'rb')
        self.archive = py7zlib.Archive7z(fp)
        self.archive_path = archive_path
        self.names_list = self.archive.getnames()
        self.cur_id = 0
        
    def __len__(self):
        return len(self.names_list)
    
    def get(self, name):
        bytes_io = BytesIO(self.archive.getmember(name).read())
        return bytes_io

    def __getitem__(self, idx):
        return self.get(self.names[idx])
      
    def __iter__(self):
        return self

    def __next__(self):
      if self.cur_id >= len(self.names_list):
          raise StopIteration
      name = self.names_list[self.cur_id]
      self.cur_id += 1
      return self.get(name)

這個類幾乎不依賴py7zlib包，它允許我們在每次調用get方法時解壓縮數據，并為我們提供存檔中的文件數。我們還定義了__iter__這將幫助我們map像在可迭代對象上一樣在該對象上啟動多處理。 

您可能知道，可以創建一個 Python 類，從中可以實例化可迭代對象。該類應滿足以下條件：覆蓋__getitem__返回self和__next__返回后續元素。我們絕對遵循這個規則。 

上面的定義為我們提供了遍歷存檔的可能性，但 它是否允許我們 對內容進行并行隨機訪問？這是一個有趣的問題，我在網上沒有找到答案，但我們可以研究源代碼py7zlib并嘗試自己回答。

在這里，我提供了來自pylzma的代碼片段：

class Archive7z(Base):
  def __init__(self, file, password=None):
    # ...
    self.files = {}
    # ...
    for info in files.files:
      # create an instance of ArchiveFile that knows location on disk
      file = ArchiveFile(info, pos, src_pos, folder, self, maxsize=maxsize)
      # ...
      self.files.append(file)
    # ...
    self.files_map.update([(x.filename, x) for x in self.files])
        
  # method that returns an ArchiveFile from files_map dictionary
  def getmember(self, name):
      if isinstance(name, (int, long)):
          try:
              return self.files[name]
          except IndexError:
              return None

      return self.files_map.get(name, None)
    
    
class Archive7z(Base):
  def read(self):
    # ...
    for level, coder in enumerate(self._folder.coders):
      # ...
      # get the decoder and decode the underlying data
      data = getattr(self, decoder)(coder, data, level, num_coders)

    return data

在代碼中，您可以看到在從存檔中讀取下一個對象期間調用的方法。我相信從上面可以清楚地看出，只要同時多次讀取存檔，就沒有理由阻止存檔。

接下來，我們快速介紹一下什么是網格和點云。 

首先，網格是頂點、邊和面的集合。頂點由空間中的(x,y,z) 坐標定義并分配有唯一編號。邊和面是相應的點對和三元組的組，并用提到的唯一點 id 定義。通常，當我們談論“網格”時，我們指的是“三角形網格”，即由三角形組成的表面。使用trimesh庫在 Python 中使用網格要容易得多。例如，它提供了.obj在內存中加載文件的接口。要在jupyter notebook一個3D 對象中顯示和交互，可以使用k3d庫。

所以，用下面的代碼片段我回答這個問題：“你怎么繪制trimesh的對象jupyter有k3d？”

import trimesh
import k3d

with open("./data/meshes/stanford-bunny.obj") as f:
    bunny_mesh = trimesh.load(f, 'obj')

plot = k3d.plot()
mesh = k3d.mesh(bunny_mesh.vertices, bunny_mesh.faces)
plot += mesh
plot.display()

k3d 顯示的斯坦福兔子網格 
其次，點云是表示空間中對象的 3D 點數組。許多 3D 掃描儀生成點云作為掃描對象的表示。出于演示目的，我們可以讀取相同的網格并將其頂點顯示為點云。

import trimesh
import k3d

with open("./data/meshes/stanford-bunny.obj") as f:
    bunny_mesh = trimesh.load(f, 'obj')
    
plot = k3d.plot()
cloud = k3d.points(bunny_mesh.vertices, point_size=0.0001, shader="flat")
plot += cloud
plot.display()

k3d繪制的
點云
如上所述，3D 掃描儀為我們提供了一個點云。假設我們有一個網格數據庫，我們想在我們的數據庫中找到一個與掃描對象對齊的網格，也就是點云。
為了解決這個問題，我們可以提出一種簡單的方法。我們將從我們的檔案中搜索給定點云的點與每個網格之間的最大距離。
如果1e-4某些網格的距離更小，我們將認為該網格與點云對齊。

最后，我們來到了多處理部分。請記住，我們的存檔中有大量文件可能無法放在一起放在內存中，因為我們更喜歡并行處理它們。

為了實現這一點，我們將使用 multiprocessing Pool，它使用map或imap/imap_unordered方法處理用戶定義函數的多次調用。

map和imap影響我們的區別在于，map在將其發送到工作進程之前將其轉換為列表。如果存檔太大而無法寫入 RAM，則不應將其解壓縮到 Python 列表中。換句話說，兩者的執行速度是相似的。

[加載網格：pool.map w/o manager] 4 個進程池耗時：37.213207403818764 秒
[加載網格：pool.imap_unordered w/o manager] 4 個進程池耗時：37.219303369522095 秒

在上面，您可以看到從適合內存的網格檔案中簡單讀取的結果。

更進一步imap：讓我們討論如何實現找到靠近點云的網格的目標。這是數據。我們有來自斯坦福模型的 5 個不同的網格。我們將通過向斯坦福兔子網格的頂點添加噪聲來模擬 3D 掃描。

import numpy as np
from numpy.random import default_rng

def normalize_pc(points):
    points = points - points.mean(axis=0)[None, :]
    dists = np.linalg.norm(points, axis=1)
    scaled_points = points / dists.max()
    return scaled_points


def load_bunny_pc(bunny_path):
    STD = 1e-3 
    with open(bunny_path) as f:
        bunny_mesh = load_mesh(f)
    # normalize point cloud 
    scaled_bunny = normalize_pc(bunny_mesh.vertices)
    # add some noise to point cloud
    rng = default_rng()
    noise = rng.normal(0.0, STD, scaled_bunny.shape)
    distorted_bunny = scaled_bunny + noise
    return distorted_bunny

當然，我們之前在下面將點云和網格頂點歸一化，以在 3D 立方體中縮放它們。

要計算點云和網格之間的距離，我們將使用igl。為了完成，我們需要編寫一個函數來調用每個進程及其依賴項。讓我們用以下代碼段來總結。

import itertools
import time
import numpy as np
from numpy.random import default_rng
import trimesh
import igl
from tqdm import tqdm
from multiprocessing import Pool
def load_mesh(obj_file):
    mesh = trimesh.load(obj_file, 'obj')
    return mesh
def get_max_dist(base_mesh, point_cloud):
    distance_sq, mesh_face_indexes, _ = igl.point_mesh_squared_distance(
        point_cloud,
        base_mesh.vertices,
        base_mesh.faces
    )
    return distance_sq.max()
def load_mesh_get_distance(args):
    obj_file, point_cloud = args[0], args[1]
    mesh = load_mesh(obj_file)
    mesh.vertices = normalize_pc(mesh.vertices)
    max_dist = get_max_dist(mesh, point_cloud)
    return max_dist
def read_meshes_get_distances_pool_imap(archive_path, point_cloud, num_proc, num_iterations):
    # do the meshes processing within a pool
    elapsed_time = []
    for _ in range(num_iterations):
        archive = MeshesArchive(archive_path)
        pool = Pool(num_proc)
        start = time.time()
        result = list(tqdm(pool.imap(
            load_mesh_get_distance,
            zip(archive, itertools.repeat(point_cloud)),
        ), total=len(archive)))
        pool.close()
        pool.join()
        end = time.time()
        elapsed_time.append(end - start)
    print(f'[Process meshes: pool.imap] Pool of {num_proc} processes elapsed time: {np.array(elapsed_time).mean()} sec')   
    for name, dist in zip(archive.names_list, result):
        print(f"{name} {dist}")    
    return result  
 if __name__ == "__main__":
    bunny_path = "./data/meshes/stanford-bunny.obj"
    archive_path = "./data/meshes.7z"
    num_proc = 4
    num_iterations = 3
    point_cloud = load_bunny_pc(bunny_path)
    read_meshes_get_distances_pool_no_manager_imap(archive_path, point_cloud, num_proc, num_iterations)

這read_meshes_get_distances_pool_imap是一個中心函數，其中完成以下操作：

• MeshesArchive并multiprocessing.Pool初始化

• tqdm 用于觀察池進度，并手動完成整個池的分析

• 執行結果的輸出

請注意我們如何傳遞參數以imap從archive和point_cloud使用zip(archive, itertools.repeat(point_cloud)). 這允許我們將點云數組粘貼到存檔的每個條目上，避免轉換archive為列表。

執行結果如下：

100%|########################################### #####################| 5/5 [00:00<00:00, 5.14it/s] 
100%|########################### ####################################| 5/5 [00:00<00:00, 5.08it/s] 
100%|########################### ####################################| 5/5 [00:00 <0時，5.18it /秒] 
[方法網眼：pool.imap W / O管理器] 4個過程的池經過時間：1.0080536206563313秒
armadillo.obj 0.16176825266293382 
beast.obj 0.28608649819198073 
cow.obj 0.41653845909820164
現貨.obj 0.22739556571296735 
stanford-bunny.obj 2.3699851136074263e-05

我們可以注意到斯坦福兔子是最接近給定點云的網格。還可以看出，我們沒有使用大量數據，但我們已經證明，即使我們在存檔中有大量網格，該解決方案也能奏效。

到此，關于“使用Python Multiprocessing庫處理3D數據的方法”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！