UPDATE: I made a version of this tool that DOES NOT use any libraries not included in the default ArcGIS Pro interpreter, however the problems mentioned in the previous post persist.
Only one of the reports is available to be generated in this example, FYI.
# -*- coding: utf-8 -*-
from uuid import uuid4
import arcpy
import os
import pandas
cwd = os.path.dirname(os.path.realpath(__file__))
class Toolbox:
def __init__(self):
"""Define the toolbox (the name of the toolbox is the name of the
.pyt file)."""
self.label = "Report Generator Toolbox"
self.alias = "report_generator_toolbox"
# List of tool classes associated with this toolbox
self.tools = [ReportGeneratorTool]
class ReportGeneratorTool:
def __init__(self):
"""Define the tool (tool name is the name of the class)."""
self.label = "Report Generator Tool DEMO"
self.description = "DEMO: Herramienta para la generación de reporte(s) HITS a Armas"
self.area_values = {"Bayamon": "Bayamon", "Guayama": "Guayama", "Humacao": "Humacao"}
self.municipality_values = {"Sanjuan": "SANJUAN", "Carolina": "CAROLINA", "Guaynabo": "GUAYNABO"}
self.pol_unit_values = {"Investigaciones De Licencias De Armas": "INVESTIGACIONES DE LICENCIAS DE ARMAS", "Policia Municipal": "POLICIA MUNICIPAL", "Violencia Domestica": "VIOLENCIA DOMESTICA"}
def getParameterInfo(self):
"""Define the tool parameters."""
# Report Type
report_type = arcpy.Parameter(displayName="Tipo de Reporte", name="report_type", datatype="String",
parameterType="Required", direction="Input")
report_type.filter.type = "ValueList"
report_type.filter.list = ['Relaciones de Eventos']
# Police Area
area_param = arcpy.Parameter(displayName="Area Policiaca", name="police_area", datatype="GPString",
parameterType="Optional", direction="Input")
area_param.filter.type = "ValueList"
area_param.filter.list = list(self.area_values.keys()) if len(self.area_values.keys()) > 0 else []
# Municipality
municipality_param = arcpy.Parameter(displayName="Municipio", name="municipality", datatype="GPString",
parameterType="Optional", direction="Input")
municipality_param.filter.type = "ValueList"
municipality_param.filter.list = list(self.municipality_values.keys()) if len(self.municipality_values.keys()) > 0 else []
# Police Unit Field
police_unit_param = arcpy.Parameter(displayName="Unidad", name="unit", datatype="GPString",
parameterType="Optional", direction="Input")
police_unit_param.filter.type = "ValueList"
police_unit_param.filter.list = list(self.pol_unit_values.keys()) if len(self.pol_unit_values.keys()) > 0 else []
# Date of Occurrence
date_occurrence_param = arcpy.Parameter(
displayName="Fecha de Ocurrencia\n(Solo escojer hasta UN(1) solo rango de Fechas)\n", name="date_range",
datatype="GPValueTable",
parameterType="Optional", direction="Input")
date_occurrence_param.columns = [['Date', 'Desde'], ['Date', 'Hasta']]
# Output Parameter
output = arcpy.Parameter(displayName="Output", name="output", datatype="DEFile",
parameterType="Derived", direction="Output")
return [report_type, area_param, municipality_param, police_unit_param, date_occurrence_param, output]
def isLicensed(self):
"""Set whether the 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."""
# Remove extra date ranges from date_range_param if entered by the user
date_idx_param = 4
date_range_param = parameters[date_idx_param].value
if (date_range_param and len(date_range_param) > 1):
parameters[date_idx_param].value = [date_range_param[0]]
return
def updateMessages(self, parameters):
"""Modify the messages created by internal validation for each tool
parameter. This method is called after internal validation."""
# Notify user only one pair of dates can be specified for the date range parameter
date_idx_param = 4
date_range_param = parameters[date_idx_param].value
if (date_range_param and len(date_range_param) > 1):
parameters[date_idx_param].setErrorMessage("Solo puede escojer hasta un solo set de fechas (Desde / Hasta)")
# Validate that date range entered is valid
if (date_range_param and len(date_range_param) == 1 and len(date_range_param[0]) == 2):
start_date, end_date = date_range_param[0]
if (start_date > end_date):
parameters[date_idx_param].setErrorMessage(
"Fecha de Inicio `Desde` debe ser menor que Fecha de Culminación `Hasta`.")
return
def execute(self, parameters, messages):
"""The source code of the tool."""
# Define the WHERE Clause that will be used for generating reports
where_clause = ""
where_holder = []
# Hacky way to obtain a chain of ANDS for the where_clause expression
if (parameters[1].value is not None):
where_holder.append(f"AreaGuns = '{self.area_values[parameters[1].value]}'")
if (parameters[2].value is not None):
where_holder.append(f"MunicipioGuns = '{self.municipality_values[parameters[2].value]}'")
if (parameters[3].value is not None):
where_holder.append(f"UnidadArma = '{self.pol_unit_values[parameters[3].value]}'")
if(parameters[4].value is not None and len(parameters[4].value[0]) == 2):
start_date, end_date = parameters[4].value[0]
date_clause = f"FechaArma >= TIMESTAMP'{start_date}' AND FechaArma <= TIMESTAMP'{end_date}'"
where_holder.append(date_clause)
del start_date, end_date, date_clause
# NOTE: Include in where clause to ignore records where num_caso_query is NULL
# NOTE: If this where clause changes, make sure to also change it when evaluating where clause
# to be used for relation
if (len(where_holder) > 0):
where_clause = " AND ".join(where_holder)
where_clause += " AND num_caso_query IS NOT NULL"
if (where_clause == ""):
where_clause = "num_caso_query IS NOT NULL"
out_file = self.report_generator(where_clause, parameters[0].value)
if(out_file):
parameters[5].value = out_file
return
def postExecute(self, parameters):
"""This method takes place after outputs are processed and
added to the display."""
return
def _find_case_relations(self, weapon_data, hits_data, global_weapon_data):
arcpy.AddMessage("Identificando Relaciones para HITS")
relationship_object = {} # Object to keep track of relationships identified
case_no_evaluated = []
hits_data_keys = list(hits_data.keys()) # Calculate Hits_Data keys ONCE to avoid same calculation in all iterations.
for case_no in weapon_data.keys():
relationship_res, cases_found_rel = self._find_case_no_relations_recursive(case_no, global_weapon_data,
hits_data, hits_data_keys, case_no_evaluated)
if relationship_res is not None:
relationship_object[case_no] = relationship_res[case_no]
case_no_evaluated.append(case_no)
case_no_evaluated += cases_found_rel
return relationship_object
def _find_case_no_relations_recursive(self, case_no, weapon_data, hits_data, hits_data_keys, case_no_already_checked=[]):
found_relations = {}
# Don't perform operation on records not considered in the weapon_data dictionary
if(case_no not in weapon_data.keys()):
return None, []
# Don't perform operation on records where there is not at least one relationship on the hits layer.
case_no_relations = [x for x in weapon_data[case_no]]
count_relations = list(filter(lambda x: x in [x['globalid'] for x in case_no_relations], hits_data_keys))
if(len(count_relations) == 0):
return None, []
found_relations[case_no] = {}
for hit_rel_parentglobalid in count_relations:
for current_case_no_rel in hits_data[hit_rel_parentglobalid]:
if current_case_no_rel in case_no_already_checked:
found_relations[case_no] = current_case_no_rel
return found_relations, case_no_already_checked
# Recursive call to this function to find all possible matches hierarchically
curr_relation_match, curr_records_checked = self._find_case_no_relations_recursive(current_case_no_rel,
weapon_data, hits_data, hits_data_keys, case_no_already_checked)
if(curr_relation_match is not None):
if(isinstance(found_relations[case_no], dict)):
found_relations[case_no][current_case_no_rel] = curr_relation_match[current_case_no_rel]
else:
# Current logic avoids case number being duplicated, ex {'2024-05-2154': '2024-05-2154': {...}}
# Fixed by doing `curr_relation_match[current_case_no_rel]`
found_relations[case_no] = {current_case_no_rel: curr_relation_match[current_case_no_rel]}
else:
found_relations[case_no][current_case_no_rel] = None
case_no_already_checked += curr_records_checked
case_no_already_checked.append(current_case_no_rel)
case_no_already_checked.append(case_no)
return found_relations, list(set(case_no_already_checked))
def _recursive_dataframe_generator(self, rel_dictionary, parent_key=None):
dataframe_collection = []
counter = 0
for key, value in rel_dictionary.items():
if(not parent_key):
counter += 1
# if (key =='2024:07-171-000637'):
# print('break here!!!!')
if(isinstance(value, str)):
df = pandas.DataFrame(data=[key, value])
df = df.transpose()
if not parent_key:
df = pandas.concat([pandas.DataFrame(data=[counter]), df], axis=1, ignore_index=True)
dataframe_collection.append(df)
elif(value is None):
df = pandas.DataFrame(data=[key])
if not parent_key:
print("`Me: This will never happen lol! - Happens anyway`")
dataframe_collection.append(df)
elif(isinstance(value, dict)):
result = self._recursive_dataframe_generator(value, key)
it_count = 0
for res in result:
if(it_count == 0):
df = pandas.DataFrame(data=[key])
else:
df = pandas.DataFrame(data=[''])
df = pandas.concat([df, res], axis=1, ignore_index=True)
# Add record numbering when required
if not parent_key:
append_cell = counter if it_count == 0 else ''
df = pandas.concat([pandas.DataFrame(data=[append_cell]), df], ignore_index=True, axis=1)
it_count += 1
dataframe_collection.append(df)
# Add "line break" row between records when necessary
if not parent_key:
top_df_width = 0
for df in dataframe_collection:
width = len(df.columns)
top_df_width = width if width > top_df_width else top_df_width
data = ['' for _ in range(0, top_df_width)]
dataframe_collection.append(pandas.DataFrame(data=data).transpose())
return dataframe_collection
def _report_dataframe_generator(self, relationship_object):
# Define a list of dictionary keys, in order to add a new line in the DataFrame after all records for this key has been processed.
object_record_keys = list(relationship_object.keys())
result = self._recursive_dataframe_generator(relationship_object)
df_width = 0
# Obtain width of largest dataframe
for df in result:
if len(df.columns) > df_width:
df_width = len(df.columns)
result_df = None
# Concatenate all resulting dataframes to obtain a final dataframe with results
count = 0
for df in result:
# if(count == 3):
# print('e')
tmp_df = df.copy()
curr_width = len(df.columns)
if(curr_width < df_width):
# Add right padding if necessary to match length of largest dataframe
padding = ['' for x in range(0, df_width - curr_width)]
padding_df = pandas.DataFrame(data=padding)
tmp_df = pandas.concat([tmp_df, padding_df], axis=1, ignore_index=True)
if result_df is None:
result_df = tmp_df
else:
result_df = pandas.concat([result_df, tmp_df], ignore_index=True)
count += 1
return result_df
def generate_event_relationship_diagram(self, weapon_data):
sheet_name = 'Relaciones_Eventos'
target_dir = os.path.join(cwd, "Output")
out_path = os.path.join(target_dir, f"{sheet_name}_{uuid4()}.xlsx")
if(not os.path.isdir(target_dir)):
os.makedirs(target_dir)
del target_dir
# NOTE: A query is normally performed, records are processed, and this is the output of the aforementioned operations
relationship_object = {"2025:06-040-000031": {"2024:06-013:018541": None}, "2024:06-040-003889": {"2024:06-067:002316": None, "2024:06-040:005722": None, "2024:07-171-000637": "2024:06-067:002316", "2025:06-067-000239": "2024:06-040:003889"}, "2024:06-040-005722": "2024:06-040-003889", "2024:06-040-003258": {"2024:06-040-003200": None}, "2025:06-040-002387": {"2025:06-013-009991": {"2025:06-040:002387": None}}}
arcpy.AddMessage("Generando reporte de Relaciones de Eventos")
# List to store instances of DataFrame Object (One per case number evaluated)
dataframe_collection = self._report_dataframe_generator(relationship_object)
# Create PDF File from Dataframe
dataframe_collection.to_excel(out_path, sheet_name=sheet_name, index=False)
return out_path
def report_generator(self, where_clause, report_selection):
arcpy.AddMessage(f"Realizando consulta seleccionada: `{where_clause}`")
# NOTE: Normally a query is performed, the results are processed and this is the output of the aforementioned operation.
# Placeholder variables - not used in this example
case_no_to_globalids = {}
hits_records = {}
out_file = None
match report_selection:
# case 'Repeticiones de Querellas':
# out_file = self.generate_case_count_report(case_no_to_globalids, hits_records)
case 'Relaciones de Eventos':
out_file = self.generate_event_relationship_diagram(case_no_to_globalids)
# case 'Querellas Relacionadas Según NIBIN Lead':
# out_file = self.generate_related_case_report(case_no_to_globalids)
case _:
arcpy.AddMessage("Selección de reporte inválida")
arcpy.AddMessage("Reporte Generado Exitosamente")
return out_file
# # Test Script - Uncomment to Run script using python interpreter (terminal).
# if __name__ == "__main__":
# cls = ReportGeneratorTool()
# # NOTE: When testing, the first two parameter is not really important, Just make sure to select one of the cases avobe for the 2nd parameter.
# cls.report_generator("", "Querellas Relacionadas Según NIBIN Lead")
