Created
April 7, 2022 18:07
-
-
Save reyemtm/35fc7b0f4afa623a28a341e7cb1d500b to your computer and use it in GitHub Desktop.
Esri syncFeatureWithTable.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # --------------------------------------------------------------------------- | |
| # MuniLink Utilities > GIS sync | |
| # --------------------------------------------------------------------------- | |
| # Created by Malcolm Meyer 03/29/2022 | |
| # Description: | |
| # This script copies view tables from the MuniLink database to the GIS database | |
| # --or-- it may create dictionaries instead | |
| # Then these GIS Database tables (or dictionaries) are used to sync data to various GIS layers | |
| # Which then syncs data over to Cartegraph in the morning via AGOL <> Cartegraph sync | |
| # --------------------------------------------------------------------------- | |
| # --------------------------------------------------------------------------- | |
| # Feature Services/GIS Layers | |
| # --------------------------------------------------------------------------- | |
| # Gas Customers - Gas.Gas_Service_Locations | |
| # Services (Sanitation) - Utilities.All_utils | |
| # (the one above is redundant and should somehow be merged with the code locations) | |
| # New Customer Data (Code) - Utilities.code_customer_locations | |
| # --------------------------------------------------------------------------- | |
| # --------------------------------------------------------------------------- | |
| # MuniLink MySQL Connection - saved from ArcCatalog | |
| # --------------------------------------------------------------------------- | |
| # username: lancasteroh | |
| # password: | |
| # HOST : | |
| # IP: | |
| # Port: 3306 - NOT NEEDED | |
| # DB: | |
| # Allowed: Whitelisted IPs: | |
| # --------------------------------------------------------------------------- | |
| # --------------------------------------------------------------------------- | |
| # Notes: | |
| # | |
| # I had issues calculating fields for all 20k records in each dataset using ArcMap | |
| # I used the geoprocessing tool Calculate Field in Pro which writes directly | |
| # to the dataset vs doing an edit session and this seems to work | |
| # --------------------------------------------------------------------------- | |
| # addresstype = location_type from utilities NOT customer_class | |
| import arcpy | |
| import os | |
| from time import sleep | |
| from lib_SyncTableToFeature import sync, compare | |
| import logging | |
| import logging.handlers as handlers | |
| import sys | |
| from os import path | |
| # --------------------------------------------------------------------------- | |
| # Logging and Email Alerts | |
| # --------------------------------------------------------------------------- | |
| logfile = 'munilink-sync-log.csv' | |
| # Check if log exists and should therefore be rolled | |
| logBackup = path.isfile(logfile) | |
| # create logger | |
| lgr = logging.getLogger('log') | |
| lgr.setLevel(logging.INFO) | |
| csv = logging.Formatter('%(asctime)s,%(levelname)s,%(message)s') | |
| logRotater = handlers.RotatingFileHandler(logfile,backupCount=14) | |
| logRotater.setFormatter(csv) | |
| lgr.addHandler(logRotater) | |
| # This is a stale log, so roll it | |
| if logBackup: | |
| # Add timestamp | |
| lgr.info('Log closed') | |
| # Roll over on application start | |
| lgr.handlers[0].doRollover() | |
| consoleHandler = logging.StreamHandler(sys.stdout) | |
| consoleHandler.setLevel(logging.DEBUG) | |
| consoleHandler.setFormatter(csv) | |
| lgr.addHandler(consoleHandler) | |
| smtp_handler = handlers.SMTPHandler(mailhost='', | |
| fromaddr='', | |
| toaddrs=[''], | |
| subject='Munilink Sync Error', | |
| # credentials=('user','pwd'), | |
| secure=None) | |
| smtp_handler.setLevel(logging.ERROR) | |
| smtp_handler.setFormatter(csv) | |
| lgr.addHandler(smtp_handler) | |
| lgr.info('sync process started') | |
| # --------------------------------------------------------------------------- | |
| # Script switches - whether or not to copy over tables and sync, or just compare with existing info | |
| # --------------------------------------------------------------------------- | |
| SYNC = True | |
| COPY = True | |
| lgr.info("sync mode: " + str(SYNC)) | |
| # --------------------------------------------------------------------------- | |
| # Pull data from MuniLink to GIS_TEST - for now using dictionaries instead | |
| # --------------------------------------------------------------------------- | |
| if COPY: | |
| arcpy.env.workspace = "" | |
| outWorkspace = ".sde\\" | |
| tables = arcpy.ListTables() | |
| try: | |
| for table in tables: | |
| # print(table) | |
| lgr.info('munilink table: ' + table) | |
| dest = outWorkspace + "ml_" + table | |
| if arcpy.Exists(dest): | |
| lgr.info("exists | deleting: " + dest) | |
| arcpy.Delete_management(dest) | |
| lgr.info("deleted | creating: " + dest) | |
| arcpy.CopyRows_management(table, dest) | |
| else: | |
| lgr.info("table doesn't exist, creating new table") | |
| arcpy.CopyRows_management(table, dest) | |
| except Exception as ex: | |
| lgr.warn(ex) | |
| exit() | |
| # --------------------------------------------------------------------------- | |
| # CREATE MUNILINK SYNC DICTIONARY FOR OWNER (currently missing) AND RESIDENT INFORMATION | |
| # --------------------------------------------------------------------------- | |
| # munilinkTable = "\\\\192.168.168.22\\gis_admin\\db-connections\\MuniLinkDB.odc\\export_accounts_final" # network connected | |
| munilinkTable = "ml_export_accounts_final" # cached local | |
| syncDict = {} | |
| # --------------------------------------------------------------------------- | |
| # CONCATENTATION FOR ADDRESS WITH POSSIBLE NULL FIELDS | |
| # https://gis.stackexchange.com/questions/291706/concatenating-address-fields-with-possible-blank-or-null-values-in-fields-using | |
| # --------------------------------------------------------------------------- | |
| def ConcatStrings(*args): | |
| return ' '.join([str(i) for i in args if i not in(None,' ')]) | |
| with arcpy.da.SearchCursor(munilinkTable,[ | |
| 'accountnumber','parcelnumber','customername', 'phone', 'owprimaryphone', 'housenumber', 'streetname', 'address2', 'addresstype', 'city', 'state', 'zip', 'utilities', 'accountstatus' | |
| ]) as munilinkCursor: | |
| for (i,row) in enumerate(munilinkCursor): | |
| address = "missing service address" | |
| gas = False | |
| sewer = False | |
| water = False | |
| sanitation = False | |
| stormwater = False | |
| u = row[12] | |
| if u and 'GAS' in u: | |
| gas = True | |
| if u and 'SANITATION' in u: | |
| sanitation = True | |
| if u and 'SEWER' in u: | |
| sewer = True | |
| if u and 'STORMWATER' in u: | |
| stormwater = True | |
| if u and 'WATER/R' in u: | |
| water = True | |
| if (row[5] or row[6] or row[7]): | |
| address = ConcatStrings(row[5], row[6], row[7]) | |
| syncDict[row[0]] = { | |
| 'munilink_id': row[0], | |
| 'pin': row[1], | |
| 'resident_full_name': row[2][:50], | |
| 'resident_phone': row[3], | |
| 'owner_phone': row[4], | |
| 'address': address[:255].strip(" "), | |
| 'service_location50': address[:50].strip(" "), | |
| 'service_location40': address[:40].strip(" "), | |
| 'location_type': row[8], | |
| 'housenumber': row[5], | |
| 'city': row[9], | |
| 'streetname': row[6], | |
| 'state': row[10], | |
| 'zip': row[11], | |
| 'gas': gas, | |
| 'sewer': sewer, | |
| 'water': water, | |
| 'sanitation': sanitation, | |
| 'stormwater': stormwater, | |
| 'accountstatus': row[13] | |
| } | |
| lgr.info('munilink sync dict created - total records: ' + str(len(syncDict))) | |
| if len(syncDict) < 19000: | |
| lgr.info("current length of munilink table is too small, quitting!") | |
| exit() | |
| del munilinkCursor | |
| # --------------------------------------------------------------------------- | |
| # SET NEW WORKSPACE FOR THE REST OF THE SCRIPT - THE CONNECTION USES WINDOWS AUTH | |
| # ON THE SERVER IT USES THE ADMINISTRATOR USER LOGIN | |
| # --------------------------------------------------------------------------- | |
| admin_workspace = "\\\\192.168.168.22\\gis_admin\\db-connections\\GIS_TEST.sde" | |
| # --------------------------------------------------------------------------- | |
| # SYNC code_customer_locations | |
| # --------------------------------------------------------------------------- | |
| currentFeature = "code_customer_locations" | |
| currentFieldNames = [ | |
| 'munilink_id', 'ParcelNumber', 'City', 'State', 'PostalCode', 'address', 'location_type', 'resident_full_name', 'resident_phone','owner_full_name', 'owner_phone'] | |
| aliases = [None, 'pin', 'city', 'state', 'zip'] | |
| try: | |
| lgr.info("starting sync/compare on " + currentFeature) | |
| lgr.info("using fields " + (','.join(currentFieldNames))) | |
| if SYNC: | |
| updated = sync(admin_workspace, syncDict, currentFeature, currentFieldNames, aliases, -1) | |
| else: | |
| updated = compare(admin_workspace, syncDict, currentFeature, currentFieldNames, aliases, -1) | |
| lgr.info("completed " + currentFeature) | |
| lgr.info("rows updated or out of sync: " + str(updated)) | |
| except Exception as ex: | |
| lgr.warn(ex) | |
| sleep(5) | |
| # --------------------------------------------------------------------------- | |
| # SYNC All_utils | |
| # --------------------------------------------------------------------------- | |
| currentFeature = "All_utils" | |
| currentFieldNames = ['munilink_id', 'PIN', 'CustomerHomePhone','OwnerHomePhone','LocationType','State','ServiceLocation', 'address'] | |
| aliases = [None, 'pin', 'resident_phone', 'owner_phone', 'location_type','state', 'service_location50'] | |
| try: | |
| lgr.info("starting sync/compare on " + currentFeature) | |
| lgr.info("using fields " + (','.join(currentFieldNames))) | |
| if SYNC: | |
| updated = sync(admin_workspace, syncDict, currentFeature, currentFieldNames, aliases, -1) | |
| else: | |
| updated = compare(admin_workspace, syncDict, currentFeature, currentFieldNames, aliases, -1) | |
| lgr.info("completed " + currentFeature) | |
| lgr.info("rows updated or out of sync: " + str(updated)) | |
| except Exception as ex: | |
| lgr.warn(ex) | |
| # sleep(5) | |
| # --------------------------------------------------------------------------- | |
| # Sync Gas | |
| # --------------------------------------------------------------------------- | |
| currentFeature = "Gas_Service_Locations" | |
| # TODO in the meters sync yet to be created add readingtype - metertype | |
| currentFieldNames = ['munilink_id', 'PIN', 'resident_name', 'ServiceLocation','location_type', 'resident_phone', 'owner_phone', 'address', 'accountstatus'] | |
| aliases = [None, 'pin', 'resident_full_name','service_location40'] | |
| try: | |
| lgr.info("starting sync/compare on " + currentFeature) | |
| lgr.info("using fields " + (','.join(currentFieldNames))) | |
| if SYNC: | |
| updated = sync(admin_workspace, syncDict, currentFeature, currentFieldNames, aliases, -1) | |
| else: | |
| updated = compare(admin_workspace, syncDict, currentFeature, currentFieldNames, aliases, -1) | |
| lgr.info("completed " + currentFeature) | |
| lgr.info("rows updated or out of sync: " + str(updated)) | |
| except Exception as ex: | |
| lgr.warn(ex) | |
| # --------------------------------------------------------------------------- | |
| # THIS FUNCTIONS UPDATES FEATURE VALUES TO A DICTIONARY USING THE FIRST FIELD IN THE FIELD NAMES AS THE JOIN FIELD | |
| # --------------------------------------------------------------------------- | |
| import arcpy | |
| def sync(admin_workspace, syncDict, inFeature, fieldNames, aliases, nFeatures): | |
| updatedRows = 0 | |
| nRows = 0 | |
| if nFeatures < 0: | |
| nRows = 1000000000 | |
| else: | |
| nRows = nFeatures | |
| arcpy.env.workspace = admin_workspace | |
| arcpy.env.autoCommit = 1000000 | |
| print("current workspace: " + arcpy.env.workspace) | |
| try: | |
| edit = arcpy.da.Editor(admin_workspace) | |
| edit.startEditing(False, True) # no undo, no multi-user | |
| edit.startOperation() #- possible not needed when using False above in startEditing? - see | |
| # https://www.reddit.com/r/gis/comments/ausimg/workspace_already_in_transaction_mode/ | |
| # then again maybe this is needed | |
| # https://gis.stackexchange.com/questions/181400/arcpy-da-updatecursor-error-the-requested-operation-is-invalid-on-a-closed-state/181433 | |
| # make sure all the data is versioned | |
| featureUpdated = False | |
| featuresCursor = arcpy.da.UpdateCursor(inFeature, fieldNames) | |
| for (i, row) in enumerate(featuresCursor): | |
| updated = False | |
| if i < nRows: | |
| if (row[0] and (row[0] in syncDict)): | |
| data = {} | |
| muniLink = syncDict[row[0]] | |
| for (valueIndex, value) in enumerate(row): | |
| fieldName = fieldNames[valueIndex] | |
| if valueIndex < len(aliases) and aliases[valueIndex]: | |
| fieldName = aliases[valueIndex] | |
| # print fieldName | |
| # if fieldName == 'ParcelNumber': | |
| # fieldName = 'pin' | |
| if fieldName in muniLink and muniLink[fieldName] and value != muniLink[fieldName]: | |
| updated = True | |
| print fieldNames[valueIndex] + ": " + str(value) + " out of sync" | |
| print (fieldNames[valueIndex] + ": " + str(muniLink[fieldName])) | |
| row[valueIndex] = muniLink[fieldName] | |
| # else: | |
| # print "skipping " + fieldName | |
| else: | |
| if row[0]: | |
| print"skipping munilink_id: " + row[0] | |
| else: | |
| print "skipping: NULL" | |
| if updated: | |
| print("updated " + str(row[0])) | |
| featuresCursor.updateRow(row) | |
| featureUpdated = True | |
| updatedRows+=1 | |
| try: | |
| print("updated rows: " + str(updatedRows)) | |
| edit.stopOperation() #- not needed to to using false in startEditing? | |
| if featureUpdated: | |
| edit.stopEditing(True) | |
| del featuresCursor | |
| print("saved with edits!") | |
| else: | |
| edit.stopEditing(False) | |
| del featuresCursor | |
| print "nothing to commit" | |
| # lgr.info("Saved") | |
| except Exception as err: | |
| edit.stopEditing(False) | |
| del featuresCursor | |
| # lgr.critical(err) | |
| print(err) | |
| finally: | |
| arcpy.ClearWorkspaceCache_management() | |
| except Exception as err: | |
| print err | |
| return updatedRows | |
| # --------------------------------------------------------------------------- | |
| # --------------------------------------------------------------------------- | |
| def compare(admin_workspace, syncDict, inFeature, fieldNames, aliases, nFeatures): | |
| updatedRows = 0 | |
| nRows = 0 | |
| if nFeatures < 0: | |
| nRows = 1000000000 | |
| else: | |
| nRows = nFeatures | |
| arcpy.env.workspace = admin_workspace | |
| print("current workspace: " + arcpy.env.workspace) | |
| try: | |
| featureUpdated = False | |
| featuresCursor = arcpy.da.SearchCursor(inFeature, fieldNames) | |
| for (i, row) in enumerate(featuresCursor): | |
| updated = False | |
| if i < nRows: | |
| if (row[0] and (row[0] in syncDict)): | |
| data = {} | |
| muniLink = syncDict[row[0]] | |
| for (valueIndex, value) in enumerate(row): | |
| fieldName = fieldNames[valueIndex] | |
| if valueIndex < len(aliases) and aliases[valueIndex]: | |
| fieldName = aliases[valueIndex] | |
| # print fieldName | |
| # if fieldName == 'ParcelNumber': | |
| # fieldName = 'pin' | |
| if fieldName in muniLink and muniLink[fieldName] and value != muniLink[fieldName]: | |
| updated = True | |
| print fieldNames[valueIndex] + ": " + str(value) + " out of sync" | |
| print (fieldNames[valueIndex] + ": " + str(muniLink[fieldName])) | |
| # row[valueIndex] = muniLink[fieldName] | |
| # else: | |
| # print "skipping " + fieldName | |
| else: | |
| if row[0]: | |
| print"skipping munilink_id: " + row[0] | |
| else: | |
| print "skipping: NULL" | |
| if updated: | |
| print("rows out of sync " + str(row[0])) | |
| # featuresCursor.updateRow(row) | |
| updatedRows+=1 | |
| featureUpdated = True | |
| try: | |
| print("rows out of sync: " + str(updatedRows)) | |
| # edit.stopOperation() #- not needed to to using false in startEditing? | |
| if featureUpdated: | |
| # edit.stopEditing(True) | |
| del featuresCursor | |
| print("compare successfull!") | |
| else: | |
| # edit.stopEditing(False) | |
| del featuresCursor | |
| print "nothing to say" | |
| # lgr.info("Saved") | |
| except Exception as err: | |
| # edit.stopEditing(False) | |
| del featuresCursor | |
| # lgr.critical(err) | |
| print(err) | |
| finally: | |
| arcpy.ClearWorkspaceCache_management() | |
| except Exception as err: | |
| print err | |
| return updatedRows | |
| # --------------------------------------------------------------------------- | |
| # --------------------------------------------------------------------------- |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment