# ============LICENSE_START====================================================
# org.onap.dcae
# =============================================================================
-# Copyright (c) 2019 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
+# Copyright (c) 2021 highstreet technologies GmbH. All rights reserved.
# =============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
import json
from jsonschema import validate
-import requests
from aoconversion import utils
-def _get_dcae_cs_schema():
- res = requests.get(
- "https://git.onap.org/dcaegen2/platform/plain/mod/component-json-schemas/component-specification/dcae-cli-v2/component-spec-schema.json"
- )
- return res.json()
-
-
def _get_format_version(target_name, data_formats):
"""
search through the data formats for name, make sure we have it, and retrieve the version
pstype = "message_router"
for method in meta["methods"]:
- df_in_name = meta["methods"][method]["input"]
+ df_in_name = utils.validate_format(meta, method, "input")
subscriber = {
"config_key": "{0}_subscriber".format(method),
"format": df_in_name,
spec["streams"]["subscribes"].append(subscriber)
- df_out_name = meta["methods"][method]["output"]
+ df_out_name = utils.validate_format(meta, method, "output")
publisher = {
"config_key": "{0}_publisher".format(method),
Returns the spec
"""
spec = _generate_spec(
- model_name, utils.get_metadata(model_repo_path, model_name), _get_dcae_cs_schema(), data_formats, docker_uri
+ model_name, utils.get_metadata(model_repo_path, model_name), utils.component_schema.get(), data_formats, docker_uri
)
fname = "{0}_dcae_component_specification.json".format(model_name)
with open("{0}/{1}".format(model_repo_path, fname), "w") as f: