diff --git a/setup_metadata_hash.py b/setup_metadata_hash.py index 82429a0..2e57332 100644 --- a/setup_metadata_hash.py +++ b/setup_metadata_hash.py @@ -1,43 +1,75 @@ -from typing import Tuple - +import sys +import json +from typing import Tuple, List, Dict import multibase import multicodec from aea.helpers.cid import to_v1 from aea_cli_ipfs.ipfs_utils import IPFSTool from utils import ( print_title, - OPERATE_HOME, MechQuickstartConfig, input_with_default_value, - OperateApp, ) -def main() -> None: +metadata_schema = { + "name": str, + "description": str, + "inputFormat": str, + "outputFormat": str, + "image": str, + "tools": List, + "toolMetadata": Dict, +} + +tool_schema = { + "name": str, + "description": str, + "input": Dict, + "output": Dict, +} +tool_input_schema = { + "type": str, + "description": str, +} +tool_output_schema = {"type": str, "description": str, "schema": Dict} + +output_schema_schema = { + "properties": Dict, + "required": List, + "type": str, +} + +properties_schema = { + "requestId": Dict, + "result": Dict, + "prompt": Dict, +} + +properties_data_schema = { + "type": str, + "description": str, +} + + +def setup_metadata_hash(mech_quickstart_config: MechQuickstartConfig) -> None: """ Push the metadata file to IPFS. """ print_title("Mech Quickstart: Metadata hash setup") - print("This script will assist you in setting up the metadata hash for your mech.") - print() - - operate = OperateApp( - home=OPERATE_HOME, - ) - operate.setup() - - path = OPERATE_HOME / "local_config.json" - if path.exists(): - mech_quickstart_config = MechQuickstartConfig.load(path) - else: - mech_quickstart_config = MechQuickstartConfig(path) metadata_hash_path = input_with_default_value( "Please provide the path to your metadata_hash.json file", "./.metadata_hash.json", ) + status, error_msg = __validate_metadata_file(metadata_hash_path) + if not status: + print(error_msg) + print("Please refer to .metadata_hash.json.example for reference") + sys.exit(1) + response = IPFSTool().client.add( metadata_hash_path, pin=True, recursive=True, wrap_with_directory=False ) @@ -47,11 +79,170 @@ def main() -> None: v1_file_hash_hex = "f01" + multihash_bytes.hex() mech_quickstart_config.metadata_hash = v1_file_hash_hex - mech_quickstart_config.store() - print() print_title("Metadata hash successfully generated and stored in config") -if __name__ == "__main__": - main() +def __validate_metadata_file(file_path) -> Tuple[bool, str]: + status = False + try: + path = file_path + with open(path, "r") as f: + metadata: Dict = json.load(f) + + except FileNotFoundError: + return (status, f"Error: Metadata file not found at {file_path}") + except json.JSONDecodeError: + return (status, "Error: Metadata file contains invalid JSON.") + + for key, expected_type in metadata_schema.items(): + if key not in metadata: + return (status, f"Missing key in metadata json: '{key}'") + + if not isinstance(metadata[key], expected_type): + expected = expected_type.__name__ + actual = type(metadata[key]).__name__ + return ( + status, + f"Invalid type for key in metadata json. Expected '{expected}', but got '{actual}'", + ) + + tools = metadata["tools"] + tools_metadata = metadata["toolMetadata"] + num_of_tools = len(tools) + num_of_tools_metadata = len(tools_metadata) + + if num_of_tools != num_of_tools_metadata: + return ( + status, + f"Number of tools does not match number of keys in 'toolMetadata'. Expected {num_of_tools} but got {num_of_tools_metadata}.", + ) + + for tool in tools: + if tool not in tools_metadata: + return (status, f"Missing toolsMetadata for tool: '{tool}'") + + for key, expected_type in tool_schema.items(): + data = tools_metadata[tool] + if key not in data: + return (status, f"Missing key in toolsMetadata: '{key}'") + + if not isinstance(data[key], expected_type): + expected = expected_type.__name__ + actual = type(data[key]).__name__ + return ( + status, + f"Invalid type for key in toolsMetadata. Expected '{expected}', but got '{actual}'", + ) + + if key == "input": + for i_key, i_expected_type in tool_input_schema.items(): + input_data = data[key] + if i_key not in input_data: + return ( + status, + f"Missing key for {tool} -> input: '{i_key}'", + ) + + if not isinstance(input_data[i_key], i_expected_type): + i_expected = i_expected_type.__name__ + i_actual = type(input_data[i_key]).__name__ + return ( + status, + f"Invalid type for '{i_key}' in {tool} -> input. Expected '{i_expected}', but got '{i_actual}'.", + ) + + elif key == "output": + for o_key, o_expected_type in tool_output_schema.items(): + output_data = data[key] + if o_key not in output_data: + return ( + status, + f"Missing key for {tool} -> output: '{o_key}'", + ) + + if not isinstance(output_data[o_key], o_expected_type): + o_expected = o_expected_type.__name__ + o_actual = type(output_data[o_key]).__name__ + return ( + status, + f"Invalid type for '{o_key}' in {tool} -> output. Expected '{o_expected}', but got '{o_actual}'.", + ) + + if o_key == "schema": + for ( + s_key, + s_expected_type, + ) in output_schema_schema.items(): + output_schema_data = output_data[o_key] + if s_key not in output_schema_data: + return ( + status, + f"Missing key for {tool} -> output -> schema: '{s_key}'", + ) + + if not isinstance( + output_schema_data[s_key], s_expected_type + ): + s_expected = s_expected_type.__name__ + s_actual = type(output_schema_data[s_key]).__name__ + return ( + status, + f"Invalid type for '{s_key}' in {tool} -> output -> schema. Expected '{s_expected}', but got '{s_actual}'.", + ) + + if ( + s_key == "properties" + and "required" in output_schema_data + ): + for ( + p_key, + p_expected_type, + ) in properties_schema.items(): + properties_data = output_schema_data[s_key] + if p_key not in properties_data: + return ( + status, + f"Missing key for {tool} -> output -> schema -> properties: '{p_key}'", + ) + + if not isinstance( + properties_data[p_key], p_expected_type + ): + p_expected = p_expected_type.__name__ + p_actual = type(properties_data[p_key]).__name__ + return ( + status, + f"Invalid type for '{p_key}' in {tool} -> output -> schema -> properties. Expected '{p_expected}', but got '{p_actual}'.", + ) + + required = output_schema_data["required"] + num_of_properties_data = len(properties_data) + num_of_required = len(required) + + if num_of_properties_data != num_of_required: + return ( + status, + f"Number of properties data does not match number of keys in 'required'. Expected {num_of_required} but got {num_of_properties_data}.", + ) + + for ( + key, + expected_type, + ) in properties_data_schema.items(): + data = properties_data[p_key] + if key not in data: + return ( + status, + f"Missing key in properties -> {p_key}: '{key}'", + ) + + if not isinstance(data[key], expected_type): + expected = expected_type.__name__ + actual = type(data[key]).__name__ + return ( + status, + f"Invalid type for key in properties. Expected '{expected}', but got '{actual}'", + ) + + return (True, "") diff --git a/utils.py b/utils.py index 88cc5e7..ed4d2ba 100644 --- a/utils.py +++ b/utils.py @@ -353,16 +353,10 @@ def get_local_config() -> MechQuickstartConfig: == "y" ) if metadata_hash: + from setup_metadata_hash import setup_metadata_hash while True: - user_input = input(f"Please enter the metadata_hash str: ") - if not isinstance(user_input, str): - print("Error: Please enter a valid str.") - continue - else: - mech_quickstart_config.metadata_hash = ( - user_input - ) - break + setup_metadata_hash(mech_quickstart_config) + break else: mech_quickstart_config.metadata_hash = ( DEFAULT_MECH_METADATA_HASH