From a5e51bb385bf33b6f1aff45171d1f556565d88ec Mon Sep 17 00:00:00 2001 From: andrew misiti Date: Fri, 15 May 2026 12:06:37 -0400 Subject: [PATCH 1/3] ct-200 start of vision cat apis --- .../pingvision/pingvision_api_client.py | 34 +++++++++++++++++- src/pingintel_api/pingvisionapi_cmd.py | 35 +++++++++++++++++++ 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/src/pingintel_api/pingvision/pingvision_api_client.py b/src/pingintel_api/pingvision/pingvision_api_client.py index 2a31f7f..603ecf6 100644 --- a/src/pingintel_api/pingvision/pingvision_api_client.py +++ b/src/pingintel_api/pingvision/pingvision_api_client.py @@ -12,7 +12,7 @@ import time from datetime import timedelta from timeit import default_timer as timer -from typing import BinaryIO, Literal, TypedDict, overload, List +from typing import BinaryIO, Literal, TypedDict, overload, List, Dict from typing import BinaryIO, TypedDict, Unpack, overload from pingintel_api.api_client_base import APIClientBase @@ -373,3 +373,35 @@ def add_data_items(self, pingid: str, action: t.DATA_ITEM_ACTIONS, items: dict[s url = self.api_url + f"/api/v1/submission/{pingid}/add_data_items" response = self.post(url, json={"items": items, "action": action}) raise_for_status(response) + + def create_acc_loc_job( + self, + pingid: str, + modeling_option_uuids: list[str] | None = None, + cat_model_type: str | None = None, + use_secondary_modifiers: bool | None = None, + use_ping_geocoding: bool | None = None, + layer_output: str | None = None, + air_modeling_workflow_name: str | None = None, + rms_edm_name: str | None = None, + ) -> Dict: # TODO - add type definition + """ initiate acc/loc file generation job""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/acc-loc-files" + + data = { + 'modeling_option_uuids': modeling_option_uuids, + 'cat_model_type': cat_model_type, + 'use_secondary_modifiers': use_secondary_modifiers, + 'use_ping_geocoding': use_ping_geocoding, + 'layer_output': layer_output, + 'air_modeling_workflow_name': air_modeling_workflow_name, + 'rms_edm_name': rms_edm_name, + } + + + response = self.post(url, data=data) + raise_for_status(response) + + self.logger.info(f"Acc/loc job created: {response.json()}") + response_data = response.json() + return response_data \ No newline at end of file diff --git a/src/pingintel_api/pingvisionapi_cmd.py b/src/pingintel_api/pingvisionapi_cmd.py index 4ae7949..b728899 100644 --- a/src/pingintel_api/pingvisionapi_cmd.py +++ b/src/pingintel_api/pingvisionapi_cmd.py @@ -362,6 +362,41 @@ def events(ctx, pretty, pingid, division, team, start, cursor_id, page_size): pprint.pprint(results) +@cli.command() +@click.pass_context +@click.option("--pingid", help="Filter by Ping ID") +@click.option("--modeling-option-uuids", multiple=True, help="Filter by modeling option UUIDs") +@click.option("--cat-model-type", help="AIR/RMS") +@click.option("--use-secondary-modifiers", type=bool, help="Use secondary modifiers (True/False)") +@click.option("--use-ping-geocoding", type=bool, help="Use Ping geocoding (True/False)") +@click.option("--layer-output", help="Layer output") +@click.option("--air-modeling-workflow-name", help="AIR modeling workflow name") +@click.option("--rms-edm-name", help="RMS EDM name") +def create_acc_loc_job(ctx, pingid, modeling_option_uuids, cat_model_type, use_secondary_modifiers, use_ping_geocoding, layer_output, air_modeling_workflow_name, rms_edm_name): + """Create an ACC LOC file generation job. + + pingvisionapi create-acc-loc-job --pingid [--modeling-option-uuids --modeling-option-uuids ...] [--cat-model-type ] [--use-secondary-modifiers True/False] [--use-ping-geocoding True/False] [--layer-output ] [--air-modeling-workflow-name ] [--rms-edm-name ] + """ + + client = get_client(ctx) + + if not pingid: + raise click.UsageError("The --pingid option is required.") + + results = client.create_acc_loc_job( + pingid=pingid, + modeling_option_uuids=modeling_option_uuids, + cat_model_type=cat_model_type, + use_secondary_modifiers=use_secondary_modifiers, + use_ping_geocoding=use_ping_geocoding, + layer_output=layer_output, + air_modeling_workflow_name=air_modeling_workflow_name, + rms_edm_name=rms_edm_name, + ) + + pprint.pprint(results) + + def main(): cli() From f9a871b5007ba8ea45a56910c7f0942f6f7ec197 Mon Sep 17 00:00:00 2001 From: andrew misiti Date: Tue, 16 Jun 2026 10:31:27 -0400 Subject: [PATCH 2/3] wip --- src/pingintel_api/pingvision/pingvision_api_client.py | 2 +- src/pingintel_api/pingvision/types.py | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/pingintel_api/pingvision/pingvision_api_client.py b/src/pingintel_api/pingvision/pingvision_api_client.py index 603ecf6..862fc2d 100644 --- a/src/pingintel_api/pingvision/pingvision_api_client.py +++ b/src/pingintel_api/pingvision/pingvision_api_client.py @@ -384,7 +384,7 @@ def create_acc_loc_job( layer_output: str | None = None, air_modeling_workflow_name: str | None = None, rms_edm_name: str | None = None, - ) -> Dict: # TODO - add type definition + ) -> t.ModelingJobResponse: """ initiate acc/loc file generation job""" url = self.api_url + f"/api/v1/submission/{pingid}/cat/acc-loc-files" diff --git a/src/pingintel_api/pingvision/types.py b/src/pingintel_api/pingvision/types.py index f902660..51ef3e0 100644 --- a/src/pingintel_api/pingvision/types.py +++ b/src/pingintel_api/pingvision/types.py @@ -229,3 +229,10 @@ class OutputData(TypedDict): scrubbed_filename: str output_format: str url: str + + +class ModelingJobResponse(TypedDict): + uuid: str + processing_pct_complete: int + status: str + modeling_sets: list[dict[str, Any]] \ No newline at end of file From 3d3ae7740adeeedff5a87c64100cd2bd0f41f005 Mon Sep 17 00:00:00 2001 From: andrew misiti Date: Tue, 16 Jun 2026 11:35:24 -0400 Subject: [PATCH 3/3] ct-199 wip --- .../pingvision/pingvision_api_client.py | 194 ++++++++++++++ src/pingintel_api/pingvision/types.py | 53 +++- src/pingintel_api/pingvisionapi_cmd.py | 242 +++++++++++++++++- 3 files changed, 487 insertions(+), 2 deletions(-) diff --git a/src/pingintel_api/pingvision/pingvision_api_client.py b/src/pingintel_api/pingvision/pingvision_api_client.py index 862fc2d..5dda244 100644 --- a/src/pingintel_api/pingvision/pingvision_api_client.py +++ b/src/pingintel_api/pingvision/pingvision_api_client.py @@ -403,5 +403,199 @@ def create_acc_loc_job( raise_for_status(response) self.logger.info(f"Acc/loc job created: {response.json()}") + response_data = response.json() + return response_data + + def duplicate_coverage_option( + self, + pingid: str, + coverage_option_uuid: str, + name: str | None = None, + ) -> t.CoverageOptionDetailResponse: + """Duplicate a coverage option, copying all of its peril and zone terms.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/coverage-option/{coverage_option_uuid}/duplicate" + + data = {} + if name is not None: + data["name"] = name + + response = self.post(url, json=data) + raise_for_status(response) + + response_data = response.json() + return response_data + + def list_coverage_options( + self, + pingid: str, + ) -> list[t.CoverageOptionResponse]: + """List all coverage options for a submission.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/coverage-options" + + response = self.get(url) + raise_for_status(response) + + response_data = response.json() + return response_data + + def create_coverage_option( + self, + pingid: str, + name: str | None = None, + ) -> t.CoverageOptionResponse: + """Create a coverage option for a submission.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/coverage-options" + + data = {} + if name is not None: + data["name"] = name + + response = self.post(url, json=data) + raise_for_status(response) + + response_data = response.json() + return response_data + + def list_layer_structures( + self, + pingid: str, + ) -> list[t.LayerStructureResponse]: + """List all layer structures for a submission.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/layer-structures" + + response = self.get(url) + raise_for_status(response) + + response_data = response.json() + return response_data + + def create_layer_structure( + self, + pingid: str, + name: str | None = None, + ) -> t.LayerStructureResponse: + """Create a layer structure for a submission.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/layer-structures" + + data = {} + if name is not None: + data["name"] = name + + response = self.post(url, json=data) + raise_for_status(response) + + response_data = response.json() + return response_data + + def duplicate_layer_structure( + self, + pingid: str, + layer_structure_uuid: str, + name: str | None = None, + ) -> t.LayerStructureDetailResponse: + """Duplicate a layer structure, copying all of its layers.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/layer-structure/{layer_structure_uuid}/duplicate" + + data = {} + if name is not None: + data["name"] = name + + response = self.post(url, json=data) + raise_for_status(response) + + response_data = response.json() + return response_data + + def add_layer_to_layer_structure( + self, + pingid: str, + layer_structure_uuid: str, + name: str | None = None, + included: bool | None = None, + attachment: float | None = None, + limit: float | None = None, + participation_amount: float | None = None, + participation_percent: float | None = None, + premium: float | None = None, + ) -> t.LayerResponse: + """Add a layer to a layer structure.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/layer-structure/{layer_structure_uuid}/layers" + + data = {} + if name is not None: + data["name"] = name + if included is not None: + data["included"] = included + if attachment is not None: + data["attachment"] = attachment + if limit is not None: + data["limit"] = limit + if participation_amount is not None: + data["participation_amount"] = participation_amount + if participation_percent is not None: + data["participation_percent"] = participation_percent + if premium is not None: + data["premium"] = premium + + response = self.post(url, json=data) + raise_for_status(response) + + response_data = response.json() + return response_data + + def add_modeling_options_all_layers( + self, + pingid: str, + layer_structure_uuid: str, + coverage_option_uuids: list[str], + ) -> t.ModelingOptionLayerStructureResponse: + """Associate coverage options with every layer in a layer structure.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/layer-structure/{layer_structure_uuid}/modeling-options/add" + + response = self.post(url, json={"coverage_option_uuids": coverage_option_uuids}) + raise_for_status(response) + + response_data = response.json() + return response_data + + def remove_modeling_options_all_layers( + self, + pingid: str, + layer_structure_uuid: str, + coverage_option_uuids: list[str], + ) -> t.ModelingOptionLayerStructureResponse: + """Remove coverage options from every layer in a layer structure, deleting the modeling option records.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/layer-structure/{layer_structure_uuid}/modeling-options/remove" + + response = self.post(url, json={"coverage_option_uuids": coverage_option_uuids}) + raise_for_status(response) + + response_data = response.json() + return response_data + + def export_modeling_options( + self, + pingid: str, + ) -> list[t.ModelingOptionExportItem]: + """Get all modeling options for a submission.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/modeling-options/export" + + response = self.get(url) + raise_for_status(response) + + response_data = response.json() + return response_data + + def get_acc_loc_job( + self, + pingid: str, + job_uuid: str, + ) -> t.ModelingJobResponse: + """Get the status of an acc/loc file generation job.""" + url = self.api_url + f"/api/v1/submission/{pingid}/cat/acc-loc-files/{job_uuid}" + + response = self.get(url) + raise_for_status(response) + response_data = response.json() return response_data \ No newline at end of file diff --git a/src/pingintel_api/pingvision/types.py b/src/pingintel_api/pingvision/types.py index 51ef3e0..c6b7404 100644 --- a/src/pingintel_api/pingvision/types.py +++ b/src/pingintel_api/pingvision/types.py @@ -235,4 +235,55 @@ class ModelingJobResponse(TypedDict): uuid: str processing_pct_complete: int status: str - modeling_sets: list[dict[str, Any]] \ No newline at end of file + modeling_sets: list[dict[str, Any]] + + +class ModelingOptionExportItem(TypedDict): + uuid: str + created_time: str + modified_time: str + coverage_option: dict[str, Any] + layer_structure: dict[str, Any] + layers: list[dict[str, Any]] + + +class ModelingOptionLayerStructureResponse(TypedDict): + layer_structure_uuid: str + layer_structure_name: str + layers: list[dict[str, Any]] + + +class CoverageOptionResponse(TypedDict): + uuid: str + name: str + created_time: str + modified_time: str + bi_limit_type: str | None + default_bi_period_days: int | None + + +class CoverageOptionDetailResponse(CoverageOptionResponse): + peril_sections: list[dict[str, Any]] + per_zone_terms: list[dict[str, Any]] + + +class LayerStructureResponse(TypedDict): + uuid: str + name: str + created_time: str + modified_time: str + + +class LayerStructureDetailResponse(LayerStructureResponse): + layers: list[dict[str, Any]] + + +class LayerResponse(TypedDict): + uuid: str + name: str + included: bool + attachment: float | None + limit: float | None + participation_amount: float | None + participation_percent: float | None + premium: float | None \ No newline at end of file diff --git a/src/pingintel_api/pingvisionapi_cmd.py b/src/pingintel_api/pingvisionapi_cmd.py index b728899..7dc5892 100644 --- a/src/pingintel_api/pingvisionapi_cmd.py +++ b/src/pingintel_api/pingvisionapi_cmd.py @@ -362,6 +362,162 @@ def events(ctx, pretty, pingid, division, team, start, cursor_id, page_size): pprint.pprint(results) +# --------------------------------------------------------------------------- +# Policy Term Layer Structure APIs +# --------------------------------------------------------------------------- + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +@click.option("--layer-structure-uuid", required=True, help="UUID of the layer structure to duplicate.") +@click.option("--name", default=None, help="Name for the duplicate. Defaults to the source name.") +def duplicate_layer_structure(ctx, pingid, layer_structure_uuid, name): + """Duplicate a layer structure, copying all of its layers. + + pingvisionapi duplicate-layer-structure --pingid --layer-structure-uuid [--name ] + """ + + client = get_client(ctx) + + results = client.duplicate_layer_structure( + pingid=pingid, + layer_structure_uuid=layer_structure_uuid, + name=name, + ) + + pprint.pprint(results) + + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +def list_layer_structures(ctx, pingid): + """List all layer structures for a submission. + + pingvisionapi list-layer-structures --pingid + """ + + client = get_client(ctx) + + results = client.list_layer_structures(pingid=pingid) + + pprint.pprint(results) + + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +@click.option("--name", default=None, help="Layer structure name. Auto-generated if omitted.") +def create_layer_structure(ctx, pingid, name): + """Create a layer structure for a submission. + + pingvisionapi create-layer-structure --pingid [--name ] + """ + + client = get_client(ctx) + + results = client.create_layer_structure(pingid=pingid, name=name) + + pprint.pprint(results) + + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +@click.option("--layer-structure-uuid", required=True, help="UUID of the layer structure.") +@click.option("--name", default=None, help="Layer name. Auto-generated from limit/attachment if omitted.") +@click.option("--included/--not-included", default=None, help="Whether the layer is included.") +@click.option("--attachment", type=float, default=None, help="Attachment point.") +@click.option("--limit", type=float, default=None, help="Layer limit.") +@click.option("--participation-amount", type=float, default=None, help="Participation as a currency amount.") +@click.option("--participation-percent", type=float, default=None, help="Participation as a percentage.") +@click.option("--premium", type=float, default=None, help="Layer premium.") +def add_layer_to_layer_structure(ctx, pingid, layer_structure_uuid, name, included, attachment, limit, participation_amount, participation_percent, premium): + """Add a layer to a layer structure. + + pingvisionapi add-layer-to-layer-structure --pingid --layer-structure-uuid [--name ] [--attachment ] [--limit ] + """ + + client = get_client(ctx) + + results = client.add_layer_to_layer_structure( + pingid=pingid, + layer_structure_uuid=layer_structure_uuid, + name=name, + included=included, + attachment=attachment, + limit=limit, + participation_amount=participation_amount, + participation_percent=participation_percent, + premium=premium, + ) + + pprint.pprint(results) + + +# --------------------------------------------------------------------------- +# Policy Term Coverage Option APIs +# --------------------------------------------------------------------------- + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +@click.option("--coverage-option-uuid", required=True, help="UUID of the coverage option to duplicate.") +@click.option("--name", default=None, help="Name for the duplicate. Defaults to the source name.") +def duplicate_coverage_option(ctx, pingid, coverage_option_uuid, name): + """Duplicate a coverage option, copying all of its peril and zone terms. + + pingvisionapi duplicate-coverage-option --pingid --coverage-option-uuid [--name ] + """ + + client = get_client(ctx) + + results = client.duplicate_coverage_option( + pingid=pingid, + coverage_option_uuid=coverage_option_uuid, + name=name, + ) + + pprint.pprint(results) + + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +def list_coverage_options(ctx, pingid): + """List all coverage options for a submission. + + pingvisionapi list-coverage-options --pingid + """ + + client = get_client(ctx) + + results = client.list_coverage_options(pingid=pingid) + + pprint.pprint(results) + + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +@click.option("--name", default=None, help="Coverage option name. Defaults to 'Coverage Option' if omitted.") +def create_coverage_option(ctx, pingid, name): + """Create a coverage option for a submission. + + pingvisionapi create-coverage-option --pingid [--name ] + """ + + client = get_client(ctx) + + results = client.create_coverage_option(pingid=pingid, name=name) + + pprint.pprint(results) + + +# --------------------------------------------------------------------------- +# Policy Term Acc/Loc Job APIs +# --------------------------------------------------------------------------- + @cli.command() @click.pass_context @click.option("--pingid", help="Filter by Ping ID") @@ -374,7 +530,7 @@ def events(ctx, pretty, pingid, division, team, start, cursor_id, page_size): @click.option("--rms-edm-name", help="RMS EDM name") def create_acc_loc_job(ctx, pingid, modeling_option_uuids, cat_model_type, use_secondary_modifiers, use_ping_geocoding, layer_output, air_modeling_workflow_name, rms_edm_name): """Create an ACC LOC file generation job. - + pingvisionapi create-acc-loc-job --pingid [--modeling-option-uuids --modeling-option-uuids ...] [--cat-model-type ] [--use-secondary-modifiers True/False] [--use-ping-geocoding True/False] [--layer-output ] [--air-modeling-workflow-name ] [--rms-edm-name ] """ @@ -397,6 +553,90 @@ def create_acc_loc_job(ctx, pingid, modeling_option_uuids, cat_model_type, use_s pprint.pprint(results) +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +@click.option("--job-uuid", required=True, help="UUID of the acc/loc file generation job.") +def get_acc_loc_job(ctx, pingid, job_uuid): + """Get the status of an ACC LOC file generation job. + + pingvisionapi get-acc-loc-job --pingid --job-uuid + """ + + client = get_client(ctx) + + results = client.get_acc_loc_job( + pingid=pingid, + job_uuid=job_uuid, + ) + + pprint.pprint(results) + + +# --------------------------------------------------------------------------- +# Policy Term Modeling Option APIs +# --------------------------------------------------------------------------- + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +@click.option("--layer-structure-uuid", required=True, help="UUID of the layer structure.") +@click.option("--coverage-option-uuids", multiple=True, required=True, help="Coverage option UUID(s) to add.") +def add_modeling_options_all_layers(ctx, pingid, layer_structure_uuid, coverage_option_uuids): + """Associate coverage options with every layer in a layer structure. + + pingvisionapi add-modeling-options-all-layers --pingid --layer-structure-uuid --coverage-option-uuids --coverage-option-uuids + """ + + client = get_client(ctx) + + results = client.add_modeling_options_all_layers( + pingid=pingid, + layer_structure_uuid=layer_structure_uuid, + coverage_option_uuids=list(coverage_option_uuids), + ) + + pprint.pprint(results) + + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +@click.option("--layer-structure-uuid", required=True, help="UUID of the layer structure.") +@click.option("--coverage-option-uuids", multiple=True, required=True, help="Coverage option UUID(s) to remove.") +def remove_modeling_options_all_layers(ctx, pingid, layer_structure_uuid, coverage_option_uuids): + """Remove coverage options from every layer in a layer structure. + + pingvisionapi remove-modeling-options-all-layers --pingid --layer-structure-uuid --coverage-option-uuids --coverage-option-uuids + """ + + client = get_client(ctx) + + results = client.remove_modeling_options_all_layers( + pingid=pingid, + layer_structure_uuid=layer_structure_uuid, + coverage_option_uuids=list(coverage_option_uuids), + ) + + pprint.pprint(results) + + +@cli.command() +@click.pass_context +@click.option("--pingid", required=True, help="Ping ID of the submission.") +def export_modeling_options(ctx, pingid): + """Export all modeling options for a submission. + + pingvisionapi export-modeling-options --pingid + """ + + client = get_client(ctx) + + results = client.export_modeling_options(pingid=pingid) + + pprint.pprint(results) + + def main(): cli()