azext_edge/edge/providers/rpsaas/adr/assets.py

# coding=utf-8
# ----------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License file in the project root for license information.
# ----------------------------------------------------------------------------------------------

import json
from rich.console import Console
from typing import TYPE_CHECKING, Dict, List, Iterable, Optional, Union
from knack.log import get_logger
from azure.cli.core.azclierror import (
    InvalidArgumentValueError,
    RequiredArgumentMissingError,
)

from .user_strings import DUPLICATE_EVENT_ERROR, DUPLICATE_POINT_ERROR, INVALID_OBSERVABILITY_MODE_ERROR
from ....util import assemble_nargs_to_dict
from ....common import FileType
from ....util.az_client import get_registry_mgmt_client, wait_for_terminal_state, REGISTRY_API_VERSION
from ....util.queryable import Queryable

if TYPE_CHECKING:
    from ....vendor.clients.deviceregistrymgmt.operations import AssetsOperations


console = Console()
logger = get_logger(__name__)
ASSET_RESOURCE_TYPE = "Microsoft.DeviceRegistry/assets"
VALID_DATA_OBSERVABILITY_MODES = frozenset(["None", "Gauge", "Counter", "Histogram", "Log"])
VALID_EVENT_OBSERVABILITY_MODES = frozenset(["None", "Log"])


class Assets(Queryable):
    def __init__(self, cmd):
        super().__init__(cmd=cmd)
        self.deviceregistry_mgmt_client = get_registry_mgmt_client(
            subscription_id=self.default_subscription_id,
            api_version=REGISTRY_API_VERSION
        )
        self.ops: "AssetsOperations" = self.deviceregistry_mgmt_client.assets

    def create(
        self,
        asset_name: str,
        endpoint_profile: str,
        instance_name: str,
        resource_group_name: str,
        custom_attributes: Optional[List[str]] = None,
        default_topic_path: Optional[str] = None,
        default_topic_retain: Optional[str] = None,
        description: Optional[str] = None,
        disabled: bool = False,
        display_name: Optional[str] = None,
        documentation_uri: Optional[str] = None,
        events: Optional[List[str]] = None,
        events_file_path: Optional[List[str]] = None,
        external_asset_id: Optional[str] = None,
        hardware_revision: Optional[str] = None,
        instance_resource_group: Optional[str] = None,
        instance_subscription: Optional[str] = None,
        location: Optional[str] = None,
        manufacturer: Optional[str] = None,
        manufacturer_uri: Optional[str] = None,
        model: Optional[str] = None,
        product_code: Optional[str] = None,
        serial_number: Optional[str] = None,
        software_revision: Optional[str] = None,
        ds_publishing_interval: int = 1000,
        ds_sampling_interval: int = 500,
        ds_queue_size: int = 1,
        ev_publishing_interval: int = 1000,
        ev_sampling_interval: int = 500,
        ev_queue_size: int = 1,
        tags: Optional[Dict[str, str]] = None,
        **kwargs
    ):
        from .helpers import get_extended_location
        extended_location = get_extended_location(
            cmd=self.cmd,
            instance_name=instance_name,
            instance_resource_group=instance_resource_group or resource_group_name,
            instance_subscription=instance_subscription
        )
        cluster_location = extended_location.pop("cluster_location")

        # Properties
        properties = {
            "assetEndpointProfileRef": endpoint_profile,
            "events": _process_asset_sub_points("event_notifier", events),
        }
        if events_file_path:
            properties["events"].extend(
                _process_asset_sub_points_file_path(file_path=events_file_path)
            )

        # Other properties
        _update_properties(
            properties,
            custom_attributes=custom_attributes,
            default_topic_path=default_topic_path,
            default_topic_retain=default_topic_retain,
            description=description,
            display_name=display_name,
            disabled=disabled,
            documentation_uri=documentation_uri,
            external_asset_id=external_asset_id,
            hardware_revision=hardware_revision,
            manufacturer=manufacturer,
            manufacturer_uri=manufacturer_uri,
            model=model,
            product_code=product_code,
            serial_number=serial_number,
            software_revision=software_revision,
            ds_publishing_interval=ds_publishing_interval,
            ds_sampling_interval=ds_sampling_interval,
            ds_queue_size=ds_queue_size,
            ev_publishing_interval=ev_publishing_interval,
            ev_sampling_interval=ev_sampling_interval,
            ev_queue_size=ev_queue_size,
        )

        asset_body = {
            "extendedLocation": extended_location,
            "location": location or cluster_location,
            "properties": properties,
            "tags": tags,
        }
        with console.status(f"Creating {asset_name}..."):
            poller = self.ops.begin_create_or_replace(
                resource_group_name,
                asset_name,
                resource=asset_body
            )
            return wait_for_terminal_state(poller, **kwargs)

    def delete(self, asset_name: str, resource_group_name: str, **kwargs):
        self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
            check_cluster=True
        )
        with console.status(f"Deleting {asset_name}..."):
            poller = self.ops.begin_delete(
                resource_group_name,
                asset_name,
            )
            return wait_for_terminal_state(poller, **kwargs)

    def show(
        self, asset_name: str, resource_group_name: str, check_cluster: bool = False
    ) -> dict:
        asset = self.ops.get(
            resource_group_name=resource_group_name, asset_name=asset_name
        )
        if check_cluster:
            from .helpers import check_cluster_connectivity
            check_cluster_connectivity(self.cmd, asset)
        return asset

    def list(self, resource_group_name: Optional[str] = None) -> Iterable[dict]:
        if resource_group_name:
            return self.ops.list_by_resource_group(resource_group_name=resource_group_name)
        return self.ops.list_by_subscription()

    def query_assets(
        self,
        asset_name: Optional[str] = None,
        custom_query: Optional[str] = None,
        default_topic_path: Optional[str] = None,
        default_topic_retain: Optional[str] = None,
        description: Optional[str] = None,
        disabled: Optional[bool] = None,
        discovered: Optional[bool] = None,
        display_name: Optional[str] = None,
        documentation_uri: Optional[str] = None,
        endpoint_profile: Optional[str] = None,
        external_asset_id: Optional[str] = None,
        hardware_revision: Optional[str] = None,
        instance_name: Optional[str] = None,
        instance_resource_group: Optional[str] = None,
        location: Optional[str] = None,
        manufacturer: Optional[str] = None,
        manufacturer_uri: Optional[str] = None,
        model: Optional[str] = None,
        product_code: Optional[str] = None,
        resource_group_name: Optional[str] = None,
        serial_number: Optional[str] = None,
        software_revision: Optional[str] = None,
    ):
        query_body = custom_query or _build_query_body(
            asset_name=asset_name,
            default_topic_path=default_topic_path,
            default_topic_retain=default_topic_retain,
            description=description,
            disabled=disabled,
            display_name=display_name,
            documentation_uri=documentation_uri,
            endpoint_profile=endpoint_profile,
            external_asset_id=external_asset_id,
            hardware_revision=hardware_revision,
            location=location,
            manufacturer=manufacturer,
            manufacturer_uri=manufacturer_uri,
            model=model,
            product_code=product_code,
            resource_group_name=resource_group_name,
            serial_number=serial_number,
            software_revision=software_revision
        )
        query = f"Resources | where type =~\"{ASSET_RESOURCE_TYPE}\" " + query_body

        if any([instance_name, instance_resource_group]):
            instance_query = "Resources | where type =~ 'microsoft.iotoperations/instances' "
            if instance_name:
                instance_query += f"| where name =~ \"{instance_name}\""
            if instance_resource_group:
                instance_query += f"| where resourceGroup =~ \"{instance_resource_group}\""

            # fetch the custom location + join on innerunique. Then remove the extra customLocation1 generated
            query = f"{instance_query} | extend customLocation = tostring(extendedLocation.name) "\
                f"| project customLocation | join kind=innerunique ({query}) on customLocation "\
                "| project-away customLocation1"
        return self.query(query=query)

    def update(
        self,
        asset_name: str,
        resource_group_name: str,
        custom_attributes: Optional[List[str]] = None,
        default_topic_path: Optional[str] = None,
        default_topic_retain: Optional[str] = None,
        description: Optional[str] = None,
        disabled: Optional[bool] = None,
        display_name: Optional[str] = None,
        documentation_uri: Optional[str] = None,
        hardware_revision: Optional[str] = None,
        manufacturer: Optional[str] = None,
        manufacturer_uri: Optional[str] = None,
        model: Optional[str] = None,
        product_code: Optional[str] = None,
        serial_number: Optional[str] = None,
        software_revision: Optional[str] = None,
        ds_publishing_interval: Optional[int] = None,
        ds_sampling_interval: Optional[int] = None,
        ds_queue_size: Optional[int] = None,
        ev_publishing_interval: Optional[int] = None,
        ev_sampling_interval: Optional[int] = None,
        ev_queue_size: Optional[int] = None,
        tags: Optional[Dict[str, str]] = None,
        **kwargs
    ):
        # get the asset
        original_asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
            check_cluster=True
        )
        if tags:
            original_asset["tags"] = tags

        # Other properties
        _update_properties(
            original_asset["properties"],
            custom_attributes=custom_attributes,
            default_topic_path=default_topic_path,
            default_topic_retain=default_topic_retain,
            description=description,
            disabled=disabled,
            documentation_uri=documentation_uri,
            display_name=display_name,
            hardware_revision=hardware_revision,
            manufacturer=manufacturer,
            manufacturer_uri=manufacturer_uri,
            model=model,
            product_code=product_code,
            serial_number=serial_number,
            software_revision=software_revision,
            ds_publishing_interval=ds_publishing_interval,
            ds_sampling_interval=ds_sampling_interval,
            ds_queue_size=ds_queue_size,
            ev_publishing_interval=ev_publishing_interval,
            ev_sampling_interval=ev_sampling_interval,
            ev_queue_size=ev_queue_size,
        )

        with console.status(f"Updating {asset_name}..."):
            poller = self.ops.begin_create_or_replace(
                resource_group_name,
                asset_name,
                original_asset
            )
            return wait_for_terminal_state(poller, **kwargs)

    # Dataset
    # TODO: multi-dataset support
    def list_datasets(
        self,
        asset_name: str,
        resource_group_name: str,
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
        )
        return asset["properties"].get("datasets", [])

    def show_dataset(
        self,
        asset_name: str,
        dataset_name: str,
        resource_group_name: str,
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name
        )
        return _get_dataset(asset, dataset_name)

    # Data points
    def add_dataset_data_point(
        self,
        asset_name: str,
        dataset_name: str,
        data_point_name: str,
        data_source: str,
        resource_group_name: str,
        observability_mode: Optional[str] = None,
        queue_size: Optional[int] = None,
        sampling_interval: Optional[int] = None,
        replace: bool = False,
        **kwargs
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
            check_cluster=True
        )
        dataset = _get_dataset(asset, dataset_name, create_if_none=True)
        dataset["dataPoints"] = dataset.get("dataPoints", [])
        point_names = [point["name"] for point in dataset["dataPoints"]]
        if not replace and data_point_name in point_names:
            raise InvalidArgumentValueError(
                DUPLICATE_POINT_ERROR.format(data_point_name)
            )
        sub_point = _build_asset_sub_point(
            data_source=data_source,
            name=data_point_name,
            observability_mode=observability_mode,
            queue_size=queue_size,
            sampling_interval=sampling_interval,
        )
        dataset["dataPoints"].append(sub_point)

        # note that update does not return the properties
        with console.status(f"Updating {asset_name}..."):
            poller = self.ops.begin_create_or_replace(
                resource_group_name,
                asset_name,
                asset
            )
            asset = wait_for_terminal_state(poller, **kwargs)
        if not isinstance(asset, dict):
            asset = asset.as_dict()

        return _get_dataset(asset, dataset_name)["dataPoints"]

    def export_dataset_data_points(
        self,
        asset_name: str,
        dataset_name: str,
        resource_group_name: str,
        extension: str = FileType.json.value,
        output_dir: str = ".",
        replace: Optional[bool] = False
    ):
        from ....util import dump_content_to_file
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name
        )
        dataset = _get_dataset(asset, dataset_name)
        fieldnames = None
        if extension in [FileType.csv.value]:
            default_configuration = dataset.get("datasetConfiguration", "{}")
            if default_configuration == "{}":
                default_configuration = asset["properties"].get("defaultDatasetsConfiguration", "{}")
            fieldnames = _convert_sub_points_to_csv(
                sub_points=dataset.get("dataPoints", []),
                sub_point_type="dataPoints",
                default_configuration=default_configuration,
                portal_friendly=extension == FileType.csv.value
            )
            extension = extension.replace("-", ".")
        file_path = dump_content_to_file(
            content=dataset.get("dataPoints", []),
            file_name=f"{asset_name}_{dataset_name}_datapoints",
            extension=extension,
            fieldnames=fieldnames,
            output_dir=output_dir,
            replace=replace
        )
        return {"file_path": file_path}

    def import_dataset_data_points(
        self,
        asset_name: str,
        dataset_name: str,
        file_path: str,
        resource_group_name: str,
        replace: bool = False,
        **kwargs
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
            check_cluster=True
        )
        # should get the direct object so this should be enough
        dataset = _get_dataset(asset, dataset_name, create_if_none=True)
        dataset["dataPoints"] = _process_asset_sub_points_file_path(
            file_path=file_path,
            original_items=dataset.get("dataPoints", []),
            point_key="name",
            replace=replace
        )

        # note that update does not return the properties
        with console.status(f"Updating {asset_name}..."):
            poller = self.ops.begin_create_or_replace(
                resource_group_name,
                asset_name,
                asset
            )
            asset = wait_for_terminal_state(poller, **kwargs)
        if not isinstance(asset, dict):
            asset = asset.as_dict()
        return _get_dataset(asset, dataset_name)["dataPoints"]

    def list_dataset_data_points(
        self,
        asset_name: str,
        dataset_name: str,
        resource_group_name: str,
    ):
        dataset = self.show_dataset(
            asset_name=asset_name,
            dataset_name=dataset_name,
            resource_group_name=resource_group_name,
        )
        return dataset.get("dataPoints", [])

    def remove_dataset_data_point(
        self,
        asset_name: str,
        dataset_name: str,
        data_point_name: str,
        resource_group_name: str,
        **kwargs,
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
            check_cluster=True
        )
        dataset = _get_dataset(asset, dataset_name)

        dataset["dataPoints"] = [dp for dp in dataset.get("dataPoints", []) if dp["name"] != data_point_name]

        # note that update does not return the properties
        with console.status(f"Updating {asset_name}..."):
            poller = self.ops.begin_create_or_replace(
                resource_group_name,
                asset_name,
                asset
            )
            asset = wait_for_terminal_state(poller, **kwargs)
        if not isinstance(asset, dict):
            asset = asset.as_dict()

        return _get_dataset(asset, dataset_name)["dataPoints"]

    # Events
    def add_event(
        self,
        asset_name: str,
        resource_group_name: str,
        event_notifier: str,
        event_name: Optional[str] = None,
        observability_mode: Optional[str] = None,
        queue_size: Optional[int] = None,
        sampling_interval: Optional[int] = None,
        topic_path: Optional[str] = None,
        topic_retain: Optional[str] = None,
        replace: bool = False,
        **kwargs
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
            check_cluster=True
        )

        asset["properties"]["events"] = asset["properties"].get("events", [])
        event_names = [event["name"] for event in asset["properties"]["events"]]
        if not replace and event_name in event_names:
            raise InvalidArgumentValueError(
                DUPLICATE_EVENT_ERROR.format(event_name)
            )
        sub_point = _build_asset_sub_point(
            event_notifier=event_notifier,
            name=event_name,
            observability_mode=observability_mode,
            queue_size=queue_size,
            sampling_interval=sampling_interval
        )
        if topic_path:
            sub_point["topic"] = {
                "path": topic_path,
                "retain": topic_retain or "Never"
            }
        asset["properties"]["events"].append(sub_point)

        # note that update does not return the properties
        with console.status(f"Updating {asset_name}..."):
            poller = self.ops.begin_create_or_replace(
                resource_group_name,
                asset_name,
                asset
            )
            asset = wait_for_terminal_state(poller, **kwargs)
        if not isinstance(asset, dict):
            asset = asset.as_dict()
        return asset["properties"]["events"]

    def export_events(
        self,
        asset_name: str,
        resource_group_name: str,
        extension: str = FileType.json.value,
        output_dir: str = ".",
        replace: Optional[bool] = False
    ):
        from ....util import dump_content_to_file
        asset_props = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
        )["properties"]
        fieldnames = None
        if extension in [FileType.csv.value]:
            default_configuration = asset_props.get("defaultEventsConfiguration", "{}")
            fieldnames = _convert_sub_points_to_csv(
                sub_points=asset_props.get("events", []),
                sub_point_type="events",
                default_configuration=default_configuration,
                portal_friendly=extension == FileType.csv.value
            )
            extension = extension.replace("-", ".")
        file_path = dump_content_to_file(
            content=asset_props.get("events", []),
            file_name=f"{asset_name}_events",
            extension=extension,
            fieldnames=fieldnames,
            output_dir=output_dir,
            replace=replace
        )
        return {"file_path": file_path}

    def import_events(
        self,
        asset_name: str,
        file_path: str,
        resource_group_name: str,
        replace: bool = False,
        **kwargs
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
            check_cluster=True
        )
        asset["properties"]["events"] = _process_asset_sub_points_file_path(
            file_path=file_path,
            original_items=asset["properties"].get("events", []),
            point_key="name",
            replace=replace
        )

        # note that update does not return the properties
        with console.status(f"Updating {asset_name}..."):
            poller = self.ops.begin_create_or_replace(
                resource_group_name,
                asset_name,
                asset
            )
            asset = wait_for_terminal_state(poller, **kwargs)
        if not isinstance(asset, dict):
            asset = asset.as_dict()
        return asset["properties"]["events"]

    def list_events(
        self,
        asset_name: str,
        resource_group_name: str
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
        )

        return asset["properties"].get("events", [])

    def remove_event(
        self,
        asset_name: str,
        event_name: str,
        resource_group_name: str,
        **kwargs
    ):
        asset = self.show(
            asset_name=asset_name,
            resource_group_name=resource_group_name,
            check_cluster=True
        )
        asset["properties"]["events"] = [
            ev for ev in asset["properties"].get("events", []) if ev["name"] != event_name
        ]

        # note that update does not return the properties
        with console.status(f"Updating {asset_name}..."):
            poller = self.ops.begin_create_or_replace(
                resource_group_name,
                asset_name,
                asset
            )
            asset = wait_for_terminal_state(poller, **kwargs)
        if not isinstance(asset, dict):
            asset = asset.as_dict()
        return asset["properties"]["events"]


# New Helpers
def _get_dataset(asset: dict, dataset_name: str, create_if_none: bool = False):
    # ensure datasets will get populated if not there
    asset["properties"]["datasets"] = asset["properties"].get("datasets", [])
    datasets = asset["properties"]["datasets"]
    matched_datasets = [dset for dset in datasets if dset["name"] == dataset_name]
    # Temporary convert empty names to default
    if not matched_datasets and dataset_name == "default":
        matched_datasets = [dset for dset in datasets if dset["name"] == ""]
    # create if add or import (and no datasets yet)
    if not matched_datasets and create_if_none:
        if dataset_name != "default":
            raise InvalidArgumentValueError("Currently only one dataset with the name default is supported.")
        matched_datasets = [{}]
        datasets.extend(matched_datasets)
    elif not matched_datasets:
        raise InvalidArgumentValueError(f"Dataset {dataset_name} not found in asset {asset['name']}.")
    # note: right now we can have datasets with the same name but this will not be allowed later
    # part of the temporary convert
    matched_datasets[0]["name"] = dataset_name
    return matched_datasets[0]


def _build_topic(
    original_topic: Optional[Dict[str, str]] = None,
    topic_path: Optional[str] = None,
    topic_retain: Optional[str] = None
) -> Dict[str, str]:
    if not original_topic:
        original_topic = {}
    if topic_path:
        original_topic["path"] = topic_path
    if topic_retain:
        original_topic["retain"] = topic_retain
    elif not original_topic.get("retain"):
        original_topic["retain"] = "Never"

    if not original_topic.get("path"):
        raise RequiredArgumentMissingError("Topic path is needed for a topic configuration.")

    return original_topic


def _process_asset_sub_points_file_path(
    file_path: str,
    original_items: Optional[List[dict]] = None,
    point_key: Optional[str] = None,
    replace: bool = False
) -> List[Dict[str, str]]:
    from ....util import deserialize_file_content
    file_points = list(deserialize_file_content(file_path=file_path))
    _convert_sub_points_from_csv(file_points)

    if point_key is None:
        return file_points

    if not original_items:
        original_items = []
    original_points = {point[point_key]: point for point in original_items}
    file_points = {point[point_key]: point for point in file_points}
    for key in file_points:
        if key in original_points and not replace:
            logger.warning(f"{key} is already present in the asset and will be ignored.")
        else:
            original_points[key] = file_points[key]
    return list(original_points.values())


def _build_query_body(
    asset_name: Optional[str] = None,
    default_topic_path: Optional[str] = None,
    default_topic_retain: Optional[str] = None,
    description: Optional[str] = None,
    disabled: Optional[bool] = None,
    display_name: Optional[str] = None,
    documentation_uri: Optional[str] = None,
    endpoint_profile: Optional[str] = None,
    external_asset_id: Optional[str] = None,
    hardware_revision: Optional[str] = None,
    location: Optional[str] = None,
    manufacturer: Optional[str] = None,
    manufacturer_uri: Optional[str] = None,
    model: Optional[str] = None,
    product_code: Optional[str] = None,
    resource_group_name: Optional[str] = None,
    serial_number: Optional[str] = None,
    software_revision: Optional[str] = None,
) -> str:
    query_body = ""
    if resource_group_name:
        query_body += f"| where resourceGroup =~ \"{resource_group_name}\""
    if location:
        query_body += f"| where location =~ \"{location}\""
    if asset_name:
        query_body += f"| where name =~ \"{asset_name}\""
    if default_topic_path:
        query_body += f"| where properties.defaultTopic.path =~ \"{default_topic_path}\""
    if default_topic_retain:
        query_body += f"| where properties.defaultTopic.retain =~ \"{default_topic_retain}\""
    if description:
        query_body += f"| where properties.description =~ \"{description}\""
    if display_name:
        query_body += f"| where properties.displayName =~ \"{display_name}\""
    if disabled is not None:
        query_body += f"| where properties.enabled == {not disabled}"
    if documentation_uri:
        query_body += f"| where properties.documentationUri =~ \"{documentation_uri}\""
    if endpoint_profile:
        query_body += f"| where properties.assetEndpointProfileUri =~ \"{endpoint_profile}\""
    if external_asset_id:
        query_body += f"| where properties.externalAssetId =~ \"{external_asset_id}\""
    if hardware_revision:
        query_body += f"| where properties.hardwareRevision =~ \"{hardware_revision}\""
    if manufacturer:
        query_body += f"| where properties.manufacturer =~ \"{manufacturer}\""
    if manufacturer_uri:
        query_body += f"| where properties.manufacturerUri =~ \"{manufacturer_uri}\""
    if model:
        query_body += f"| where properties.model =~ \"{model}\""
    if product_code:
        query_body += f"| where properties.productCode =~ \"{product_code}\""
    if serial_number:
        query_body += f"| where properties.serialNumber =~ \"{serial_number}\""
    if software_revision:
        query_body += f"| where properties.softwareRevision =~ \"{software_revision}\""

    query_body += "| extend customLocation = tostring(extendedLocation.name) "\
        "| extend provisioningState = properties.provisioningState "\
        "| project id, customLocation, location, name, resourceGroup, provisioningState, tags, "\
        "type, subscriptionId "
    return query_body


# Helpers
def _build_asset_sub_point(
    data_source: Optional[str] = None,
    event_notifier: Optional[str] = None,
    name: Optional[str] = None,
    observability_mode: Optional[str] = None,
    queue_size: Optional[int] = None,
    sampling_interval: Optional[int] = None,
) -> Dict[str, str]:
    custom_configuration = _build_default_configuration(
        original_configuration="{}",
        sampling_interval=sampling_interval,
        queue_size=queue_size
    )
    result = {"name": name}
    observability_mode = observability_mode.capitalize() if observability_mode else "None"

    if data_source:
        result["dataSource"] = data_source
        result["dataPointConfiguration"] = custom_configuration
        if observability_mode not in VALID_DATA_OBSERVABILITY_MODES:
            raise InvalidArgumentValueError(
                INVALID_OBSERVABILITY_MODE_ERROR.format(data_source, ', '.join(VALID_DATA_OBSERVABILITY_MODES))
            )
    elif event_notifier:
        result["eventNotifier"] = event_notifier
        result["eventConfiguration"] = custom_configuration
        if observability_mode not in VALID_EVENT_OBSERVABILITY_MODES:
            raise InvalidArgumentValueError(
                INVALID_OBSERVABILITY_MODE_ERROR.format(event_notifier, ', '.join(VALID_EVENT_OBSERVABILITY_MODES))
            )

    result["observabilityMode"] = observability_mode
    return result


def _build_default_configuration(
    original_configuration: str,
    publishing_interval: Optional[int] = None,
    sampling_interval: Optional[int] = None,
    queue_size: Optional[int] = None,
) -> str:
    defaults = json.loads(original_configuration)
    if publishing_interval:
        defaults["publishingInterval"] = int(publishing_interval)
    if sampling_interval:
        defaults["samplingInterval"] = int(sampling_interval)
    if queue_size:
        defaults["queueSize"] = int(queue_size)
    return json.dumps(defaults)


def _build_ordered_csv_conversion_map(sub_point_type: str, portal_friendly: bool = False) -> Dict[str, str]:
    """Results in an ordered dict for headers"""
    from collections import OrderedDict
    csv_conversion_map = [
        ("queueSize", "QueueSize" if portal_friendly else "Queue Size"),
        ("observabilityMode", "ObservabilityMode" if portal_friendly else "Observability Mode"),
    ]
    if not portal_friendly or sub_point_type == "dataPoints":
        csv_conversion_map.append(("samplingInterval", "Sampling Interval Milliseconds"))
    if not portal_friendly:
        csv_conversion_map.append(("capabilityId", "Capability Id"))
    if sub_point_type == "dataPoints":
        csv_conversion_map.insert(0, ("dataSource", "NodeID" if portal_friendly else "Data Source"))
        csv_conversion_map.insert(1, ("name", "TagName" if portal_friendly else "Name"))
    else:
        csv_conversion_map.insert(0, ("eventNotifier", "EventNotifier" if portal_friendly else "Event Notifier"))
        csv_conversion_map.insert(1, ("name", "EventName" if portal_friendly else "Name"))

    # datasource, name, queuesize, observabilitymode, samplinginterval, capabilityid
    return OrderedDict(csv_conversion_map)


def _convert_sub_points_from_csv(sub_points: List[Dict[str, str]]):
    csv_conversion_map = {
        "CapabilityId": "capabilityId",
        "Capability Id": "capabilityId",
        "Data Source": "dataSource",
        "EventName": "name",
        "EventNotifier": "eventNotifier",
        "Event Notifier": "eventNotifier",
        "Name": "name",
        "NodeID": "dataSource",
        "ObservabilityMode": "observabilityMode",
        "Observability Mode": "observabilityMode",
        "QueueSize": "queueSize",
        "Queue Size": "queueSize",
        "Sampling Interval Milliseconds": "samplingInterval",
        "TagName" : "name",
    }
    for point in sub_points:
        # point has csv values
        point.pop("", None)
        for key, value in csv_conversion_map.items():
            if key in point:
                point[value] = point.pop(key)
        # now the point has the normal values - do some final transformations
        if point.get("observabilityMode"):
            point["observabilityMode"] = point["observabilityMode"].capitalize()
        configuration = {}
        if point.get("samplingInterval"):
            configuration["samplingInterval"] = int(point.pop("samplingInterval"))
        else:
            point.pop("samplingInterval", None)
        if point.get("queueSize"):
            configuration["queueSize"] = int(point.pop("queueSize"))
        else:
            point.pop("queueSize", None)
        if configuration:
            config_key = "dataPointConfiguration" if "dataSource" in point else "eventConfiguration"
            point[config_key] = json.dumps(configuration)


def _convert_sub_points_to_csv(
    sub_points: List[Dict[str, str]],
    sub_point_type: str,
    default_configuration: str,
    portal_friendly: bool = False
) -> List[str]:
    csv_conversion_map = _build_ordered_csv_conversion_map(sub_point_type, portal_friendly)
    default_configuration = json.loads(default_configuration) if portal_friendly else {}
    for point in sub_points:
        configuration = point.pop(f"{sub_point_type[:-1]}Configuration", "{}")
        point.update(json.loads(configuration))
        if portal_friendly:
            point.pop("capabilityId", None)
            if sub_point_type == "events":
                point.pop("samplingInterval", None)
        for asset_key, csv_key in csv_conversion_map.items():
            point[csv_key] = point.pop(asset_key, default_configuration.get(asset_key))
    return list(csv_conversion_map.values())


def _process_asset_sub_points(required_arg: str, sub_points: Optional[List[str]]) -> List[Dict[str, str]]:
    """This is for the main create/update asset commands"""
    if not sub_points:
        return []
    point_type = "Data point" if required_arg == "data_source" else "Event"
    invalid_arg = "event_notifier" if required_arg == "data_source" else "data_source"
    processed_points = []
    for point in sub_points:
        parsed_points = assemble_nargs_to_dict(point)

        if not parsed_points.get(required_arg):
            raise RequiredArgumentMissingError(f"{point_type} ({point}) is missing the {required_arg}.")
        if parsed_points.get(invalid_arg):
            raise InvalidArgumentValueError(f"{point_type} does not support {invalid_arg}.")

        processed_point = _build_asset_sub_point(**parsed_points)
        processed_points.append(processed_point)

    return processed_points


def _process_custom_attributes(current_attributes: Dict[str, str], custom_attributes: List[str]):
    custom_attributes = assemble_nargs_to_dict(custom_attributes)
    for key, value in custom_attributes.items():
        if value == "":
            current_attributes.pop(key, None)
        else:
            current_attributes[key] = value


def _update_properties(
    properties: Dict[str, Union[str, List[Dict[str, str]]]],
    custom_attributes: Optional[List[str]] = None,
    default_topic_path: Optional[str] = None,
    default_topic_retain: Optional[str] = None,
    description: Optional[str] = None,
    disabled: Optional[bool] = None,
    display_name: Optional[str] = None,
    documentation_uri: Optional[str] = None,
    external_asset_id: Optional[str] = None,
    hardware_revision: Optional[str] = None,
    manufacturer: Optional[str] = None,
    manufacturer_uri: Optional[str] = None,
    model: Optional[str] = None,
    product_code: Optional[str] = None,
    serial_number: Optional[str] = None,
    software_revision: Optional[str] = None,
    ds_publishing_interval: Optional[int] = None,
    ds_sampling_interval: Optional[int] = None,
    ds_queue_size: Optional[int] = None,
    ev_publishing_interval: Optional[int] = None,
    ev_sampling_interval: Optional[int] = None,
    ev_queue_size: Optional[int] = None,
) -> None:
    if description:
        properties["description"] = description
    if disabled is not None:
        properties["enabled"] = not disabled
    if display_name:
        properties["displayName"] = display_name
    if documentation_uri:
        properties["documentationUri"] = documentation_uri
    if external_asset_id:
        properties["externalAssetId"] = external_asset_id
    if hardware_revision:
        properties["hardwareRevision"] = hardware_revision
    if manufacturer:
        properties["manufacturer"] = manufacturer
    if manufacturer_uri:
        properties["manufacturerUri"] = manufacturer_uri
    if model:
        properties["model"] = model
    if product_code:
        properties["productCode"] = product_code
    if serial_number:
        properties["serialNumber"] = serial_number
    if software_revision:
        properties["softwareRevision"] = software_revision

    if custom_attributes:
        if "attributes" not in properties:
            properties["attributes"] = {}
        _process_custom_attributes(
            properties["attributes"], custom_attributes=custom_attributes
        )

    # Defaults
    properties["defaultDatasetsConfiguration"] = _build_default_configuration(
        original_configuration=properties.get("defaultDatasetsConfiguration", "{}"),
        publishing_interval=ds_publishing_interval,
        sampling_interval=ds_sampling_interval,
        queue_size=ds_queue_size
    )

    properties["defaultEventsConfiguration"] = _build_default_configuration(
        original_configuration=properties.get("defaultEventsConfiguration", "{}"),
        publishing_interval=ev_publishing_interval,
        sampling_interval=ev_sampling_interval,
        queue_size=ev_queue_size
    )

    # TODO: unit test
    if any([default_topic_path, default_topic_retain]):
        properties["defaultTopic"] = _build_topic(
            original_topic=properties.get("defaultTopic"),
            topic_path=default_topic_path,
            topic_retain=default_topic_retain
        )