Background
My project is a Python toolbox. This is my first bigger coding project and it took me quite some time to code this toolbox. I learned a lot from the start point to this day. I changed this code hundreds of times, whenever I learned about new trick or method. This code works. I tested it with various datasets, fixed bugs or any syntax errors.
Cons:
- The code ha some Polish names for: variable, functions etc.
- GUI is all Polish
- I started using Python about 3 months ago
What my code does:
Main purpose of this toolbox was to automate OpenStreetMap (OSM) data transformation from voivodeship shapefiles into country sized one, from which values were selected by their attributes to visualize features (for example, roads were selected and symbolized).
The code consists of three classes which are three scripts inside of my toolbox.
It is used in ArcGIS Pro to help non-programmer user to replicate my work.
My goal
Can someone who is more experienced than me in Python give me some useful advice?
Terms used in this code
- shp - shapefile
- osm - OpenStreetMap
- fc - feature class
- gdb - geodatabase
I added comments to my code to help understand what is happening.
My code
# -*- coding: CP1250 -*-
import arcpy
import os
import pandas as pd
import shutil
import xlrd
from xml.etree import ElementTree as ET
import glob
from itertools import starmap
import re
class Toolbox(object):
def __init__(self):
"""Define the toolbox (the name of the toolbox is the name of the
.pyt file)."""
self.label = "NardzedziaDoEskportu"
self.alias = ""
# List of tool classes associated with this toolbox
self.tools = [Przygotowanie_do_eksportu, SkryptDoEksportu, XML_export]
class SkryptDoEksportu(object):
def __init__(self):
"""Define the tool (tool name is the name of the class)."""
self.label = "OSM Polska"
self.description = "Skrypt eksportuje wybrane kolumny zawarte w tabeli atrybutow klas obiektow z geobazy."
self.canRunInBackground = False
def getParameterInfo(self):
"""Define parameter definitions"""
# Pierwszy parametr
inside = arcpy.Parameter(
displayName="Wejsciowa geobaza",
name="in_gdb",
datatype="DEWorkspace",
parameterType="Required",
direction="Input")
# drugi parametr
klasy = arcpy.Parameter(
displayName="Warstwy w geobazie (mozliwy tylko podglad)",
name="fcs_of_gdb",
datatype="DEFeatureClass",
parameterType="Required",
direction="Input",
multiValue=True)
# trzeci parametr
kolumny = arcpy.Parameter(
displayName="Wybierz kolumny do selekcji",
name="colli",
datatype="GPString",
parameterType="Required",
direction="Input",
multiValue=True)
kolumny.filter.type = "ValueList"
# Czwarty parametr
plikExcel = arcpy.Parameter(
displayName="Plik *.XLS z domenami",
name="excelik",
datatype="DEType",
parameterType="Required",
direction="Input")
# Piaty parametr
plikShpWoj = arcpy.Parameter(
displayName="Plik *.Shp okreslajacy granice wojewodztw",
name="ShpWoj",
datatype="DEShapefile",
parameterType="Required",
direction="Input")
# Szosty parametr
plikBoundary = arcpy.Parameter(
displayName="Plik *.Shp bedacy poprawiona wersja Polska_boundary_ply",
name="shpBoundary",
datatype="DEShapefile",
parameterType="Required",
direction="Input")
p = [inside, klasy, kolumny, plikExcel, plikShpWoj, plikBoundary]
return p
def isLicensed(self):
"""Set whether tool is licensed to execute."""
return True
def updateParameters(self, parameters):
"""Modify the values and properties of parameters before internal
validation is performed. This method is called whenever a parameter
has been changed."""
parameters[1].enabled = 0
if parameters[0].value:
arcpy.env.workspace = parameters[0].value
fclist = arcpy.ListFeatureClasses()
parameters[1].value = fclist
if parameters[1].value:
fcs = parameters[1].value.exportToString()
single = fcs.split(";")
fields = arcpy.ListFields(single[0])
l1 = [f.name for f in fields]
l2 = ["OBJECTID", "Shape", "OSMID", "osmTags", "osmuser", "osmuid", "osmvisible",
"osmversion", "osmchangeset", "osmtimestamp", "osmMemberOf", "osmSupportingElement",
"osmMembers", " Shape_Length", "Shape_Area", "wayRefCount"]
l3 = [czynnik for czynnik in l1 if czynnik not in l2]
parameters[2].filter.list = l3
return
def updateMessages(self, parameters):
"""Modify the messages created by internal validation for each tool
parameter. This method is called after internal validation."""
def execute(self, parameters, messages):
# Variables
arcpy.env.overwriteOutput = True
gdb = parameters[0].valueAsText
wybor_uzytkownika = parameters[2].valueAsText
excel = parameters[3].valueAsText
granice_woj_shp = parameters[4].valueAsText
boundary_ply_shp = parameters[5].valueAsText
arcpy.env.workspace = gdb
warunek = " <> ''"
tymczasowa_nazwa = "tymczasowaNazwaDlaFC"
lista_ln = []
lista_ply = []
lista_pt = []
# Appends feature classes to lists and then merges them to single fc based on geometry
fc_lista = arcpy.ListFeatureClasses()
listy_append(
fc_lista, lista_ln, lista_ply, lista_pt)
tupel_merge = (
[lista_ln, "Polska_ln"],
[lista_ply, "Polska_ply"],
[lista_pt, "Polska_pt"])
list(starmap(
arcpy.Merge_management,tupel_merge))
fc_lista = arcpy.ListFeatureClasses()
# Deleting useless feature classes
for fc in fc_lista:
czlon_nazwy = fc.split("_")
if czlon_nazwy[0] != "Polska":
arcpy.Delete_management(fc)
# Column split
kolumny_split(
wybor_uzytkownika, tymczasowa_nazwa, warunek,
gdb, granice_woj_shp, boundary_ply_shp)
# File import from excel to create domain lists
import_excel(
excel, gdb)
# Adding domains
nadaj_domene(
gdb, wybor_uzytkownika)
return
class XML_export(object):
def __init__(self):
"""Define the tool (tool name is the name of the class)."""
self.label = "Eksport danych z XML"
self.description = "Skrypt przygotowuje dane i eksportuje wybrane aspkety z XML"
self.canRunInBackground = False
def getParameterInfo(self):
"""Define parameter definitions"""
# Pierwszy parametr
inside = arcpy.Parameter(
displayName = "Wejsciowa geobaza",
name = "in_gdb",
datatype = "DEWorkspace",
parameterType = "Required",
direction = "Input",
multiValue = False)
# drugi parametr
rodzaj = arcpy.Parameter(
displayName = "Wybierz typ geometrii",
name = "geom",
datatype = "GPString",
parameterType = "Required",
direction = "Input",
multiValue = False)
rodzaj.filter.type = "ValueList"
rodzaj.filter.list = ['pt','ln','ply']
# trzeci parametr
klasy = arcpy.Parameter(
displayName = "Wybrane klasy",
name = "fcs_of_gdb",
datatype = "DEFeatureClass",
parameterType = "Required",
direction = "Input",
multiValue = True)
# czwarty
wojewodztwa_string = arcpy.Parameter(
displayName = "Wybierz wojewodztwa",
name = "colli",
datatype = "GPString",
parameterType = "Required",
direction = "Input",
multiValue = True)
wojewodztwa_string.filter.type = "ValueList"
#piaty
warstwa = arcpy.Parameter(
displayName = "Wybierz warstwe",
name = "fl_gdb",
datatype = "GPFeatureLayer",
parameterType = "Required",
direction = "Input")
# szosty
wyrazenie = arcpy.Parameter(
displayName = "Wpisz wyrazenie do selekcji",
name = "expres",
datatype = "GPSQLExpression",
parameterType = "Required",
direction = "Input")
wyrazenie.parameterDependencies = [warstwa.name]
# siodmy
folder_xml = arcpy.Parameter(
displayName = "Wskaz folder gdzie znajduja sie pliki w formacie XML",
name = "XMLdir",
datatype = "DEFolder",
parameterType = "Required",
direction = "Input")
# osmy
folder_csv = arcpy.Parameter(
displayName = "Wskaz folder gdzie maja zostac zapisane pliki CSV",
name = "CSVdir",
datatype = "DEFolder",
parameterType = "Required",
direction = "Input")
#dziewiaty
kolumny = arcpy.Parameter(
displayName = "Wybierz kolumne",
name = "colli2",
datatype = "GPString",
parameterType = "Required",
direction = "Input",
multiValue = False)
kolumny.filter.type = "ValueList"
#dziesiaty
check_1 = arcpy.Parameter(
displayName = "Zaznacz aby dokonac zapisu do CSV (niezalecane odznaczanie)",
name = "check1",
datatype = "GPBoolean",
parameterType = "Optional",
direction = "Input",
multiValue = False)
check_1.value = True
#jedenasty
check_2 = arcpy.Parameter(
displayName = "Zaznacz aby polaczyc pliki CSV w jeden - odznaczenie spowoduje brak laczenia",
name = "check2",
datatype = "GPBoolean",
parameterType = "Optional",
direction = "Input",
multiValue = False)
p = [inside, rodzaj, klasy, wojewodztwa_string,
kolumny, warstwa, wyrazenie, folder_xml, folder_csv,
check_1, check_2]
return p
def isLicensed(self):
"""Set whether tool is licensed to execute."""
return True
def updateParameters(self, parameters):
"""Modify the values and properties of parameters before internal
validation is performed. This method is called whenever a parameter
has been changed."""
wejsciowa_gdb = parameters[0]
wybrana_geometria = parameters[1]
lista_klas = parameters[2]
wybor_wojewodztwa = parameters[3]
wybor_kolumny = parameters[4]
check_box_wartosc_1 = parameters[9].value
check_box_wartosc_2 = parameters[10].value
lista_klas.enabled = 0
arcpy.env.workspace = wejsciowa_gdb.value
fclist = arcpy.ListFeatureClasses()
fc_o_wybranej_geometrii = []
wybor = wybrana_geometria.valueAsText
if check_box_wartosc_2 and check_box_wartosc_1 == False:
parameters[0].enabled = 0
parameters[1].enabled = 0
parameters[3].enabled = 0
parameters[4].enabled = 0
parameters[5].enabled = 0
parameters[6].enabled = 0
if check_box_wartosc_1 and check_box_wartosc_2 == False:
parameters[0].enabled = 1
parameters[1].enabled = 1
parameters[3].enabled = 1
parameters[4].enabled = 1
parameters[5].enabled = 1
parameters[6].enabled = 1
for fc in fclist:
try:
split_nazwy = fc.split('_')
if len (split_nazwy) == 2 and split_nazwy[1] == wybor:
fc_o_wybranej_geometrii.append(fc)
except IndexError:
pass
lista_klas.value = fc_o_wybranej_geometrii
if lista_klas.value:
fcs = lista_klas.value.exportToString()
fcs_lista = fcs.split(";")
wybor_wojewodztwa.filter.list = fcs_lista
if wybrana_geometria.value:
if wybor == 'ln':
lista_ln = [
'highway', 'waterway', 'boundary'
]
wybor_kolumny.filter.list = lista_ln
elif wybor == 'pt':
lista_pt = [
'natural', 'aeroway', 'historic',
'leisure', 'waterway', 'shop',
'railway', 'tourism', 'highway',
'amenity'
]
wybor_kolumny.filter.list = lista_pt
elif wybor == 'ply':
lista_ply = [
'landuse', 'building', 'natural',
'amenity'
]
wybor_kolumny.filter.list = lista_ply
def updateMessages(self, parameters):
"""Modify the messages created by internal validation for each tool
parameter. This method is called after internal validation."""
def execute(self, parameters, messages):
# Zmienne
# -*- coding: CP1250 -*-
arcpy.env.overwriteOutput = True
tymczasowa_nazwa = "tymczasowaNazwaDlaFC"
gdb = parameters[0].valueAsText
user_geometry_choice = parameters[1].valueAsText
user_wojewodztwo_choice = parameters[3].valueAsText
user_column_choice = parameters[4].valueAsText
user_expression = parameters[6].valueAsText
dir_xml = parameters[7].valueAsText
dir_csv = parameters[8].valueAsText
field_osm = 'OSMID'
xml_parent_way = 'way'
xml_parent_node = 'node'
xml_atr_parent = 'id'
xml_child = 'tag'
xml_atr_child = 'k'
xml_value_child_1 = 'name'
xml_value_child_2 = 'v'
xml_value_child_3 = 'ele'
xml_value_child_4 = 'addr:housenumber'
xml_value_child_5 = 'ref'
id_csv = 'id_robocze'
id_csv_2 = 'id_elementu'
nazwa_csv = 'nazwa'
natural_name = "nazwa_ele"
natural_name_2 = "wysokosc"
building_name = "budynki_nazwa"
building_name_2 = "buydnki_numery"
natural_csv_name = 'natural_nazwa'
natural_csv_name_2 = 'natural_wysokosc'
building_csv_name = 'budynki_nazwa'
building_csv_name_2 = 'budynki_numery'
highway_name = 'ulice'
highway_name_2 = 'nr_drogi'
highway_csv_name = 'ulice'
highway_csv_name_2 = 'nr_drogi'
check_box_wartosc_1 = parameters[9].value
check_box_wartosc_2 = parameters[10].value
dir_natural = os.path.join(
dir_csv,'Polska_{0}_{1}.csv'.format(natural_csv_name,
user_geometry_choice))
dir_natural_2 = os.path.join(
dir_csv,'Polska_{0}_{1}.csv'.format(natural_csv_name_2,
user_geometry_choice))
dir_any = os.path.join(
dir_csv,'Polska_{0}_{1}.csv'.format(user_column_choice,
user_geometry_choice))
dir_building = os.path.join(
dir_csv,'Polska_{0}_{1}.csv'.format(building_csv_name,
user_geometry_choice))
dir_building_2 = os.path.join(
dir_csv,'Polska_{0}_{1}.csv'.format(building_csv_name_2,
user_geometry_choice))
dir_highway = os.path.join(
dir_csv,'Polska_{0}_{1}.csv'.format(highway_csv_name,
user_geometry_choice))
dir_highway_2 = os.path.join(
dir_csv,'Polska_{0}_{1}.csv'.format(highway_csv_name_2,
user_geometry_choice))
# Selekcja z geobazy plikow, ktore zostana wykorzystane do stworzenia list fc
if check_box_wartosc_1:
selektor_pre(
gdb, user_geometry_choice, user_wojewodztwo_choice,
user_column_choice, tymczasowa_nazwa, user_expression)
get_csv(
gdb, user_geometry_choice, user_column_choice, field_osm, dir_xml,
xml_parent_node, xml_atr_parent, xml_child, xml_atr_child,
xml_value_child_1, xml_value_child_3, dir_csv, natural_csv_name,
natural_csv_name_2, id_csv, natural_name, natural_name_2,
xml_value_child_4, building_csv_name, building_csv_name_2,
building_name, building_name_2, xml_value_child_2, nazwa_csv,
xml_parent_way, highway_csv_name, highway_csv_name_2,
highway_name, highway_name_2, xml_value_child_5,
user_geometry_choice, user_column_choice, check_box_wartosc_1,
check_box_wartosc_2, id_csv_2, dir_natural, dir_natural_2, dir_any,
dir_building, dir_building_2, dir_highway, dir_highway_2)
return
class Przygotowanie_do_eksportu(object):
def __init__(self):
"""Define the tool (tool name is the name of the class)."""
self.label = "Eliminacja datasetow"
self.description = "Skrypt przygotowuje dane w geobazie, aby spelnialy wymagania nastepnego skryptu."
self.canRunInBackground = False
def getParameterInfo(self):
"""Define parameter definitions"""
# Pierwszy parametr
inside = arcpy.Parameter(
displayName="Wejsciowa geobaza",
name="in_gdb",
datatype="DEWorkspace",
parameterType="Required",
direction="Input")
p =[inside]
return p
def isLicensed(self):
"""Set whether tool is licensed to execute."""
return True
def updateParameters(self, parameters):
"""Modify the values and properties of parameters before internal
validation is performed. This method is called whenever a parameter
has been changed."""
def updateMessages(self, parameters):
"""Modify the messages created by internal validation for each tool
parameter. This method is called after internal validation."""
def execute(self, parameters, messages):
arcpy.env.overwriteOutput = True
arcpy.env.workspace = parameters[0].valueAsText
alt = arcpy.env.workspace
datalist = arcpy.ListDatasets()
#clears gdb out of data sets
for data in datalist:
for fc in arcpy.ListFeatureClasses("*", "ALL", data):
czesc = fc.split("_")
arcpy.FeatureClassToFeatureClass_conversion(
fc, alt, '{0}_{1}'.format(czesc[0], czesc[2]))
arcpy.Delete_management(data)
return
def import_excel(
in_excel, out_gdb):
"""
Opens excel file from path
Make a list from sheets in file
Iterates through sheets
"""
workbook = xlrd.open_workbook(in_excel)
sheets = [sheet.name for sheet in workbook.sheets()]
for sheet in sheets:
out_table = os.path.join(
out_gdb,
arcpy.ValidateTableName(
"{0}".format(sheet),
out_gdb))
arcpy.ExcelToTable_conversion(in_excel, out_table, sheet)
def iter_kolumny(
user_input, tymczasowa_mazwa,
warunek):
"""
Selection based on user choice
"""
lista_kolumn = user_input.split(";")
arcpy.AddMessage(
"Wybrales nastepujace parametry: {0}".format(lista_kolumn))
fc_lista = arcpy.ListFeatureClasses()
for fc in fc_lista:
czlon_nazwy = fc.split("_")
for kolumna in lista_kolumn:
arcpy.MakeFeatureLayer_management(fc, tymczasowa_mazwa)
try:
arcpy.SelectLayerByAttribute_management(
tymczasowa_mazwa, "NEW_SELECTION", '{0}{1}'.format(kolumna, warunek))
arcpy.CopyFeatures_management(
tymczasowa_mazwa, '{0}_{1}_{2}'.format(czlon_nazwy[0], kolumna, czlon_nazwy[1]))
except arcpy.ExecuteError:
pass
arcpy.Delete_management(fc)
def kolumny_split(
user_input, tymczasowa_mazwa, warunek,
gdb, wojewodztwa_shp, boundary_ply):
"""
After iter_kolumny call faulty column is deleted,
and new fc is imported which will be substitute for it
"""
iter_kolumny(
user_input, tymczasowa_mazwa, warunek)
arcpy.Delete_management(
'Polska_boundary_ply')
arcpy.FeatureClassToFeatureClass_conversion(
wojewodztwa_shp, gdb, 'GraniceWojewodztw')
arcpy.FeatureClassToFeatureClass_conversion(
boundary_ply, gdb, 'Polska_boundary_ply')
def listy_append(
listaFc, liniowa, polygon, punkty):
"""
Simple list appender
"""
for fc in listaFc:
czlon_nazwy = fc.split("_")
if czlon_nazwy[1] == "ln":
liniowa.append(fc)
elif czlon_nazwy[1] == "ply":
polygon.append(fc)
elif czlon_nazwy[1] == "pt":
punkty.append(fc)
def nadaj_domene(
work_space, wybor_uzytkownika):
"""
Function firstly makes list out of
user choice, then appends only those fcs which
are in gdb, then applies only domains which are wanted by user
(determined by fc choice)
"""
arcpy.env.workspace = work_space
lista_kolumn = wybor_uzytkownika.split(";")
all_tabele_gdb = arcpy.ListTables()
lista_poprawiona_o_kolumny = []
for tabela in all_tabele_gdb:
pierwszy_czlon_nazwy = tabela.split("_")[0]
if pierwszy_czlon_nazwy in lista_kolumn:
lista_poprawiona_o_kolumny.append(tabela)
elif pierwszy_czlon_nazwy == 'man':
lista_poprawiona_o_kolumny.append(tabela)
else:
arcpy.Delete_management(tabela)
for tabela in lista_poprawiona_o_kolumny:
lista_robocza = []
lista_robocza.append(tabela)
nazwa_domeny = lista_robocza[0]
arcpy.TableToDomain_management(
tabela, 'CODE', 'DESCRIPTION', work_space, nazwa_domeny, '-', 'REPLACE')
arcpy.Delete_management(tabela)
def selektor_pre(
baza_in, geometria, wojewodztwa,
kolumna, tymczasowa_nazwa, user_expression):
"""
Selects features based on user expression
"""
arcpy.env.workspace = baza_in
fc_lista = wojewodztwa.split(';')
arcpy.AddMessage(fc_lista)
for fc in fc_lista:
arcpy.MakeFeatureLayer_management(
fc, tymczasowa_nazwa)
arcpy.SelectLayerByAttribute_management(
tymczasowa_nazwa, "NEW_SELECTION", user_expression)
arcpy.CopyFeatures_management(
tymczasowa_nazwa, '{0}_{1}'.format(fc, kolumna))
arcpy.AddMessage(
'Seleckja skonczona dla {0}_{1}'.format(fc, kolumna))
def compare_save_to_csv(
gdb, pole_osm, xml_folder,
kolumna, parent,atrybut_parent, child,
child_atrybut, child_value_1, child_value_2,
csv_dir, nazwa_pliku, nazwa_id, nazwa_atrybutu,
user_geometry_choice):
"""
Iterates over feature classes in geodatabase,
checks for only those which user needs,
creates list of ids which will be used in xml_parser
"""
arcpy.env.workspace = gdb
wszystkie_fc = arcpy.ListFeatureClasses()
for fc in wszystkie_fc:
try:
split = fc.split('_')
if split[2] == kolumna and split[1] == user_geometry_choice:
czesc_nazwy = split[0]
geom = split[1]
nazwa_pliku = '{0}_{1}'.format(kolumna, geom)
lista_id_arcgis = [row[0]
for row in arcpy.da.SearchCursor(fc, pole_osm)]
arcpy.AddMessage("Dlugosc listy: {0}".format(
str(len(lista_id_arcgis))))
xml_parser(
'{0}\{1}.xml'.format(xml_folder, czesc_nazwy),
lista_id_arcgis, parent,
atrybut_parent, child, child_atrybut,
child_value_1, child_value_2, nazwa_pliku,
csv_dir, nazwa_id, nazwa_atrybutu,czesc_nazwy)
except IndexError:
pass
def compare_save_to_csv_wyjatek(
gdb, user_geometry_choice, user_column_choice,
pole_osm, xml_folder, kolumna, parent, atrybut_parent,
child, child_atrybut, child_value_1, child_value_2,
child_value_3, sciezka_csv, csv_name, csv_name_2,
nazwa_id, nazwa_atrybutu, nazwa_atrybutu_2):
"""
Iterates over feature classes in geodatabase,
checks for only those which user needs,
creates list of ids which will be used in xml_parser_wyjatki
"""
arcpy.env.workspace = gdb
wszystkie_fc = arcpy.ListFeatureClasses()
for fc in wszystkie_fc:
try:
split = fc.split('_')
if split[2] == kolumna and split[1] == user_geometry_choice:
czesc_nazwy = split[0]
lista_id_arcgis = [row[0]
for row in arcpy.da.SearchCursor(fc, pole_osm)]
arcpy.AddMessage("Dlugosc listy: {0}".format(
str(len(lista_id_arcgis))))
xml_parser_wyjatki(
'{0}\{1}.xml'.format(xml_folder, czesc_nazwy),
lista_id_arcgis, parent, atrybut_parent, child, child_atrybut,
child_value_1, child_value_2, child_value_3, sciezka_csv,
csv_name, csv_name_2, nazwa_id, nazwa_atrybutu, nazwa_atrybutu_2, czesc_nazwy)
except IndexError:
pass
def merge_csv(
sciezka_csv, fragment_nazwy, nazwa_csv):
"""
Merges csv in specifed directory based on name scheme
"""
results = pd.DataFrame([])
for counter, file in enumerate(glob.glob("{0}\*{1}*".format(sciezka_csv, fragment_nazwy))):
name_dataframe = pd.read_csv(
file, usecols=[0, 1],encoding = 'CP1250' )
results = results.append(
name_dataframe)
results.to_csv(
'{0}\{1}.csv'.format(sciezka_csv, nazwa_csv), encoding = 'CP1250')
def zapis_do_csv(
lista_1, lista_2, nazwa_1,
nazwa_2, csv_name, katalog,
czesc_nazwy):
"""
Saves to CSV, based on 2 lists.
"""
raw_data = {nazwa_1: lista_1,
nazwa_2: lista_2}
df = pd.DataFrame(raw_data, columns=[nazwa_1, nazwa_2])
df.to_csv(
'{0}\{1}_{2}.csv'.format(katalog, czesc_nazwy, csv_name),
index=False, header=True, encoding = 'CP1250')
def xml_parser(
xml, lista_agis, parent,
atrybut_parent, child, child_atrybut,
child_value_1, child_value_2, nazwa_pliku,
sciezka_csv, nazwa_id, nazwa_atrybutu,
czesc_nazwy):
"""
Function to pick from xml files tag values.
Firstly it creates tree of xml file and then
goes each level down and when final condtion is fullfiled
id and value from xml file is appended to list in the end of
xml file list is saved to CSV.
"""
rootElement = ET.parse(xml).getroot()
l1 = []
l2 = []
for subelement in rootElement:
if subelement.tag == parent:
if subelement.get(atrybut_parent) in lista_agis:
for sselement in subelement:
if sselement.tag == child:
if sselement.attrib[child_atrybut] == child_value_1:
l1.append(
subelement.get(atrybut_parent))
l2.append(
sselement.get(child_value_2))
zapis_do_csv(
l1, l2, nazwa_id,
nazwa_atrybutu, nazwa_pliku,
sciezka_csv, czesc_nazwy)
arcpy.AddMessage('Zapisalem {0}'.format(nazwa_pliku))
arcpy.AddMessage('Zapsialem tyle id: {0}'.format((len(l1))))
arcpy.AddMessage('Zapsialem tyle nazw: {0}'.format((len(l2))))
def xml_parser_wyjatki(
xml, lista_agis,
parent, atrybut_parent, child,
child_atrybut, child_value_1, child_value_2,
child_value_3, sciezka_csv, nazwa_pliku, nazwa_pliku_2,
nazwa_id, nazwa_atrybutu, nazwa_atrybutu_2, czesc_nazwy):
"""
Function to pick from xml files tag values.
Firstly it creates tree of xml file and then
goes each level down and when final condtion is fullfiled
id and value from xml file is appended to list in the end of
xml file list is saved to CSV.
Added 'elif' for some feature classes that are described
by 2 value tags.
"""
rootElement = ET.parse(xml).getroot()
l1 = []
l2 = []
l3 = []
l4 = []
for subelement in rootElement:
if subelement.tag == parent:
if subelement.get(atrybut_parent) in lista_agis:
for sselement in subelement:
if sselement.tag == child:
if sselement.attrib[child_atrybut] == child_value_1:
l1.append(
subelement.get(atrybut_parent))
l2.append(
sselement.get(child_value_2))
arcpy.AddMessage('Dodalem {0}'.format(sselement.get(child_value_2)))
elif sselement.attrib[child_atrybut] == child_value_3:
l3.append(
subelement.get(atrybut_parent))
l4.append(
sselement.get(child_value_2))
arcpy.AddMessage('Dodalem {0}'.format(sselement.get(child_value_2)))
zapis_do_csv(
l1, l2,
nazwa_id, nazwa_atrybutu,
nazwa_pliku, sciezka_csv, czesc_nazwy)
zapis_do_csv(
l3, l4,
nazwa_id, nazwa_atrybutu_2,
nazwa_pliku_2, sciezka_csv, czesc_nazwy)
def replace_csv(
csv, symbol_1, symbol_2):
'''
Function replace certain symbol to prevent
ArcGIS Pro from crashing during table import.
'''
my_csv_path = csv
with open(my_csv_path, 'r') as f:
my_csv_text = f.read()
find_str = symbol_1
replace_str = symbol_2
csv_str = re.sub(find_str, replace_str, my_csv_text)
with open(my_csv_path, 'w') as f:
f.write(csv_str)
def get_csv(
gdb, geom_choice, column_choice, field_osm, dir_xml,
xml_parent_node, xml_atr_parent, xml_child, xml_atr_child,
xml_value_child_1, xml_value_child_3, dir_csv, natural_csv_name,
natural_csv_name_2, id_csv, natural_name, natural_name_2,
xml_value_child_4, building_csv_name, building_csv_name_2,
building_name, building_name_2, xml_value_child_2, nazwa_csv,
xml_parent_way, highway_csv_name, highway_csv_name_2,
highway_name, highway_name_2, xml_value_child_5, user_geometry_choice,
user_column_choice, check_box_wartosc_1, check_box_wartosc_2,
id_csv_2, dir_natural, dir_natural_2, dir_any, dir_building,
dir_building_2, dir_highway, dir_highway_2):
'''
Combination of all other functions to deliver new fields in feature
classes in geodatabase.
'''
wybrana_kolumna = column_choice
if geom_choice == 'pt':
if wybrana_kolumna == 'natural':
if check_box_wartosc_1:
compare_save_to_csv_wyjatek(
gdb, user_geometry_choice, user_column_choice,
field_osm, dir_xml, wybrana_kolumna,
xml_parent_node, xml_atr_parent, xml_child,
xml_atr_child, xml_value_child_1, xml_value_child_2,
xml_value_child_3, dir_csv, natural_csv_name,
natural_csv_name_2, id_csv, natural_name, natural_name_2)
if check_box_wartosc_2:
tupel_pt = (
[dir_natural],
[dir_natural_2])
list(starmap(
merge_csv, tupel_pt))
tupel_pt_2 = (
[dir_natural, ';', ' '],
[dir_natural, ':', ' '],
[dir_natural_2, ';', ' '],
[dir_natural_2, ':', ' '])
list(starmap(
replace_csv, tupel_pt_2))
tupel_pt_3 = (
[dir_natural, gdb, id_csv_2, id_csv,
'Polska_natural_pt', field_osm, natural_name],
[dir_natural_2, gdb, id_csv_2, id_csv,
'Polska_natural_pt', field_osm, natural_name_2])
list(starmap(
import_fix_join, tupel_pt_3))
else:
if check_box_wartosc_1:
compare_save_to_csv(
gdb, field_osm, dir_xml, wybrana_kolumna,
xml_parent_node, xml_atr_parent, xml_child,
xml_atr_child, xml_value_child_1, xml_value_child_2,
dir_csv, wybrana_kolumna, id_csv, nazwa_csv, user_geometry_choice)
if check_box_wartosc_2:
merge_csv(
dir_csv, wybrana_kolumna,
'Polska_{0}_{1}'.format(wybrana_kolumna, geom_choice))
tupel_pt_4 = (
[dir_any, ':', ' '],
[dir_any, ';', ' '])
list(starmap(
replace_csv, tupel_pt_4))
import_fix_join(
dir_any, gdb, id_csv_2, id_csv,
'Polska_{0}_{1}'.format(wybrana_kolumna, geom_choice),
field_osm, nazwa_csv)
elif geom_choice == 'ply':
if wybrana_kolumna == 'building':
if check_box_wartosc_1:
compare_save_to_csv_wyjatek(
gdb, user_geometry_choice, user_column_choice,
field_osm, dir_xml, wybrana_kolumna,
xml_parent_way, xml_atr_parent, xml_child,
xml_atr_child, xml_value_child_1, xml_value_child_2,
xml_value_child_4, dir_csv, building_csv_name,
building_csv_name_2, id_csv, building_name, building_name_2)
if check_box_wartosc_2:
tupel_ply = ([
dir_csv, building_csv_name, 'Polska_{0}_{1}'.format(building_csv_name,
user_geometry_choice),
dir_csv, building_csv_name_2, 'Polska_{0}_{1}'.format(building_csv_name_2,
user_geometry_choice)])
list(starmap(
merge_csv, tupel_ply))
tupel_ply_2 = (
[dir_building, ';', ' '],
[dir_building, ':', ' '],
[dir_building_2, ':', ' '],
[dir_building_2, ';', ' '])
list(starmap(
replace_csv, tupel_ply_2))
tupel_ply_3 = (
[dir_building, gdb, id_csv_2, id_csv,
'Polska_building_ply', field_osm, building_name],
[dir_building_2, gdb, id_csv_2, id_csv,
'Polska_building_ply', field_osm, building_name_2])
list(starmap(
import_fix_join, tupel_ply_3))
else:
if check_box_wartosc_1:
compare_save_to_csv(
gdb, field_osm, dir_xml, wybrana_kolumna,
xml_parent_way, xml_atr_parent, xml_child,
xml_atr_child, xml_value_child_1, xml_value_child_2,
dir_csv, wybrana_kolumna, id_csv, nazwa_csv, user_geometry_choice)
if check_box_wartosc_2:
merge_csv(
dir_csv, wybrana_kolumna,
'Polska_{0}_{1}'.format(wybrana_kolumna, geom_choice))
tupel_ply_4 = (
[dir_any , ':', ' '],
[dir_any , ';', ' '])
list(starmap(
replace_csv, tupel_ply_4))
import_fix_join(
dir_any, gdb, id_csv_2, id_csv,
'Polska_{0}_{1}'.format(wybrana_kolumna, geom_choice),
field_osm, nazwa_csv)
elif geom_choice == 'ln':
if wybrana_kolumna == 'highway':
if check_box_wartosc_1:
compare_save_to_csv_wyjatek(
gdb, user_geometry_choice, user_column_choice,
field_osm, dir_xml, wybrana_kolumna,
xml_parent_way, xml_atr_parent, xml_child,
xml_atr_child, xml_value_child_1, xml_value_child_2,
xml_value_child_5, dir_csv, highway_csv_name,
highway_csv_name_2, id_csv, highway_name, highway_name_2)
if check_box_wartosc_2:
tupel_ln = ([
dir_csv, highway_csv_name, 'Polska_{0}_{1}'.format(highway_csv_name,
user_geometry_choice),
dir_csv, highway_csv_name_2, 'Polska_{0}_{1}'.format(highway_csv_name_2,
user_geometry_choice)])
list(starmap(
merge_csv, tupel_ln))
tupel_ln_2 = (
[dir_highway, ';', ' '],
[dir_highway, ':', ' '],
[dir_highway_2, ':', ' '],
[dir_highway_2, ';', ' '])
list(starmap(
replace_csv, tupel_ln_2))
tupel_ln_3 = (
[dir_building, gdb, id_csv_2, id_csv,
'Polska_highway_ln', field_osm, highway_name],
[dir_building_2, gdb, id_csv_2, id_csv,
'Polska_highway_ln', field_osm, highway_name_2])
list(starmap(
replace_csv, tupel_ln_3))
else:
if check_box_wartosc_1:
compare_save_to_csv(
gdb, field_osm, dir_xml, wybrana_kolumna,
xml_parent_way, xml_atr_parent, xml_child,
xml_atr_child, xml_value_child_1, xml_value_child_2,
dir_csv, wybrana_kolumna, id_csv, nazwa_csv, user_geometry_choice)
if check_box_wartosc_2:
merge_csv(
dir_csv, wybrana_kolumna,
'Polska_{0}_{1}'.format(wybrana_kolumna, geom_choice))
tupel_ln_4 = (
[dir_any, ':', ' '],
[dir_any, ';', ' '])
list(starmap(
replace_csv, tupel_ln_4))
import_fix_join(
dir_any, gdb, id_csv_2, id_csv,
'Polska_{0}_{1}'.format(wybrana_kolumna, geom_choice),
field_osm, nazwa_csv)
def fix_field(
tabela , nazwa, pole):
"""
Imported tables has got not
valid field with ID. This fix that problem
by creating new on in text type, copying value
and deleting old one.
"""
arcpy.AddField_management(
tabela, nazwa, "TEXT", field_length = 20)
try:
with arcpy.da.UpdateCursor(tabela, [pole,nazwa]) as cursor:
for row in cursor:
row[1] = row[0]
cursor.updateRow(row)
except RuntimeError:
print(row[1])
del row,cursor
arcpy.DeleteField_management(tabela, [pole])
def import_fix_join(
in_table, out_gdb, nazwa,
id_csv, in_fc, field_osm,
pole_to_join):
"""
Imports table to geodatabase
Fixes its column
Join field to feature class.
"""
arcpy.TableToGeodatabase_conversion(
[in_table], out_gdb)
fix_field(
in_table, nazwa, id_csv)
pole = [pole_to_join]
arcpy.env.workspace = out_gdb
arcpy.JoinField_management(
in_fc, field_osm, in_table,
nazwa, pole)
Three scripts in ArcGIS Pro software.
Script number one GUI
Script number two GUI
Script number three GUI
