Skip to content

Commit

Permalink
added metadata hash json verification and script runs inline
Browse files Browse the repository at this point in the history
  • Loading branch information
KahanMajmudar committed Nov 21, 2024
1 parent b2fe914 commit 0472c0b
Show file tree
Hide file tree
Showing 2 changed files with 216 additions and 31 deletions.
235 changes: 213 additions & 22 deletions setup_metadata_hash.py
Original file line number Diff line number Diff line change
@@ -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
)
Expand All @@ -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, "")
12 changes: 3 additions & 9 deletions utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down

0 comments on commit 0472c0b

Please sign in to comment.