Vícevláknové zpracování v Pythonu pro ArcGIS - Příklad integrace do geoprocesních nástrojů

ShinjiKATAYA1

Úvod

V minulých příspěvcích na blogu jsme představili „Vícevláknové zpracování v Pythonu pro ArcGIS“ a „Vícevláknové zpracování v Pythonu pro ArcGIS - praktické příklady“, kde jsme ukázali vícevláknové zpracování pomocí ArcPy a Pythonu, představili nejlepší praktiky a vzory kódu s příklady.Tentokrát přidáme k těmto tématům také aktualizaci pro kódovací vzory v Pythonu 3.3 a vyšším a zároveň přidáme příklad integrace do geoprocesních nástrojů ArcGIS Pro.

Poznámka: Kód vysvětlený v tomto článku je již publikován jako „Farmland polygon conversion tools“ a je okamžitě použitelný jako Python toolbox. Doporučujeme se na něj podívat.

 

Kódovací vzory

V předchozím článku jsme uvedli tři hlavní body týkající se kódovacích vzorů.Jsou to následující tři:

1) Funkce pracovní dávky: batch_convert

2) Wrapper funkce pro pracovní dávku: multi_run_batch_convert ※ nyní není potřeba

3) Zpracování vícevláknového zpracování: exec_batch_convert

Nicméně od verze Python 3.3 byla přidána metoda starmap třídy Pool modulu multiprocessing, která umožňuje volání s více argumenty. Proto při použití v prostředí ArcGIS Pro již není potřeba wrapper funkce a kód lze zjednodušit.

Nyní tedy vysvětlíme hlavní body zpracování založené na příkladu.

 

Ukázkový kód (konverze zemědělských polygonů ve formátu GeoJSON)

 Data určená ke konverzi jsou stejná jako v předchozím článku, konkrétně data „Zemědělské hranice (polygonové GeoJSON)“ zveřejněná Ministerstvem zemědělství, lesnictví a rybolovu.Jako příklad používáme tato data.

※ Vzhledem k tomu, že struktura polygonů se od předchozího článku liší, ukázkový soubor použitý v tomto blogu je založen na nejnovějších datech polygonů (ve formátu GeoJSON).Ukázkový soubor tentokrát (MP_Farmland_JsonToFeatureClass.py) zpracovává jeden "prefekturní adresář" po rozbalení ZIP souboru polygonů.

Po rozbalení obsahuje "prefekturní adresář" několik souborů GeoJSON jednotlivých měst, okresů a vesnic.

Vstupní adresář: Prefekturní adresář obsahující soubory GeoJSON jednotlivých měst, okresů a vesnic

Příklad)
    |-2024_02
        |-2024_022012.json # GeoJSON jednotlivých měst, okresů a vesnic
        |-2024_022021.json
        |-2024_022039.json
        .....

V ukázkovém programu tentokrát zadáváme tento prefekturní adresář jako vstupní adresář a provádíme následující zpracování.

1. Vytvoření geodatabáze (FGDB) databáze prostorových dat pro každý soubor města, okresu a vesnice a převod polygonů zemědělské půdy do feature class (toto zpracování se provádí ve vícevláknovém režimu)
2. Vytvoření geodatabáze (FGDB) prefekturního adresáře a sloučení feature class jednotlivých měst, okresů a vesnic do jedné feature class prefekturní geodatabáze
3. Odstranění feature class jednotlivých měst, okresů a vesnic
Výsledkem je, že v prefekturní geodatabázi FGDB zůstane pouze sloučená feature class Farmland (viz obrázek níže).

Výstupní adresář: Cíl vytvořené sloučené prefekturní geodatabáze FGDB

Příklad)
    |-2024_02_filegdb
        |-2024_022012.gdb # FGDB jednotlivých měst, okresů a vesnic bude po sloučení odstraněna z prefekturního adresáře
            |   c_2024_022012
        |-2024_022021.gdb
            |   c_2024_022021
            .....
        |-2024_02_filegdb.gdb # Prefekturní geodatabáze obsahující sloučené feature class jednotlivých měst, okresů a vesnic po sloučení
            |   Farmland # Název feature class (fixní uvnitř kódu)

 

Nyní vysvětlíme implementační body kódovacích vzorů souvisejících s vícevláknovým zpracováním.

1) Funkce pracovní dávky : batch_convert

Jako kódovací vzor popisujeme nejprve funkci pracovní dávky podobnou běžnému zpracování v ArcPy. Důležité je zde, že aby bylo možné načíst data jako nejlepší praxi, a aby toto zpracování odpovídalo zpracování v jednom procesu, je nutné zajistit přístup k výstupnímu umístění buď přes vícenásobné procesy (více funkcí pracovních dávek), nebo použít Enterprise geodatabázi nebo zajistit, aby se nepoužívaly stejné soubory geodatabází.

※ Tentokrát proto nepoužíváme stejnou výstupní geodatabázi jako u předchozího příkladu, ale vytváříme novou geodatabázi souborů měst, okresů a vesnic uvnitř funkce pracovní dávky a provádíme konverzi uvnitř této geodatabáze.

def batch_convert(in_jsonfile :str, outws :str) -> str:
'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"'"',' field 62 ':' Zpracování prováděné v jednom procesu:',' field 63 ':' 1) Zápis do FGDB nelze provést ve více procesech současně kvůli omezením',' field 64 ':'    Převod GeoJSON souboru jednoho města/okresu/vesnice do feature class ve vlastní třídě FarmlandGeojsonToFeaturesEx',' field 65 ':' ',' field 66 ':' in_jsonfile : cesta k GeoJSON souboru města/okresu/vesnice',' field 67 ':' outws       : výstupní gdb',' field 68 ':' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ',' field 69 ':' if not arcpy.Exists(outws):',' field 70 ':' outfolder = u"{0}".format(os.path.dirname(outws))',' field 71 ':' foldername= u"{0}".format(os.path.basename(outws))',' field 72 ':' arcpy.management.CreateFileGDB(outfolder, foldername, "CURRENT")',' field 73 ':' filename = os.path.basename(in_jsonfile)',' field 74 ':' in_json_file = os.path.splitext(filename)[0]',' field 75 ':' newfc = u"c_{0}".format(in_json_file) # Název souboru začínající číslem nelze změnit přímo, proto přidáváme prefix c_.',' field 76 ':' # Převod pomocí vlastní třídy FarmlandGeojsonToFeaturesEx',' field 77 ':' #arcpy.conversion.JSONToFeatures(in_jsonfile, os.path.join(outws, newfc))',' field 78 ':' convGeojson = FarmlandGeojsonToFeaturesEx()',' field 79 ':' convGeojson.geojson_to_features(in_jsonfile, os.path.join(outws, newfc))',' field 80 ':' del convGeojson',' field 81 ':' return u"    Konverze dokončena: {0}".format(outws)',' field 82 ':' ',' field 83 ':'2) Wrapper funkce pro pracovní dávku',' field 84 ':'Jak bylo uvedeno výše, není nyní potřeba ji definovat pro použití v aktuálním prostředí ArcGIS Pro.',' field 85 ':' ',' field 86 ':'3) Zpracování vícevláknového zpracování : exec_batch_convert',' field 87 ':' Předáváme cestu k pythonw.exe pro každý proces pomocí multiprocessing.set_executable() a předáváme dva argumenty (in_jsonfile, outws) jako seznam params do funkce pracovní dávky.',' field 88 ':' Poté ',' field 89 ':' pool = multiprocessing.Pool(cpu_cnt) zajistí vytvoření poolu procesů ',' field 90 ':' results = pool.starmap(batch_convert, params) spustí funkci batch_convert s parametry params.',' field 91 ':' Po dokončení všech funkcí pracovních dávek ',' field 92 ':' pool.close()',' field 93 ':' pool.join()',' field 94 ':' zavoláme.',' field 95 ':'def exec_batch_convert(infolder :str, outfolder :str, cpu_cnt :int):',' field 96 ':'    '''',' field 97 ':'    Vícevláknové zpracování:',' field 98 ':'    '''',' field 99 ':'    try:',' field 100 ':'        start = datetime.datetime.now()',' field 101 ':'        arcpy.AddMessage(u"-- Start: MP_Farmland_JsonToFeatureClass --:{0}".format(start))',' field 102 ':'        ',' field 103 ':'        #a) Nastavení pythonw.exe pro každý proces',' field 104 ':'        python_path = sys.exec_prefix',' field 105 ':'        multiprocessing.set_executable(os.path.join(python_path,"pythonw.exe"))',' field 106 ':'        #multiprocessing.set_executable(os.path.join(python_path,"python.exe")) # Používáme pythonw.exe protože CMD okno by se spustilo při python.exe',' field 107 ':'        ',' field 108 ':'        #b) Serializace parametrů předaných každému procesu pracovního dávky',' field 109 ':'        arcpy.AddMessage(u"  Převod každého GeoJSON souboru : multiprocessing")',' field 110 ':'        arcpy.env.workspace = infolder',' field 111 ':'        infiles = arcpy.ListFiles("*.json")',' field 112 ':'        params=[]',' field 113 ':'        for infile in infiles:',' field 114 ':'            param1 = os.path.join(infolder,infile) # GeoJSON soubor města/okresu/vesnice',' field 115 ':'            filename = os.path.basename(infile)',' field 116 ':'            gdbname = u"{0}.gdb".format(os.path.splitext(filename)[0])',' field 117 ':'            param2 = os.path.join(outfolder,gdbname) # Výstupní FGDB soubor města/okresu/vesnice',' field 118 ':'            params.append((param1, param2))',' field 119 ':'        if len(infiles) < cpu_cnt: # Pokud je méně souborů než CPU jader není třeba spouštět nadbytečné procesy',' field 120 ':'            cpu_cnt = len(infiles)',' field 121 ':'        pool = multiprocessing.Pool(cpu_cnt) # Vytvoření počtu procesů podle počtu CPU jader',' field 122 ':'        results = pool.starmap(batch_convert, params) # Spustit paralelně podle parametrů (podpora starmap od Python3.3)',' field 123 ':'        pool.close()',' field 124 ':'        pool.join()',' field 125':'        ',' field126':'        # Výpis výsledků zpracování každého procesu pracovního dávky', 'field127':'        for r in results:', 'field128':'            arcpy.AddMessage(u"{0}".format(r))', 'field129':'        ', 'field130':'        #c) Vytvoření sloučené FGDB prefektury ze sloučených feature class jednotlivých měst/okresů/vesnic (
    1プロセスで実行する処理:
      1) FGDBへの書込みは仕様で複数プロセスで書込みできないため
           1市区町村のGeoJSONファイルを 独自のFarmlandGeojsonToFeaturesEx クラスで1市区町村.gdb内のフィーチャクラスに変換
    
    in_jsonfile : 市区町村のGeoJSONファイルへのパス
    outws       : 出力するgdb
    '''
    if not arcpy.Exists(outws):
        outfolder = u"{0}".format(os.path.dirname(outws))
        foldername= u"{0}".format(os.path.basename(outws))
        arcpy.management.CreateFileGDB(outfolder, foldername, "CURRENT")
    filename = os.path.basename(in_jsonfile)
    in_json_file = os.path.splitext(filename)[0]
    newfc = u"c_{0}".format(in_json_file) #数値で始まるファイル名はそのまま変換できないので接頭にc_を入れる
    #独自のFarmlandGeojsonToFeaturesEx クラスで変換
    #arcpy.conversion.JSONToFeatures(in_jsonfile, os.path.join(outws, newfc))
    convGeojson = FarmlandGeojsonToFeaturesEx()
    convGeojson.geojson_to_features(in_jsonfile, os.path.join(outws, newfc))
    del convGeojson
    return u"    変換済：{0}".format(outws)

 

2)ワーカー関数のラッパー

上述のように、現在のArcGIS Pro の環境で利用する際は不要のため、定義していません。

 

3)マルチプロセスの処理 : exec_batch_convert

各プロセス用のpythonw.exe へのパスを、multiprocessing.set_executable() で設定し、ワーカー関数へ渡す2つの引数（in_jsonfile, outws）を params にリスト化して入れます。その後、pool = multiprocessing.Pool(cpu_cnt) で確保した pool にresults = pool.starmap(batch_convert, params) で関数名 (batch_convert) と params を指定します。また、各ワーカー関数での処理が終了したら、

pool.close()

pool.join()

を呼び出します。

def exec_batch_convert(infolder :str, outfolder :str, cpu_cnt :int):
    '''
    マルチプロセスでの処理：
    '''
    try:
        start = datetime.datetime.now()
        arcpy.AddMessage(u"-- Strat: MP_Farmland_JsonToFeatureClass --:{0}".format(start))
        
        #a) 各プロセス用の pythonw.exe を設定
        python_path = sys.exec_prefix
        multiprocessing.set_executable(os.path.join(python_path,'pythonw.exe'))
        #multiprocessing.set_executable(os.path.join(python_path,'python.exe')) #CMDプロンプトの画面が起動するので'pythonw.exe'を使う
        
        #b) 各プロセスに渡すパラメータをリスト化
        arcpy.AddMessage(u"  Convert each GeoJSON files : multiprocessing")
        arcpy.env.workspace = infolder
        infiles = arcpy.ListFiles("*.json")
        params=[]
        for infile in infiles:
            param1 = os.path.join(infolder,infile) #市区町村のGeoJSONファイル
            filename = os.path.basename(infile)
            gdbname = u"{0}.gdb".format(os.path.splitext(filename)[0])
            param2 = os.path.join(outfolder,gdbname) # 出力する市区町村ファイルジオデータベース
            params.append((param1, param2))
        if len(infiles) < cpu_cnt: # 処理ファイル数がCPUコアより少ない場合無駄なプロセスを起動不要
            cpu_cnt = len(infiles)
        pool = multiprocessing.Pool(cpu_cnt) # cpu数分プロセス作成
        results = pool.starmap(batch_convert, params) # 割り当てプロセスで順次実行される（Python3.3で追加されたstarmapは複数の引数に対応）
        pool.close()
        pool.join()
        
        # 各プロセスでの処理結果を出力
        for r in results:
            arcpy.AddMessage(u"{0}".format(r))
        
        #c) 各プロセスで変換されたフィーチャクラスを都道府県のFGDBへマージしたものを作成（"Farmland"）
        #～省略～
        
        # フィールドエイリアスを設定
        #～省略～
        
        # コード値ドメインを設定
        #～省略～
        
        #d) マージが終わったので後片付け 各市区町村のFGDBを削除
        #～省略～
        
        fin = datetime.datetime.now()
        arcpy.AddMessage(u"-- Finish: MP_Farmland_JsonToFeatureClass --:{0}".format(fin))
        arcpy.AddMessage(u"     Elapsed time:{0}".format(fin-start))
    except:
        arcpy.AddError(u"Exception:{0}".format(sys.exc_info()[2]))

 

ジオプロセシングツールへの組み込み

Zpracování převodu polygonů zemědělské půdy ve formátu GeoJSON končí u uvedeného kódu, proto dálePython toolbox vysvětlíme volání z geoprocessingového toolboxu definovaného v

V Python toolboxu je, jak je uvedeno v nápovědě „Co je Python toolbox?“, geoprocessingový toolbox a geoprocessingový nástroj definovány jako třídy.

1) Definice toolboxu a přidání nástrojů

Dvě geoprocessingové třídy definované jako class AgrilandJsonConvTool a class AgrilandShpConvTool jsou při inicializaci třídy Toolbox

self.tools = [AgrilandJsonConvTool, AgrilandShpConvTool]

přidány.

#Import modulů pro multithread processing
from MP_Farmland_JsonToFeatureClass import exec_batch_convert as exec_batch_json_convert
from MP_Farmland_ShapefileToFeatureClass import exec_batch_convert as exec_batch_shp_convert

#Definice Python toolboxu
class Toolbox:
    def __init__(self):
    self.label = "Nástroje pro převod polygonů zemědělské půdy"
    self.alias = ""
    self.tools = [AgrilandJsonConvTool, AgrilandShpConvTool]

#Definice jednotlivých geoprocessingových nástrojů
class AgrilandJsonConvTool:
    #～vynecháno～
class AgrilandShpConvTool:
    #～vynecháno～

 

2) Definice nástroje

Každý geoprocessingový nástroj musí mít minimálně definovány metody __init__, getParameterInfo a execute, aby mohl fungovat jako geoprocessingový nástroj v ArcGIS Pro.

V tomto příkladu metoda execute pouze volá funkci exec_batch_convert definovanou ve skriptu MP_Farmland_JsonToFeatureClass.py (pro stejný název funkce ve formátu shapefile je použito alias exec_batch_shp_convert při importu).

class AgrilandJsonConvTool:
    def __init__(self):
        self.label = "01_Multithread processing sample_Nástroj pro převod polygonů zemědělské půdy (formát GeoJSON)"
        self.description = ""
        
    def getParameterInfo(self):
		    param0 = arcpy.Parameter(
		        displayName="Vstupní složka (rozbalená složka prefektury)", 
		        name="input_folder", 
		        datatype="DEFolder", 
		        parameterType="Required",
		        direction="Input")
		    #～vynecháno～
        params = [param0, param1, param2]
        return params
        
		def execute(self, parameters, messages):
        infolder = parameters[0].valueAsText
        outfolder = parameters[1].valueAsText
        cpu_cnt = int(parameters[2].valueAsText)
        exec_batch_json_convert(infolder, outfolder, cpu_cnt)
        return

 

3) Uživatelské rozhraní při spuštění

Toto je uživatelské rozhraní skutečného nástroje „Farmland polygon conversion tools“.

 

Poznámka (převod souboru GeoJSON)

Pro převod souboru GeoJSON v ArcGIS Pro se běžně používá „JSON → Features (JSON To Features)“,」で変換するのが一般的ですが、次のような制限があるため、今回は自前のクラス（class FarmlandGeojsonToFeaturesEx）で対応しています。

• ジオメトリがPolygonZ として定義される
• フィールド型が自動的に決まり、後から型の再変換が必要となる場合が生じる
• 農地筆ポリゴンのジオメトリのCRSコードの定義にツールが未対応

 

まとめ

過去にブログでとりあげた、ArcPy と Python でのマルチプロセス処理 を題材に、

• Python 3.3 以上でのコーディング パターンでアップデート
• ArcGIS Pro のジオプロセシング ツールへの組み込み例を追加

した「Farmland polygon conversion tools」のサンプルをもとに、コーティングする際のポイントを中心に解説させていただきました。

データの処理の上で、必ずしもマルチプロセスで行う必要はないかもしれませんが、必要に応じて、本記事のベスト プラクティスやコーディング パターンを参考にしていただければと思います。

 

関連リンク

ArcGIS の Python でのマルチプロセス処理

ArcGIS の Python でのマルチプロセス処理-実装例