Samples

Basic Concepts

A quality verification is a test run based on a specific quality specification. A quality specification holds one or more quality conditions. Each quality condition represents a test algorithm configured for one or more specific datasets.

In order to execute a verification the prosuite.verification.service class is used to create the communication channel to the server and to start the verification of a specific quality specification.

service = prosuite.verification.service(host_name='localhost', port_nr=5151)
service.verify(specification=my_specification)

A quality verification can be based on an XML specification (exported from the ProSuite Data Dictionary) or on a specification created in code, containing a list of quality conditions. Quality conditions are created with the factory class prosuite.factories.quality_conditions.Conditions which contains all available test algorithms. Intellisense provides the method parameters and help/docstrings but for an overview over the available tests or a general introduction, refer to the ProSuite HTML help or the ‘Quick Reference’.

Before running the verification in Python, make sure the server is running, for example by starting prosuite-qa-microservice.exe. By default the communication channel is http://localhost:5151.

Verify a specification created in code

The Specification class holds a set of conditions that can be configured programmatically.

1. Define the Data Model (= workspace)

2. Create the Datasets (= feature classes, tables with optional filter) in the model

3. Create a Service instance containing the connection properties

4. Define the specification: create a prosuite.quality instance

5. Create Condition instances using the static Conditions class and add them to the specification

6. Optionally define the verification perimeter

7. Optionally define the verification output directory

8. Execute the verification

import prosuite as ps

model = ps.Model("TopoModel", "D:\Test Data\ExtractStGallen.gdb") # From "ProSuite Documentation / SampleData.zip"
datasets = [ps.Dataset("TLM_FLIESSGEWAESSER", model),
            ps.Dataset("TLM_STRASSE", model)]

service = ps.Service(host_name='localhost', port_nr=5151) # You might want to change this to the host and port of your ProSuite installation

simpleSpecification = ps.Specification(
    name='MinimumLengthSpecification',
    description='A very simple quality specification checking feature and segment length of roads and rivers')

for dataset in datasets:
    simpleSpecification.add_condition(ps.Conditions.qa_min_length_0(dataset, limit=10, is3_d=False))
    simpleSpecification.add_condition(ps.Conditions.qa_segment_length_0(dataset, 1.5, False))

envelope = ps.EnvelopePerimeter(x_min=2750673, y_min=1215551, x_max=2765845, y_max=1206640)

out_dir = 'C:/temp/verification_output' # You might want to change this to a directory that exists on your system, also make sure no Issue.gdb exists in this directory

verification_responses = service.verify(specification=simpleSpecification, output_dir=out_dir, perimeter=envelope)

for verification_response in verification_responses:
    print(verification_response.message)

Response Messages

Creating external issue file geodatabase
Starting quality verification using quality specification MinimumLengthSpecification with verification tile size 5000Extent: 15172 x 8911
X-Min: 2750673
Y-Min: 1206640
X-Max: 2765845
Y-Max: 1215551

Verifying quality conditions per cached tiles (container tests)
Processing tile 0 of 8: XMin: 2’750’673.00 YMin: 1’206’640.00 XMax: 2’755’673.00 YMax: 1’211’640.00
Processing tile 1 of 8: XMin: 2’755’673.00 YMin: 1’206’640.00 XMax: 2’760’673.00 YMax: 1’211’640.00
Processing tile 2 of 8: XMin: 2’760’673.00 YMin: 1’206’640.00 XMax: 2’765’673.00 YMax: 1’211’640.00
Processing tile 3 of 8: XMin: 2’765’673.00 YMin: 1’206’640.00 XMax: 2’765’845.00 YMax: 1’211’640.00
Processing tile 4 of 8: XMin: 2’750’673.00 YMin: 1’211’640.00 XMax: 2’755’673.00 YMax: 1’215’551.00
Processing tile 5 of 8: XMin: 2’755’673.00 YMin: 1’211’640.00 XMax: 2’760’673.00 YMax: 1’215’551.00
Processing tile 6 of 8: XMin: 2’760’673.00 YMin: 1’211’640.00 XMax: 2’765’673.00 YMax: 1’215’551.00
Processing tile 7 of 8: XMin: 2’765’673.00 YMin: 1’211’640.00 XMax: 2’765’845.00 YMax: 1’215’551.00
Quality verification finishedNumber of verified datasets: 2.
Number of verified conditions: 4
No category
QaMinLength(0) TLM_FLIESSGEWAESSER - errors: 290
QaMinLength(0) TLM_STRASSE - errors: 974
QaSegmentLength(0) TLM_FLIESSGEWAESSER - errors: 1733
QaSegmentLength(0) TLM_STRASSE - errors: 1939
Warning count: 0
Error count: 4’936
The quality specification is not fulfilled

Issues written to C:\temp\verification_output\Issues.gdb

Verification report written to C:\temp\verification_output\verification.xml
Html report:
C:\temp\verification_output\verification.html
Quality specification report:
C:\temp\verification_output\qualityspecification.html

Control verification with the Issue class

The Issue class can be used to control the verification process. It can be used to stop the verification process when a certain issue condition is met. A sample Python script can be found in the ProSuite Python Samples repository.

issue_allowable = True

 for verification_response in verification_responses:
     if len(verification_response.issues) > 0:
         for issue in verification_response.issues:
             # Demo Prints

             # print(issue.description)
             # print(issue.involved_objects)
             # print(issue.geometry)
             # print(issue.issue_code)
             # print(issue.allowable)
             # print(issue.stop_condition)

             if issue.allowable is False:
                 print(f"Not allowed issue met: {issue.description} in {issue.involved_objects[0].table_name}")
                 print("Stopping verification")
                 issue_allowable = False
                 break

     if issue_allowable is False:
         break

Verification using XML Specification

1. Create a Service instance. In this example the service runs on a remote server machine.

2. Define the quality specification: create a XmlSpecification instance from a specification.qa.xml file.

3. Define the verification output directory

4. Optionally define the verification perimeter

5. Execute the verification

import prosuite

service = prosuite.verification.service(host_name='arcgis_server', port_nr=5151)

xml_file = "\\share\QA\specifications\road_specification.qa.xml"
sde_file = "\\share\connection_files\production_QA_version.sde"

# Data source replacements: see notes below
xml_spec = prosuite.XmlSpecification(specification_file=xml_file,
                                     specification_name="Produktionsunterstuetzung",
                                     data_source_replacements=[["ProductionModel", sde_file]])

out_dir = '\\share\QA\results\verification_output'

for verification_response in service.verify(specification=xml_spec, output_dir = out_dir):
    print(verification_response.message_level)
    print(verification_response.service_call_status)
    print(verification_response.message)

Notes:

Directories: The specified paths must be accessible by the server, hence use UNC-paths.

Data Source Replacements: The datasets in the XML specifications use a Workspace ID as reference to the database. The Workspace ID is defined at the bottom of the XML file. Data source replacements are provided in the format of string pairs [<workspace_id>, <path_to_geodatabase>].

Examples:

• [“TLM_Data”, “C:/full/path/to/connection.sde”]

• [“ERM_Data”, “C:/full/path/to/data.gdb”]

For each Workspace ID defined at the very end of the XML file, provide a path to a file geodatabase or an sde file.

• If no data source replacement is provided, a connectionString must be defined for the respective workspace in the XML. This is achieved by exporting the full workspace information when exporting the XML specification in the data dictionary editor by checking the Option ‘Export workspace connection information’.

• If one or more datasource replacements are provided they will be used to open the referenced datasets in the XML by using the specified geodatabase for the respective workspace id.

For more details see the documentation of the verification service: prosuite.verification.Service.

Get specification names from XmlSpecification

import prosuite
xml_file = 'C:/temp/road_specification.qa.xml'
names_list = prosuite.XmlSpecification.get_specification_names(xml_file)
print(names_list)

Verification using DDX Specification

Using the DDX Specification, you can directly run quality verification defined in your Data Dictionary Editor through the Python API. This approach allows you to leverage predefined quality specifications and dataset configurations, making it simple to execute targeted data quality checks programmatically based on the settings in your data dictionary.

1. Create a Service instance. In this example, the service runs on a local server machine.

2. Define the quality specification by creating a DdxSpecification instance using the ddx_id and the project_short_name. These identifiers can be found in the Data Dictionary Editor.

3. Define the verification parameters, including any specific dataset ID and object IDs to verify. Both the dataset ID and object IDs can also be checked in the Data Dictionary Editor.

4. Optionally define advanced parameters such as update_issues_in_verified_model and save_verification_statistics.

5. Define the verification output directory.

6. Execute the verification.

import prosuite

service = prosuite.Service(host_name='localhost', port_nr=5151)

ddx_id = 9
project_short_name = "..."
specification = prosuite.DdxSpecification(ddx_id=ddx_id, project_short_name=project_short_name)

road_dataset_id = 138
road_object_ids_to_verify = [3606433]
rail_dataset_id = 141
rail_object_ids_to_verify = [2105461, 2105452, 2105593]

params = prosuite.VerificationParameters(user_name="UJR")

# Optionally provided object ID lists per dataset (use dataset ID from data dictionary)
params.add_objects_to_verify(road_dataset_id, road_object_ids_to_verify)
params.add_objects_to_verify(rail_dataset_id, rail_object_ids_to_verify)

# Optionally, write the issue to the error datasets in the data model
params.update_issues_in_verified_model = True

# Optionally, save the verification statistics to the respective data dictionary table (requires DDX schema 1.0.0.1)
params.save_verification_statistics = True

out_dir = 'C:/temp/verification_output'
extent = None  # Optionally define the verification extent

for verification_response in service.verify(specification=specification, output_dir=out_dir, perimeter=extent, parameters=params):
    print(verification_response.message_level)
    print(verification_response.service_call_status)
    print(verification_response.message)

Notes:

• Directories: Ensure the specified paths are accessible by the server, especially for output directories.

• update_issues_in_verified_model: If set to True, this updates issues within the error datasets in the verified model.

• save_verification_statistics: If set to True, this option saves verification statistics into the Data Dictionary database.

Verification using DDX Specification with Automatic Object List Generation

In addition to specifying individual object IDs, you can also automatically retrieve all feature IDs within a dataset for comprehensive quality verification. This can be done by using a function (e.g., get_object_ids_from_feature_class) to fetch all object IDs, allowing you to apply checks across an entire dataset. This approach is especially useful for verifying large datasets.

Use this method to streamline quality verification for complete datasets or to apply selection queries for filtered checks. Simply generate the object list and add it to your verification parameters.

import arcpy

def get_object_ids_from_feature_class(feature_class, selection_query=None):
    object_ids = []
    with arcpy.da.SearchCursor(feature_class, ["OBJECTID"], selection_query) as cursor:
        for row in cursor:
            object_ids.append(row[0])
    return object_ids

ddx_id = 9
ddx_project_short_name = ""
road_dataset_id = 138


road_feature_class = r"C:\DIRA\... sde\feature_class"
road_selection_query = None  # Optional: SQL filter query

road_object_ids_to_verify = get_object_ids_from_feature_class(road_feature_class, road_selection_query)

Verification on Secure Channel

In this example, the grpc.ssl_channel_credentials object is created by a utility method, that gets the required root certificates automatically from the windows certificate store. For advanced scenarios or credentials on a non-windows platform, see the gRPC Python docs.

import prosuite
ssl_credentials = prosuite.utils.get_ssl_channel_credentials()

# if channel_credentials are passed to the Verification constructor, a secure channel will be established.
service = prosuite.verification.service(host_name='localhost', port_nr=5151, channel_credentials=ssl_credentials)

Define a WKB perimeter

import prosuite
poly_as_hex_string = '01ee0300000100000001eb03000001000000050000004060e5e8cfd5434100c3640aa44f32410000000000000000f8065f282dd6434100c3640aa44f32410000000000000000f8065f282dd6434170d71262d64f324100000000000000004060e5e8cfd5434170d71262d64f324100000000000000004060e5e8cfd5434100c3640aa44f32410000000000000000'
wkb_perimeter = prosuite.WkbPerimeter(bytes.fromhex(poly_as_hex_string))

# the wkb_perimeter can be assigned to the perimeter parameter in verify()

Note

The variable ‘poly_as_hex_string’ is the hex string representation of a polygon or envelope. It can be produced for example from an arcpy.Geometry. Any arcpy.Geometry can be converted to WKB and encoded as hex based string:

poly_as_hex_string = arcpy_polygon_geometry.WKB.hex()

Acessing a verification response

service.verify() returns an iterable of ResponseVerification objects. It is iterable because the verification service returns a reponse stream. Hence the progress can be printed in real-time.

for verification_response in service.verify():
    print(verification_response.message_level)
    print(verification_response.service_call_status)
    print(verification_response.message)

Advanced Parameters

Optionally, change advanced verification parameters, such as the Verification tile_size (the default is 5000m)

import prosuite

xml_file = 'C:/temp/road_specification.qa.xml'
service = prosuite.verification.service(host_name='localhost', port_nr=5151)

xml_spec = prosuite.XmlSpecification(
    specification_file=xml_file, specification_name="Produktionsunterstuetzung",
                                     data_source_replacements=[["ProductionModel", sde_file]])

params = prosuite.verification.VerificationParameters(tile_size=10000)

out_dir = 'C:/temp/verification_output'

for verification_response in service.verify(specification=spec, output_dir=out_dir, parameters=params):
    print(verification_response)

for verification_response in service.verify(specification=spec, output_dir = out_dir):
    print(verification_response)

Start and stop the local service process

If no service is constantly running and the python script should run without interaction, e.g. as a batch job, the server process can be started directly from python on the local machine. In this example, an XML specification is used.

import time
import subprocess
import prosuite

# Start the service from a local server installation with the default port.
# It will fail and shut down immediately if another service is already serving on the same port.
server_process = subprocess.Popen(r"C:\ProSuite\Server\prosuite-qa-microservice.exe")

# Alternatively, provide a host name and custom port like this:
# server_process = subprocess.Popen(
#     [r"C:\ProSuite\Server\prosuite-qa-microservice.exe",
#     "--hostname", "LOCALHOST", "--port", "12345"])

# Wait for the process to start, initialize the ArcGIS license and the communication channl
time.sleep(10)

service = prosuite.verification.service(host_name='LOCALHOST', port_nr=5151)

xml_file = "C:/Data/specifications/road_specification.qa.xml"
workspace = "C:/Data/TopographicData.gdb"

xml_spec = prosuite.XmlSpecification(specification_file=xml_file,
                                    specification_name="Produktionsunterstuetzung",
                                    data_source_replacements=[["ProductionModel", workspace]])

out_dir = 'C:/Temp/verification_output'

for verification_response in service.verify(specification=spec, output_dir = out_dir):
    print(verification_response)

# Stop the service
server_process.kill()