by
1 Solution
Accepted Solutions
by
by
by
Your python script has sistanxis and semantic problems in the Python_Toolbox.pyt, it can alter your script, open and run in ArcGIS Pro 2.9.2 see image. I recommend you to create a new New_Python_Toolbox.pyt, correct the problem with the dbfread module and use the modified attached scripts as reference.
# -*- coding: utf-8 -*-
import datetime
import os
import arcpy
import pandas as pd
#import dbfread
arcpy.env.overwriteOutput = True
class Toolbox(object):
def __init__(self):
"""Define the toolbox (the name of the toolbox is the name of the
.pyt file)."""
self.label = "Backfill Data Toolbox"
self.alias = "Backfill Data Toolbox"
# List of tool classes associated with this toolbox
self.tools = [Tool]
class Tool(object):
def __init__(self):
"""Define the tool (tool name is the name of the class)."""
self.label = "Backfill Data Toolbox"
self.description = "Script to process collars data and add it to the Locations Database."
self.canRunInBackground = False
def getParameterInfo(self):
"""Define parameter definitions"""
in_table = arcpy.Parameter(
name='in_table',
displayName='Dbase or Excel Input',
datatype='DEFile',
direction='Input',
parameterType='Required')
in_table.filter.list = ['dbf', 'xlsx', 'xsl', 'csv']
out_gdb = arcpy.Parameter(
name='out_gdb',
displayName='Output to GDB',
datatype='DEWorkspace',
direction='Output',
parameterType='Required'
)
animal_species = arcpy.Parameter(
name='animal_species',
displayName='Animal Species',
datatype='GPString',
direction='Input',
parameterType='Required'
)
animal_species.value = 'Wolf'
animal_species.filter.type = 'ValueList'
animal_species.filter.list = ['Mule Deer', 'Elk', 'Pronghorn', 'Big Horn Sheep', 'Wolf']
collar_manufacturer = arcpy.Parameter(
name='collar_manufacturer',
displayName='Collar Manufacturer',
datatype='GPString',
direction='Input',
parameterType='Required'
)
collar_manufacturer.value = 'Lotek'
collar_manufacturer.filter.type = 'ValueList'
collar_manufacturer.filter.list = ['Vectronix', 'Telonics', 'Sirtrack', 'Lotek', 'Manual Entry', 'Wufoo']
pack = arcpy.Parameter(
name='pack',
displayName=f'Pack',
datatype='GPString',
direction='Input',
parameterType='Optional',
enabled=False)
pack.filter.type = 'ValueList'
pack_list = ['a', 'list', 'of', 'strings']
pack_list.sort()
pack.filter.list = pack_list
entry_staff = arcpy.Parameter(
name='entry_staff',
displayName='Entry Staff',
datatype='GPString',
direction='Input',
parameterType='Optional',
enabled=False)
add_adm_location = arcpy.Parameter(
name='add_adm_location',
displayName='Add Administrative Locations (County, WMU, and Ownership)',
datatype='GPBoolean',
direction='Input',
parameterType='Optional',
enabled=False)
add_adm_location.value = True
status_table = arcpy.Parameter(
name='status_table',
displayName='Status Inputs',
datatype='GPGAValueTable',
direction='Input',
parameterType='Optional',
enabled=False)
status_table.columns = ([['GPString', 'Status'], ['GPDate', 'Start'], ['GPDate', 'End']])
status_table.filters[0].type = 'ValueList'
status_table.filters[0].list = ['InPack', 'Dispersed', 'Roaming']
params = [in_table, out_gdb, animal_species, collar_manufacturer, pack, entry_staff, add_adm_location,
status_table]
return params
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."""
if parameters[2].valueAsText == 'Wolf':
parameters[4].enabled = True
parameters[5].enabled = True
parameters[6].enabled = True
parameters[7].enabled = True
else:
parameters[4].enabled = False
parameters[5].enabled = False
parameters[6].enabled = False
parameters[7].enabled = False
return
def updateMessages(self, parameters):
"""Modify the messages created by internal validation for each tool
parameter. This method is called after internal validation."""
return
"""The source code of the tool."""
def xy_to_fc(self, parameters, table_for_xy, output_name, input_format):
collar_manufacturer = parameters[3].valueAsText
output_path = os.path.join(parameters[1].valueAsText, output_name)
# If output exists delete it, sometimes not deleting caused errors when processing it again
if arcpy.Exists(output_path):
arcpy.AddMessage(f'Deleting old feature class: {output_path}...')
arcpy.management.Delete(output_path)
if collar_manufacturer == 'Vectronix':
in_x_field = 'LONGITUDE'
in_y_field = 'LATITUDE'
in_z_field = 'HEIGHT'
elif collar_manufacturer == 'Sirtrack' or collar_manufacturer == 'Telonics':
if input_format == 'csv':
in_x_field = 'Longitude'
in_y_field = 'Latitude'
in_z_field = None
else:
in_x_field = 'Longitude_'
in_y_field = 'Latitude_d'
in_z_field = None
elif collar_manufacturer == 'Lotek':
in_x_field = 'Longitude'
in_y_field = 'Latitude'
in_z_field = 'Altitude'
else:
raise Exception('Error in collar_manufacturer parameter.')
# Process: Make XY Event Layer (Make XY Event Layer) (management)
arcpy.AddMessage('Making XY Event Layer...')
xy_event_layer = 'memory/xy_event_layer'
arcpy.management.MakeXYEventLayer(table=table_for_xy,
in_x_field=in_x_field,
in_y_field=in_y_field,
out_layer=xy_event_layer,
spatial_reference=arcpy.SpatialReference(4269),
in_z_field=in_z_field)
# Generate GDB if it doesn't exist
if not arcpy.Exists(os.path.join(parameters[1].valueAsText)):
full_path = parameters[1].valueAsText
full_path_list = full_path.split('\\')
out_folder_path = '\\'.join(full_path_list[:-1])
out_name = full_path_list[-1]
arcpy.AddMessage(out_folder_path)
arcpy.AddMessage(out_name)
arcpy.management.CreateFileGDB(out_folder_path, out_name)
# Process: Feature Class To Feature Class (Feature Class To Feature Class) (conversion)
arcpy.AddMessage('Copying event layer to a feature class...')
arcpy.conversion.FeatureClassToFeatureClass(in_features=xy_event_layer,
out_path=os.path.join(parameters[1].valueAsText),
out_name=output_name,
where_clause="",
config_keyword="")
# Grab original fields to delete later
original_fields = arcpy.ListFields(os.path.join(parameters[1].valueAsText, output_name))
# Calculate Fields:
arcpy.AddMessage('Calculating the Collar_ID field...')
if collar_manufacturer == 'Vectronix':
arcpy.management.CalculateField(output_path, 'Collar_ID',
expression=f'!CollarID!',
field_type='TEXT'
)
elif collar_manufacturer == 'Sirtrack':
arcpy.management.CalculateField(output_path, 'Collar_ID',
expression=f'!Tag_ID!',
field_type='TEXT'
)
elif collar_manufacturer == 'Lotek':
arcpy.management.CalculateField(output_path, 'Collar_ID',
expression='!Device_ID!',
field_type='TEXT'
)
arcpy.AddMessage('Calculating the Lat_DDeg field...')
arcpy.management.CalculateField(output_path, 'Lat_DDeg',
expression=f'!{in_y_field}!',
field_type='DOUBLE'
)
arcpy.AddMessage('Calculating the Long_DDeg field...')
arcpy.management.CalculateField(output_path, 'Long_DDeg',
expression=f'!{in_x_field}!',
field_type='DOUBLE'
)
arcpy.AddMessage('Calculating the LL_Datum field...')
arcpy.management.CalculateField(output_path, 'LL_Datum',
expression="""'WGS84'""",
field_type='TEXT'
)
arcpy.AddMessage('Calculating the Accuracy field...')
if collar_manufacturer == 'Vectronix' or collar_manufacturer == 'Lotek':
arcpy.management.CalculateField(output_path, 'Accuracy',
expression='dop(!DOP!)',
expression_type='PYTHON 3',
code_block="""def dop(dop):
if dop <= 6:
return "< 20 yds"
elif dop >= 20:
return "250 yds - 1/2 mile"
else:
return "20 - 250 yds"
""",
field_type='TEXT'
)
arcpy.AddMessage('Calculating the Species field...')
arcpy.management.CalculateField(output_path, 'Species',
expression=f"""'{parameters[2].valueAsText}'""",
field_type='TEXT'
)
arcpy.AddMessage('Calculating the Type field...')
arcpy.management.CalculateField(output_path, 'Type',
expression="""'Collar data'""",
field_type='TEXT'
)
arcpy.AddMessage('Calculating the Source field...')
arcpy.management.CalculateField(output_path, 'Source',
expression=f"""'{collar_manufacturer} collar'""",
field_type='TEXT'
)
arcpy.AddMessage('Calculating the Entry_Staff field from the Entry_Staff parameter...')
arcpy.management.CalculateField(output_path, 'Entry_Staff',
expression=f"""'{parameters[5].valueAsText}'""",
field_type='TEXT'
)
arcpy.AddMessage('Calculating the Entry_Date field...')
arcpy.management.CalculateField(output_path, 'Entry_Date',
expression=f"""'{datetime.datetime.now().strftime('%d/%m/%Y')}'""",
field_type='DATE'
)
arcpy.AddMessage('Calculating the Visual field...')
arcpy.management.CalculateField(output_path, 'Visual',
expression="""'No'""",
field_type='TEXT'
)
if parameters[2].valueAsText == 'Wolf':
arcpy.AddMessage('Adding the Pack field from the Pack parameter...')
arcpy.management.CalculateField(output_path, 'Pack',
expression=f"""'{parameters[4].valueAsText}'""",
field_type='TEXT'
)
if parameters[3].valueAsText == 'Vectronix':
arcpy.AddMessage('Adding the Wolf_Name field from the CollarID field...')
arcpy.management.CalculateField(output_path, 'Wolf_Name',
expression=f"""'name(!CollarID!)'""",
expression_type='PYTHON 3',
code_block="""def name(collar_id):
really long if statement that didnt need to be included""",
field_type='TEXT'
)
arcpy.AddMessage('Adding the Wolf ID field from the Wolf_Name field...')
arcpy.management.CalculateField(output_path, 'Wolf_ID',
expression=f"""'id(!Wolf_Name!)'""",
expression_type='PYTHON 3',
code_block="""def id(wolf_name):
really long if statement that didnt need to be included""",
field_type='TEXT'
)
elif collar_manufacturer == 'Sirtracks':
arcpy.AddMessage('Wolf_Name and Wolf_ID are null for Sirtracks data, please add manually.')
arcpy.AddMessage('Adding the Pack field from the Pack parameter...')
arcpy.management.CalculateField(output_path, 'Wolf_Name',
expression=f'!IndividualName!',
expression_type='PYTHON 3',
field_type='TEXT'
)
# Calculate all Date Fields:
arcpy.AddMessage('Calculating the UTC field...')
if parameters[3].valueAsText == 'Vectronix':
arcpy.management.CalculateField(output_path, 'UTC',
expression='uct(!UTC_DATE!, "!UTC_TIME!")',
expression_type='PYTHON 3',
code_block="""def uct(date, time):
date_str = date.strftime('%m/%d/%Y')
datetime_str = date_str + ' ' + time
dt = datetime.datetime.strptime(datetime_str, '%m/%d/%Y %H:%M:%S')
return dt
""",
field_type='DATE'
)
elif parameters[3].valueAsText == 'Sirtrack':
arcpy.management.CalculateField(output_path, 'UTC',
expression='utc(!UTC_Date!, "!UTC_Time!")',
expression_type='PYTHON 3',
code_block="""def utc(date, time):
date_str = date.strftime('%m/%d/%Y')
datetime_str = date_str + ' ' + time
dt = datetime.datetime.strptime(datetime_str, '%m/%d/%Y %H:%M:%S')
return dt
""",
field_type='DATE'
)
elif parameters[3].valueAsText == 'Telonics':
arcpy.management.CalculateField(output_path, 'UTC',
expression='utc(!ObservationDateTimePST!)',
expression_type='PYTHON 3',
code_block="""def utc(pst):
import pytz
pst.replace(tzinfo=pytz.timezone('America/Los_Angeles'))
utc = pst.astimezone(pytz.timezone('Europe/London'))
return utc
""",
field_type='DATE'
)
elif parameters[3].valueAsText == 'Lotek':
arcpy.management.CalculateField(output_path, 'UTC',
expression='!Date___Tim!',
expression_type='PYTHON 3',
code_block="""""",
field_type='DATE'
)
arcpy.AddMessage('Calculating the PST and PDT fields from the UTC field...')
arcpy.management.ConvertTimeZone(output_path, 'UTC', 'UTC', 'PST', 'Pacific_Standard_Time', False, False)
arcpy.management.ConvertTimeZone(output_path, 'UTC', 'UTC', 'PDT', 'Pacific_Standard_Time', False, True)
arcpy.AddMessage('Calculating the Start_Time field...')
arcpy.management.CalculateField(output_path, 'Start_Time',
expression='!PST!',
field_type='DATE'
)
arcpy.AddMessage('Calculating the End_Time field...')
arcpy.management.AddField(output_path, 'End_Time', 'DATE')
arcpy.management.CalculateEndTime(output_path, 'Start_Time', 'End_Time')
arcpy.AddMessage('Calculating the TimeInt field...')
arcpy.management.CalculateField(output_path, 'TimeInt',
expression='time_diff(!Start_Time!, !End_Time!)',
expression_type='PYTHON 3',
code_block="""def time_diff(start, end):
diff = end - start
return diff.seconds//60
""",
field_type='LONG'
)
# Switch output to target now that the ultimate feature class has been created.
target_features = output_path
# Spatially join administrative boundaries: WMU, County, and Ownership
if parameters[6].value and parameters[2].valueAsText == 'Wolf':
ownership_input = r'some network path'
ownership_ouput = r'memory\OwnershipIntermediate'
wmu_input = r'some network path'
wmu_output = r'memory\WMUIntermediate'
county_input = r'some network path'
county_output = r'memory\CountiesIntermediate'
arcpy.AddMessage('Calculating Ownership...')
arcpy.analysis.SpatialJoin(target_features=target_features, join_features=ownership_input,
out_feature_class=ownership_ouput, join_operation="JOIN_ONE_TO_ONE",
join_type="KEEP_ALL", field_mapping="", match_option="INTERSECT",
search_radius="", distance_field_name="")
arcpy.management.JoinField(in_data=target_features, in_field="OBJECTID", join_table=ownership_ouput,
join_field="TARGET_FID", fields=["OWN_TYPE"])
arcpy.management.CalculateField(in_table=target_features, field="Ownership", expression='func(!OWN_TYPE!)',
expression_type="PYTHON3",
code_block="""def func(own):
if own:
return own
else:
return 'Outside Oregon'""",
field_type="TEXT",
enforce_domains="NO_ENFORCE_DOMAINS")
arcpy.AddMessage('Calculating County...')
arcpy.analysis.SpatialJoin(target_features=target_features, join_features=county_input,
out_feature_class=county_output, join_operation="JOIN_ONE_TO_ONE",
join_type="KEEP_ALL", field_mapping="", match_option="INTERSECT",
search_radius="", distance_field_name="")
arcpy.management.JoinField(in_data=target_features, in_field="OBJECTID", join_table=county_output,
join_field="TARGET_FID", fields=["Name"])
arcpy.management.CalculateField(in_table=target_features, field='County', expression='func(!Name!)',
expression_type='PYTHON3',
code_block="""def func(county):
if county:
return county
else:
return 'Outside Oregon'""",
field_type='TEXT',
enforce_domains='NO_ENFORCE_DOMAINS')
arcpy.AddMessage('Calculating WMU...')
arcpy.analysis.SpatialJoin(target_features=target_features, join_features=wmu_input,
out_feature_class=wmu_output, join_operation='JOIN_ONE_TO_ONE',
join_type='KEEP_ALL', field_mapping='', match_option='INTERSECT',
search_radius='', distance_field_name='')
arcpy.management.JoinField(in_data=target_features, in_field='OBJECTID', join_table=wmu_output,
join_field='TARGET_FID', fields=['UNIT_NAME'])
arcpy.management.CalculateField(in_table=target_features, field='WMU', expression='func(!UNIT_NAME!)',
expression_type='PYTHON3',
code_block="""def func(wmu):
if wmu:
return wmu
else:
return 'Outside Oregon'""",
field_type='TEXT',
enforce_domains='NO_ENFORCE_DOMAINS')
arcpy.management.DeleteField(target_features, 'OWN_TYPE')
arcpy.management.DeleteField(target_features, 'NAME')
arcpy.management.DeleteField(target_features, 'UNIT_NAME')
if parameters[7].value:
# Adding Status field based on input from the Status Table parameter.
arcpy.AddMessage('Calculating Status field from the Status Table parameter...')
fields_list = arcpy.ListFields(target_features)
for field in fields_list:
if 'Status' not in field.name:
arcpy.management.AddField(target_features, "Status", "TEXT")
st = parameters[7].valueAsText
arcpy.AddMessage(st)
st_split = st.split(";")
fields = ["Status", "Start_Time"]
for entry in st_split:
entry_split = entry.split(" ")
status = entry_split[0]
start_time = entry_split[1]
end_time = entry_split[2]
sql = f"Start_Time >= date '{start_time}' AND Start_Time <= date '{end_time}'"
with arcpy.da.UpdateCursor(target_features, fields, sql) as cursor:
for row in cursor:
row[0] = status
cursor.updateRow(row)
else:
arcpy.AddMessage('Add Administrative Boundaries not selected...')
# Delete unused fields:
arcpy.AddMessage('Deleting unused fields...')
for field in original_fields:
if not field.required:
arcpy.management.DeleteField(output_path, field.name)
def execute(self, parameters, messages):
# """The source code of the tool."""
input_format = parameters[0].valueAsText.split('.')[-1].lower()
collar_manufacturer = parameters[3].valueAsText
scratch = arcpy.env.scratchFolder
if input_format in ['xlsx', 'xls', 'csv', 'dbf', 'txt']:
arcpy.AddMessage(f'File detected as {input_format}, Converting {input_format} to Table')
input_path = os.path.join(parameters[0].valueAsText)
if input_format == 'dbf':
dbf = dbfread.DBF(input_path)
df = pd.DataFrame(iter(dbf))
table_for_xy = os.path.join(parameters[0].valueAsText)
elif input_format == 'csv':
excel_path = os.path.join(parameters[0].valueAsText[:-3] + 'xlsx')
df = pd.read_csv(input_path)
df = df.rename(columns={df.columns[0]: 'Device_ID'})
df.to_excel(excel_path, index=False, columns=['Device_ID', 'Date & Time [GMT]', 'Latitude', 'Longitude',
'Altitude', 'DOP'])
table_for_xy = arcpy.conversion.ExcelToTable(excel_path, scratch)
elif input_format == 'txt':
df = pd.read_table(input_path)
table_for_xy = arcpy.conversion.TableToTable(input_path, scratch, 'TempTxtTable')
elif input_format == 'xlsx' or input_format == 'xls':
df = pd.read_excel(input_path)
table_for_xy = arcpy.conversion.ExcelToTable(input_path, scratch)
else:
raise Exception('Incorrect format of input file.')
# Build list of all unique Collar ID values that will be used to make a selection so all the data is
# processed at the same time
if collar_manufacturer == 'Vectronix' or collar_manufacturer == 'Telonics':
collar_list = df.CollarID.unique()
elif collar_manufacturer == 'Sirtrack':
collar_list = df.Tag_ID.unique()
elif collar_manufacturer == 'Lotek':
collar_list = df['Device_ID'].unique()
else:
raise Exception('Collar Manufacturer doesnt match.')
if len(collar_list) != 0:
counter = 0
for collar_id in collar_list:
counter = counter + 1
arcpy.AddMessage(f'Processing {collar_id}, {counter} of {len(collar_list)}...')
if collar_manufacturer == 'Vectronix':
subframe = df[df[f'CollarID'] == collar_id]
min_date = subframe.UTC_DATE.min().strftime('%y%m%d')
max_date = subframe.UTC_DATE.max().strftime('%y%m%d')
output_name = f'{collar_manufacturer[0].upper()}_{collar_id}_{min_date}_{max_date}'
selection = arcpy.management.SelectLayerByAttribute(table_for_xy, 'NEW_SELECTION',
f"""CollarID = {collar_id}""")
elif collar_manufacturer == 'Sirtrack':
df['UTC'] = df[['UTC_Date', 'UTC_Time']].agg(' '.join, axis=1)
df['UTC'] = pd.to_datetime(df['UTC'], format='%m/%d/%Y %H:%M:%S')
subframe = df[df[f'Tag_ID'] == collar_id]
min_date = subframe.UTC.min().strftime('%y%m%d')
max_date = subframe.UTC.max().strftime('%y%m%d')
output_name = f'{collar_manufacturer[0].upper()}_{collar_id}_{min_date}_{max_date}'
selection = arcpy.management.SelectLayerByAttribute(table_for_xy, 'NEW_SELECTION', f"""Tag_ID = '{collar_id} And Latitude <> 0 And Longitude <> 0'""")
elif collar_manufacturer == 'Telonics':
subframe = df[df[f'CollarID'] == collar_id]
min_date = subframe.ObservationDateTimePST.min().strftime('%y%m%d')
max_date = subframe.ObservationDateTimePST.max().strftime('%y%m%d')
output_name = f'{collar_manufacturer[0].upper()}_{collar_id}_{min_date}_{max_date}'
selection = arcpy.management.SelectLayerByAttribute(table_for_xy, 'NEW_SELECTION', f"""Tag_ID = '{collar_id}' And Latitude <> 0 And Longitude <> 0""")
elif collar_manufacturer == 'Lotek':
if 'UTC' not in df.columns:
df['UTC'] = pd.to_datetime(df['Date & Time [GMT]'], format='%Y-%m-%d %H:%M:%S')
subframe = df[df['Device_ID'] == collar_id]
min_date = subframe.UTC.min().strftime('%y%m%d')
max_date = subframe.UTC.max().strftime('%y%m%d')
output_name = f'{collar_manufacturer[0].upper()}_{collar_id}_{min_date}_{max_date}'
selection = arcpy.management.SelectLayerByAttribute(table_for_xy, 'NEW_SELECTION', f"""Device_ID = {collar_id} And Latitude <> 0 And Longitude <> 0""")
selection = arcpy.conversion.TableToTable(selection, 'memory', output_name)
else:
raise Exception('Collar Manufacturer doesnt match.')
self.xy_to_fc(parameters=parameters, table_for_xy=selection, output_name=output_name,
input_format=input_format)
else:
raise Exception('No Collar ID in Table.')
else:
arcpy.AddMessage(
f'File detected as {input_format}, this format is unsupported. Please use dbf, xlsx or xls')
raise Exception('Invalid input table format!')
arcpy.AddMessage('Data processing is complete!')
return