nav[aria-label="Primary Navigation"] { padding: 0; & ul { list-style: none; width: 100%; display: flex; flex-direction: row; justify-content: start; align-items: start; gap: 30px; padding: 0; & li { margin: 0; } & ul li { list-style: none; } } }

Forcing da.SearchCursor to retrieve new data?

AustinBachurski

I put together a tool that runs in the system tray and compares an attribute table column for a feature class to information pulled from a website to let me know when there's new work to be done. It works for the most part but I'm having issues with it not updating after I add new information to the table. I'm using arcpy.da.SearchCursor() in a function to retrieve the data from the table, but each time I call the function, I get the data that was retrieved on the first call, I need it to retrieve the latest data. I've tried calling the .reset() method on the object it returns, deleting the object, and calling arcpy.ResetEnvironments to no avail. The only success I have with the tool or in a python terminal is closing the tool/terminal and reopening it. Is there a way to force SearchCursor to pull new data? Maybe there's another method that I should be using? Thanks.

def get_mapped_permits():
    building_permits_2022 = "path.to.sde"
    mapped = [each for permit in arcpy.da.SearchCursor(building_permits_2022, "PERMIT_NUM") for each in permit]
    return mapped

Trouver plus de publications taguées avec

Accepted answers

AustinBachurski

Well after trying everything suggested and then some including manually deleting the cursor outside of a comprehension, using a context manager, moving it into separate functions, etc. etc. I finally got it to work the way I wanted it to. I eventually found out it was something to do with calling it from a while loop. I'm not 100% sure why, but if I got rid of that it worked as expected. So, what I got rid of the while loop entirely, and just have the function call itself after sleeping. From everything I've read it seems like this is an acceptable solution, and it actually works properly.

So instead of this.

def monitoring():
    while True:
        app.check_now("")
        # Check every 30 minutes.
        time.sleep(1800)

I'm using this.

def monitoring():
    """Checks for updates every 30 minutes."""
    app.check_now("auto")
    time.sleep(1800)
    threading.Thread(target=monitoring, daemon=True).start()

Which calls the searchcursor via.

def get_mapped_permits(self):
    """Returns a list of building permits found in the BuildingPermits
    feature class for the current year."""
    with arcpy.da.SearchCursor(self.feature_class, "PERMIT_NUM") as search:
        return [permit[0] for permit in search]

Thanks for all the help!

One last note on this in case it will help anyone in the future, I added another call to the function containing the SearchCursor() and started having the same problem again. Was able to fix it by changing the function to target an instance variable rather than return the value, then wrapping the function call in a daemon thread (terminology may be incorrect, noob, sry...) this produced the expected result.

def get_mapped_permits(self):
    with arcpy.da.SearchCursor(self.feature_class, "PERMIT_NUM") as search:
        self.mapped_permits = [permit[0] for permit in search]

# called from

self.mapped_permits.clear()
t = threading.Thread(target=self.get_mapped_permits, daemon=True)
t.start()

All comments

AustinBachurski

So instead of this.

def monitoring():
    while True:
        app.check_now("")
        # Check every 30 minutes.
        time.sleep(1800)

I'm using this.

def monitoring():
    """Checks for updates every 30 minutes."""
    app.check_now("auto")
    time.sleep(1800)
    threading.Thread(target=monitoring, daemon=True).start()

Which calls the searchcursor via.

def get_mapped_permits(self):
    """Returns a list of building permits found in the BuildingPermits
    feature class for the current year."""
    with arcpy.da.SearchCursor(self.feature_class, "PERMIT_NUM") as search:
        return [permit[0] for permit in search]

Thanks for all the help!

def get_mapped_permits(self):
    with arcpy.da.SearchCursor(self.feature_class, "PERMIT_NUM") as search:
        self.mapped_permits = [permit[0] for permit in search]

# called from

self.mapped_permits.clear()
t = threading.Thread(target=self.get_mapped_permits, daemon=True)
t.start()

AustinBachurski

Spent some time tonight looking through this, there's so much here! So many things I didn't think about or know about, such simpler solutions to some of the things I was doing. The fact that the class (somewhat?) gets rid of the global variables alone is awesome, still learning how to best avoid them, but this is great. Spurs a pile of additional questions but I'll save those for google. I really can't thank you enough for taking the time to do this, this is going to help move me forward so much, thank you thank you thank you!

P.S. I still need to try your suggestion for the SearchCursor issue, but I haven't had a chance yet - thanks again.

AustinBachurski

Oh wow, thanks for taking time out of your day to go through that. I'll be sure to look over it tomorrow when I have some more time. Really appreciate it!

blake.terhune

+1 for yeah_toast() method.

Anonymous User

I did some refractoring while looking at your code. Changed some lines to show alternatives, but this is how I would set it up using a class.

import arcpy
import os
import threading
import time
import requests
import subprocess
from collections import defaultdict
from infi.systray import SysTrayIcon
from win11toast import toast


class ApplicationWatcher():

    def __int__(self):
        # set class properties
        self.missing = []
        self.search_results = defaultdict(dict)

    def check_now(self, method):
        """
        method to trigger respective data manipulations
        """

        # clear the previous iteration values
        self.missing.clear()
        self.search_results.clear()

        # get a token
        cityworks_token = self.get_token(os.environ.get("CITYWORKS_UN"), os.environ.get("CITYWORKS_PW"))

        # process the parcels depending on the value of method
        if method == "update":
            self.update_search(cityworks_token)

        self.search_results = self.get_issued_permits(cityworks_token)

        # Get list of case numbers from the result
        issued_permits = [result['CASE_NUMBER'] for result in self.search_results]

        # get list of mapped permits
        mapped_permits = self.get_mapped_permits()
        # Debugging
        # print("Issued:")
        # for issued in issued_permits:
        #     print(issued)
        # print("Mapped:")
        # for mapped in mapped_permits:
        #     print(mapped)
        
        # Compare the issued permits and mapped permits
        if method == "clicked" or method == "":
            # compare the issued permits to the mapped permits
            self.missing = self.compare(issued_permits, mapped_permits)
        elif method == "crosscheck":
            # compare the mapped permits to the issued permits
            self.missing = self.compare(mapped_permits, issued_permits)
        elif method == "duplicates":
            # find duplicates in the mapped permits
            self.missing = self.duplicates(mapped_permits)
        
        
        # Get permits needing to be mapped
        if method == "clicked" and self.missing or method == "" and self.missing:
            if len(self.missing) == 1:
                message = "The following permit needs to be mapped:"
            else:
                message = f"The following {len(self.missing)} permits need to be mapped:"
            for permit in self.missing:
                message += f"\n{permit}"
            self.yeah_toast(message)
            
        elif method == "crosscheck" and self.missing:
            if len(self.missing) == 1:
                message = "The following permit was found on the map but not in Cityworks:"
            else:
                message = f"The following {len(self.missing)} permits were found on the map but not in Cityworks:"
            for permit in self.missing:
                message += f"\n{permit}"
            self.yeah_toast(message)

        elif method == "duplicates" and self.missing:
            if len(self.missing) == 1:
                message = "The following permit was found as a duplicate:"
            else:
                message = f"The following {len(self.missing)} permits were found as duplicates on the map:"
            for permit in self.missing:
                message += f"\n{permit}"
            self.yeah_toast(message)
        
        # Display 
        if method != "" and not self.missing:
            self.no_toast()

    def compare(self, base, tester):
        return [permit for permit in base if permit not in tester]
        
    # def compare(self, issued, mapped):
    #     permits = [permit for permit in issued if permit not in mapped]
    #     return permits
    # 
    # def crosscheck(self, issued, mapped):
    #     permits = [permit for permit in mapped if permit not in issued]
    #     return permits

    def duplicates(self, mapped):
        # permits = []
        # comparitor = []
        #
        # for permit in mapped:
        #     if permit not in comparitor:
        #         comparitor.append(permit)
        #     else:
        #         permits.append(permit)

        # shorten this to a list comprehension by counting the values in the list and returning
        # those that are equal or grater than 2
        return [x for x in mapped if mapped.count(x) >= 2]

    def get_token(self, username, password):
        arguments = 'data={"LoginName":"' + username + '","Password":"' + password + '"}'
        response = requests.get("https://cityworks.ci.kalispell.mt.us/Cityworks/Services/General/"
                                "Authentication/Authenticate", params=arguments, timeout=5)
        cityworks_token = (response.json()["Value"]['Token'])
        return cityworks_token

    def get_issued_permits(self, token):
        # Search ID for BPISSUED is 994
        bp_url = "https://cityworks.ci.kalispell.mt.us/Cityworks/services/Ams/Search/Execute?" \
                 "data={%22SearchId%22:994}&token=" + token
        results = requests.get(bp_url, timeout=10)
        return results.json()["Value"]

    def get_mapped_permits(self):
        building_permits_2022 = r"J:\Austin\Projects\SDE Maintenance\Building.sde\Kalispell.BUILDING.BuildingPermits\Kalispell.BUILDING.BuildingPermits2022"
        
        # the 'for each in permi' conditional is redundant and you can just use the index.
        # mapped = [each for permit in arcpy.da.SearchCursor(building_permits_2022, "PERMIT_NUM") for each in permit]
        return [permit[0] for permit in arcpy.da.SearchCursor(building_permits_2022, "PERMIT_NUM")]

    def no_toast(self):
        toast("Nothing to do...")

    def open_web_pages(self):
        if self.missing:
            for case_number in self.missing:
                for result in self.search_results:
                    if result['CASE_NUMBER'] == case_number:
                        ca_obj_id = int(result['CA_OBJECT_ID'])
                        ca_case_type_id = int(result['CASE_TYPE_ID'])
                        subprocess.Popen([r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
                                          f"https://cityworks.ci.kalispell.mt.us/Cityworks/CwPermit/UF/Case/Page/CUF.aspx?"
                                          f"CaObjId={ca_obj_id}&CaCaseTypeId={ca_case_type_id}&CaSubTypeId="])
        else:
            toast("Nothing to show...")

    # def parse_results(self, results):
    #     permits = [result['CASE_NUMBER'] for result in results]
    #     return permits

    def update_search(self, token):
        url = f"https://cityworks.ci.kalispell.mt.us/Cityworks/Services/Ams/Search/PllSaved?token={token}"
        results = requests.get(url)
        result_values = results.json()["Value"]
        filtered_results = [result for result in result_values if "BACHURSKI" in result["EmployeeName"]]
        display = ""
        for each in filtered_results:
            display += str(f"SearchId: {each['SearchId']}\nSearchName: {each['SearchName']}\n\n")
        toast("New Building Permit", display)

    def yeah_toast(self, info):
        toast("New Building Permit", info, on_click=lambda args: self.open_web_pages())


if __name__ == "__main__":

    # Create instance
    app = ApplicationWatcher()

    menu_options = (("Check Permits Now", None, lambda args: app.check_now("clicked")),
                    ("Open Permit Web Page(s)", None, app.open_web_pages),
                    ("Crosscheck Permits", None, lambda args: app.check_now("crosscheck")),
                    ("Check for Duplicates", None, lambda args: app.check_now("duplicates")),
                    ("Update Search ID", None, lambda args: app.check_now("update")))

    systray = SysTrayIcon("TrayIcon.ico", "Building Permit Monitor", menu_options)
    systray.start()

    def monitoring():
        while True:
            app.check_now("")
            # Check every 30 minutes.
            time.sleep(1800)


    threading.Thread(target=monitoring, daemon=True).start()

AustinBachurski

I'm pretty sure I tried moving it out of the comprehension and it didn't change anything, but I'll have to try it as you suggested with deleting it after assembling the list and see what happens. I haven't messed with classes at all yet, I've watched several tutorials but have yet to be able to get my brain to understand where they would be useful and how to implement them. Any chance you could point out a couple areas where you feel it would be beneficial? I'll figure out how to get it implemented, but I think I need some help understanding when they would be useful. No biggie if not, thanks for your response.

Anonymous User

Returning the whole row might be keeping the reference to the cursor iterator, which might cause it to not be released and the dataset (cached for speed) to be used again. Could try moving it out of the list comprehension and explicitly deleting the cursor when the list is built.

I think converting some parts of the script to a Class would help manage the application state better, and then just instantiate a new class in each new cycle to process the data.

AustinBachurski

In response to JoshuaBixby, I did just test it with a scratch database on my local machine and it updates as expected, at least in the console window. Pretty much says it has to be something to do with the fact that it's on an enterprise SDE or whatever. Just seems odd that closing/opening the tool would change things if that's the case. Not sure what the deal is...

AustinBachurski

The reasoning for the comprehension on line 3 is that the SearchCursor object is a list of tuples, that was what I came up with to get the contents of each tuple as a string, probably a better way to do it, but as stated I'm still learning... If it helps, I put the code up on github fair warning, I'm new to coding in general so it's probably horrible, but I'd gladly take any criticism - gotta learn somehow. As far as I'm aware I'm the only one who is editing this particular feature, our organization is fairly small. There are a couple map services that reference the layer, but I guess I wouldn't expect that to affect this.

RhettZufelt

Was wondering what you meant "runs in the system tray" and how it is initiated.

However, if it still doesn't update in python console, then something else is going on. Suspect maybe a version issue as @JoshuaBixby suggests.

I did my testing on SDE FC with traditional moving edits to base, so any changes are represented on save.

RhettZufelt

I thought this at first also, but testing gives the same result with both list comprehension syntax.

blake.terhune

I'm not sure this is the solution to your problem, but the list comprehension on line 3 seems wrong. It should be something like

def get_mapped_permits():
    building_permits_2022 = "path.to.sde"
    mapped = [permit[0] for permit in arcpy.da.SearchCursor(building_permits_2022, "PERMIT_NUM")]
    return mapped

JoshuaBixby

Have you tested it outside of an enterprise geodatabase? Is the code your running and the changes to the data working on the same version of the data? If I have users modifying data in version X and they don't reconcile and post, I won't see them in version Y.

AustinBachurski

I'm still learning so I may not be answering your question correctly... The code block in my first post (aside from the path, stored in a variable) is the exact function I'm using, have a while loop running in a thread that calls it every 30 minutes, the exact line I'm using for the call is:

mapped_permits = get_mapped_permits()

At first, I thought it was something goofy with the threading, but I tested it in PyCharm using the python console there and it won't update, I can delete all the variables I'm using in the console but still, every time I call it, it returns the same data.

RhettZufelt

Normally, the way to get SearchCursor to "pull new data" is to establish it again.

Not sure how you are calling it, but if I test your code in IDLE shell, it reports all additions to the FC each time I run it (as long as I saved the changes). Of course, I have to replace "path.to.sde" with the path/name to actual feature class).

Rejoindre la conversation

Les membres connectés peuvent publier, suivre les mises à jour, et plus encore. Nouveau ici ? Inscrivez-vous gratuitement.

Liens rapides

Community Help
Find useful guides, FAQs, and documents to help you navigate and make the most of Esri Community.