scripts/parse_registries.py - openbmc/bmcweb - Gitiles

 #!/usr/bin/env python3
 import argparse
 import json
 import os
 import typing as t
 from collections import OrderedDict

 import requests

 PRAGMA_ONCE: t.Final[
     str
 ] = """#pragma once
 """

 WARNING = """/****************************************************************
  *                 READ THIS WARNING FIRST
  * This is an auto-generated header which contains definitions
  * for Redfish DMTF defined messages.
  * DO NOT modify this registry outside of running the
  * parse_registries.py script.  The definitions contained within
  * this file are owned by DMTF.  Any modifications to these files
  * should be first pushed to the relevant registry in the DMTF
  * github organization.
  ***************************************************************/"""

 COPYRIGHT: t.Final[
     str
 ] = """// SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright OpenBMC Authors
 """

 INCLUDES = """
 #include "registries.hpp"

 #include <array>

 // clang-format off

 namespace redfish::registries::{}
 {{
 """

 REGISTRY_HEADER: t.Final[str] = f"{COPYRIGHT}{PRAGMA_ONCE}{WARNING}{INCLUDES}"

 SCRIPT_DIR: t.Final[str] = os.path.dirname(os.path.realpath(__file__))

 INCLUDE_PATH: t.Final[str] = os.path.realpath(
     os.path.join(SCRIPT_DIR, "..", "redfish-core", "include", "registries")
 )

 PROXIES: t.Final[t.Dict[str, str]] = {
     "https": os.environ.get("https_proxy", "")
 }

 RegistryInfo: t.TypeAlias = t.Tuple[str, t.Dict[str, t.Any], str, str]


 def make_getter(
     dmtf_name: str, header_name: str, type_name: str
 ) -> RegistryInfo:
     url = "https://redfish.dmtf.org/registries/{}".format(dmtf_name)
     dmtf = requests.get(url, proxies=PROXIES)
     dmtf.raise_for_status()
     json_file = json.loads(dmtf.text, object_pairs_hook=OrderedDict)
     path = os.path.join(INCLUDE_PATH, header_name)
     return (path, json_file, type_name, url)


 def openbmc_local_getter() -> RegistryInfo:
     url = "https://raw.githubusercontent.com/openbmc/bmcweb/refs/heads/master/redfish-core/include/registries/openbmc.json"
     with open(
         os.path.join(
             SCRIPT_DIR,
             "..",
             "redfish-core",
             "include",
             "registries",
             "openbmc.json",
         ),
         "rb",
     ) as json_file_fd:
         json_file = json.load(json_file_fd)

     path = os.path.join(INCLUDE_PATH, "openbmc_message_registry.hpp")
     return (path, json_file, "openbmc", url)


 def update_registries(files: t.List[RegistryInfo]) -> None:
     # Remove the old files
     for file, json_dict, namespace, url in files:
         try:
             os.remove(file)
         except BaseException:
             print("{} not found".format(file))

         with open(file, "w") as registry:
             version_split = json_dict["RegistryVersion"].split(".")

             registry.write(REGISTRY_HEADER.format(namespace))
             # Parse the Registry header info
             registry.write(
                 "const Header header = {{\n"
                 '    "{json_dict[@Redfish.Copyright]}",\n'
                 '    "{json_dict[@odata.type]}",\n'
                 "    {version_split[0]},\n"
                 "    {version_split[1]},\n"
                 "    {version_split[2]},\n"
                 '    "{json_dict[Name]}",\n'
                 '    "{json_dict[Language]}",\n'
                 '    "{json_dict[Description]}",\n'
                 '    "{json_dict[RegistryPrefix]}",\n'
                 '    "{json_dict[OwningEntity]}",\n'
                 "}};\n"
                 "constexpr const char* url =\n"
                 '    "{url}";\n'
                 "\n"
                 "constexpr std::array registry =\n"
                 "{{\n".format(
                     json_dict=json_dict,
                     url=url,
                     version_split=version_split,
                 )
             )

             messages_sorted = sorted(json_dict["Messages"].items())
             for messageId, message in messages_sorted:
                 registry.write(
                     "    MessageEntry{{\n"
                     '        "{messageId}",\n'
                     "        {{\n"
                     '            "{message[Description]}",\n'
                     '            "{message[Message]}",\n'
                     '            "{message[MessageSeverity]}",\n'
                     "            {message[NumberOfArgs]},\n"
                     "            {{".format(
                         messageId=messageId, message=message
                     )
                 )
                 paramTypes = message.get("ParamTypes")
                 if paramTypes:
                     for paramType in paramTypes:
                         registry.write(
                             '\n                "{}",'.format(paramType)
                         )
                     registry.write("\n            },\n")
                 else:
                     registry.write("},\n")
                 registry.write(
                     '            "{message[Resolution]}",\n        }}}},\n'.format(
                         message=message
                     )
                 )

             registry.write("\n};\n\nenum class Index\n{\n")
             for index, (messageId, message) in enumerate(messages_sorted):
                 messageId = messageId[0].lower() + messageId[1:]
                 registry.write("    {} = {},\n".format(messageId, index))
             registry.write(
                 "}};\n}} // namespace redfish::registries::{}\n".format(
                     namespace
                 )
             )


 def get_privilege_string_from_list(
     privilege_list: t.List[t.Dict[str, t.Any]],
 ) -> str:
     privilege_string = "{{\n"
     for privilege_json in privilege_list:
         privileges = privilege_json["Privilege"]
         privilege_string += "    {"
         last_privelege = ""
         for privilege in privileges:
             last_privelege = privilege
             if privilege == "NoAuth":
                 continue
             privilege_string += '"'
             privilege_string += privilege
             privilege_string += '",\n'
         if last_privelege != "NoAuth":
             privilege_string = privilege_string[:-2]
         privilege_string += "}"
         privilege_string += ",\n"
     privilege_string = privilege_string[:-2]
     privilege_string += "\n}}"
     return privilege_string


 def get_variable_name_for_privilege_set(
     privilege_list: t.List[t.Dict[str, t.Any]],
 ) -> str:
     names = []
     for privilege_json in privilege_list:
         privileges = privilege_json["Privilege"]
         names.append("And".join(privileges))
     return "Or".join(names)


 PRIVILEGE_HEADER = (
     COPYRIGHT
     + PRAGMA_ONCE
     + WARNING
     + """
 #include "privileges.hpp"

 #include <array>

 // clang-format off

 namespace redfish::privileges
 {
 """
 )


 def get_response_code(entry_id: str) -> str | None:
     codes = {
         "AccessDenied": "forbidden",
         "AccountForSessionNoLongerExists": "forbidden",
         "AccountModified": "ok",
         "AccountRemoved": "ok",
         "CouldNotEstablishConnection": "not_found",
         "Created": "created",
         "EventSubscriptionLimitExceeded": "service_unavailable",
         "GeneralError": "internal_server_error",
         "GenerateSecretKeyRequired": "forbidden",
         "InsufficientPrivilege": "forbidden",
         "InsufficientStorage": "insufficient_storage",
         "InternalError": "internal_server_error",
         "MaximumErrorsExceeded": "internal_server_error",
         "NoValidSession": "forbidden",
         "OperationFailed": "bad_gateway",
         "OperationNotAllowed": "method_not_allowed",
         "OperationTimeout": "internal_server_error",
         "PasswordChangeRequired": None,
         "PreconditionFailed": "precondition_failed",
         "PropertyNotWritable": "forbidden",
         "PropertyValueExternalConflict": "conflict",
         "PropertyValueModified": "ok",
         "PropertyValueResourceConflict": "conflict",
         "ResourceAtUriUnauthorized": "unauthorized",
         "ResourceCannotBeDeleted": "method_not_allowed",
         "ResourceExhaustion": "service_unavailable",
         "ResourceInStandby": "service_unavailable",
         "ResourceInUse": "service_unavailable",
         "ResourceNotFound": "not_found",
         "RestrictedRole": "forbidden",
         "ServiceDisabled": "service_unavailable",
         "ServiceInUnknownState": "service_unavailable",
         "ServiceShuttingDown": "service_unavailable",
         "ServiceTemporarilyUnavailable": "service_unavailable",
         "SessionLimitExceeded": "service_unavailable",
         "SessionTerminated": "ok",
         "SubscriptionTerminated": "ok",
         "Success": None,
     }

     return codes.get(entry_id, "bad_request")


 def make_error_function(
     entry_id: str,
     entry: t.Dict[str, t.Any],
     is_header: bool,
     registry_name: str,
     namespace_name: str,
 ) -> str:
     arg_nonstring_types = {
         "const boost::urls::url_view_base&": {
             "AccessDenied": [1],
             "CouldNotEstablishConnection": [1],
             "GenerateSecretKeyRequired": [1],
             "InvalidObject": [1],
             "PasswordChangeRequired": [1],
             "PropertyValueResourceConflict": [3],
             "ResetRequired": [1],
             "ResourceAtUriInUnknownFormat": [1],
             "ResourceAtUriUnauthorized": [1],
             "ResourceCreationConflict": [1],
             "ResourceMissingAtURI": [1],
             "SourceDoesNotSupportProtocol": [1],
         },
         "const nlohmann::json&": {
             "ActionParameterValueError": [1],
             "ActionParameterValueFormatError": [1],
             "ActionParameterValueTypeError": [1],
             "PropertyValueExternalConflict": [2],
             "PropertyValueFormatError": [1],
             "PropertyValueIncorrect": [2],
             "PropertyValueModified": [2],
             "PropertyValueNotInList": [1],
             "PropertyValueOutOfRange": [1],
             "PropertyValueResourceConflict": [2],
             "PropertyValueTypeError": [1],
             "QueryParameterValueFormatError": [1],
             "QueryParameterValueTypeError": [1],
         },
         "uint64_t": {
             "ArraySizeTooLong": [2],
             "InvalidIndex": [1],
             "StringValueTooLong": [2],
             "TaskProgressChanged": [2],
         },
     }

     out = ""
     args = []
     argtypes = []
     for arg_index, arg in enumerate(entry.get("ParamTypes", [])):
         arg_index += 1
         typename = "std::string_view"
         for typestring, entries in arg_nonstring_types.items():
             if arg_index in entries.get(entry_id, []):
                 typename = typestring

         argtypes.append(typename)
         args.append(f"{typename} arg{arg_index}")
     function_name = entry_id[0].lower() + entry_id[1:]
     arg = ", ".join(args)
     out += f"nlohmann::json {function_name}({arg})"

     if is_header:
         out += ";\n\n"
     else:
         out += "\n{\n"
         to_array_type = ""
         arg_param = "{}"
         if argtypes:
             outargs = []
             for index, typename in enumerate(argtypes):
                 index += 1
                 if typename == "const nlohmann::json&":
                     out += f"std::string arg{index}Str = arg{index}.dump(-1, ' ', true, nlohmann::json::error_handler_t::replace);\n"
                 elif typename == "uint64_t":
                     out += f"std::string arg{index}Str = std::to_string(arg{index});\n"

             for index, typename in enumerate(argtypes):
                 index += 1
                 if typename == "const boost::urls::url_view_base&":
                     outargs.append(f"arg{index}.buffer()")
                     to_array_type = "<std::string_view>"
                 elif typename == "const nlohmann::json&":
                     outargs.append(f"arg{index}Str")
                     to_array_type = "<std::string_view>"
                 elif typename == "uint64_t":
                     outargs.append(f"arg{index}Str")
                     to_array_type = "<std::string_view>"
                 else:
                     outargs.append(f"arg{index}")
             argstring = ", ".join(outargs)
             arg_param = f"std::to_array{to_array_type}({{{argstring}}})"
         out += f"    return getLog(redfish::registries::{namespace_name}::Index::{function_name}, {arg_param});"
         out += "\n}\n\n"
     if registry_name == "Base":
         args.insert(0, "crow::Response& res")
         if entry_id == "InternalError":
             if is_header:
                 args.append(
                     "std::source_location location = std::source_location::current()"
                 )
             else:
                 args.append("const std::source_location location")
         arg = ", ".join(args)
         out += f"void {function_name}({arg})"
         if is_header:
             out += ";\n"
         else:
             out += "\n{\n"
             if entry_id == "InternalError":
                 out += """BMCWEB_LOG_CRITICAL("Internal Error {}({}:{}) `{}`: ", location.file_name(),
                             location.line(), location.column(),
                             location.function_name());\n"""

             if entry_id == "ServiceTemporarilyUnavailable":
                 out += "res.addHeader(boost::beast::http::field::retry_after, arg1);"

             res = get_response_code(entry_id)
             if res:
                 out += f"    res.result(boost::beast::http::status::{res});\n"
             args_out = ", ".join([f"arg{x + 1}" for x in range(len(argtypes))])

             addMessageToJson = {
                 "PropertyDuplicate": 1,
                 "ResourceAlreadyExists": 2,
                 "CreateFailedMissingReqProperties": 1,
                 "PropertyValueFormatError": 2,
                 "PropertyValueNotInList": 2,
                 "PropertyValueTypeError": 2,
                 "PropertyValueError": 1,
                 "PropertyNotWritable": 1,
                 "PropertyValueModified": 1,
                 "PropertyMissing": 1,
             }

             addMessageToRoot = [
                 "SessionTerminated",
                 "SubscriptionTerminated",
                 "AccountRemoved",
                 "Created",
                 "Success",
                 "PasswordChangeRequired",
             ]

             if entry_id in addMessageToJson:
                 out += f"    addMessageToJson(res.jsonValue, {function_name}({args_out}), arg{addMessageToJson[entry_id]});\n"
             elif entry_id in addMessageToRoot:
                 out += f"    addMessageToJsonRoot(res.jsonValue, {function_name}({args_out}));\n"
             else:
                 out += f"    addMessageToErrorJson(res.jsonValue, {function_name}({args_out}));\n"
             out += "}\n"
     out += "\n"
     return out


 def create_error_registry(
     registry_info: RegistryInfo,
     registry_name: str,
     namespace_name: str,
     filename: str,
 ) -> None:
     file, json_dict, namespace, url = registry_info
     base_filename = filename + "_messages"

     error_messages_hpp = os.path.join(
         SCRIPT_DIR, "..", "redfish-core", "include", f"{base_filename}.hpp"
     )
     messages = json_dict["Messages"]

     with open(
         error_messages_hpp,
         "w",
     ) as out:
         out.write(PRAGMA_ONCE)
         out.write(WARNING)
         out.write(
             """
 // These generated headers are a superset of what is needed.
 // clang sees them as an error, so ignore
 // NOLINTBEGIN(misc-include-cleaner)
 #include "http_response.hpp"

 #include <boost/url/url_view_base.hpp>
 #include <nlohmann/json.hpp>

 #include <cstdint>
 #include <source_location>
 #include <string_view>
 // NOLINTEND(misc-include-cleaner)

 namespace redfish
 {

 namespace messages
 {
 """
         )
         for entry_id, entry in messages.items():
             message = entry["Message"]
             for index in range(1, 10):
                 message = message.replace(f"'%{index}'", f"<arg{index}>")
                 message = message.replace(f"%{index}", f"<arg{index}>")

             if registry_name == "Base":
                 out.write("/**\n")
                 out.write(f"* @brief Formats {entry_id} message into JSON\n")
                 out.write(f'* Message body: "{message}"\n')
                 out.write("*\n")
                 arg_index = 0
                 for arg_index, arg in enumerate(entry.get("ParamTypes", [])):
                     arg_index += 1

                     out.write(
                         f"* @param[in] arg{arg_index} Parameter of message that will replace %{arg_index} in its body.\n"
                     )
                 out.write("*\n")
                 out.write(
                     f"* @returns Message {entry_id} formatted to JSON */\n"
                 )

             out.write(
                 make_error_function(
                     entry_id, entry, True, registry_name, namespace_name
                 )
             )
         out.write("    }\n")
         out.write("}\n")

     error_messages_cpp = os.path.join(
         SCRIPT_DIR, "..", "redfish-core", "src", f"{base_filename}.cpp"
     )
     with open(
         error_messages_cpp,
         "w",
     ) as out:
         out.write(WARNING)
         out.write(f'\n#include "{base_filename}.hpp"\n')
         headers = []

         headers.append('"registries.hpp"')
         if registry_name == "Base":
             reg_name_lower = "base"
             headers.append('"error_message_utils.hpp"')
             headers.append('"http_response.hpp"')
             headers.append('"logging.hpp"')
             headers.append("<boost/beast/http/field.hpp>")
             headers.append("<boost/beast/http/status.hpp>")
             headers.append("<boost/url/url_view_base.hpp>")
             headers.append("<source_location>")
         else:
             reg_name_lower = namespace_name.lower()
         headers.append(f'"registries/{reg_name_lower}_message_registry.hpp"')

         headers.append("<nlohmann/json.hpp>")
         headers.append("<array>")
         headers.append("<cstddef>")
         headers.append("<span>")

         if registry_name not in ("ResourceEvent", "HeartbeatEvent"):
             headers.append("<cstdint>")
             headers.append("<string>")
         headers.append("<string_view>")

         for header in headers:
             out.write(f"#include {header}\n")

         out.write(
             """
 // Clang can't seem to decide whether this header needs to be included or not,
 // and is inconsistent.  Include it for now
 // NOLINTNEXTLINE(misc-include-cleaner)
 #include <utility>

 namespace redfish
 {

 namespace messages
 {
 """
         )
         out.write(
             """
 static nlohmann::json getLog(redfish::registries::{namespace_name}::Index name,
                              std::span<const std::string_view> args)
 {{
     size_t index = static_cast<size_t>(name);
     if (index >= redfish::registries::{namespace_name}::registry.size())
     {{
         return {{}};
     }}
     return getLogFromRegistry(redfish::registries::{namespace_name}::header,
                               redfish::registries::{namespace_name}::registry, index, args);
 }}

 """.format(
                 namespace_name=namespace_name
             )
         )
         for entry_id, entry in messages.items():
             out.write(
                 f"""/**
  * @internal
  * @brief Formats {entry_id} message into JSON
  *
  * See header file for more information
  * @endinternal
  */
 """
             )
             message = entry["Message"]
             out.write(
                 make_error_function(
                     entry_id, entry, False, registry_name, namespace_name
                 )
             )

         out.write("    }\n")
         out.write("}\n")
     os.system(f"clang-format -i {error_messages_hpp} {error_messages_cpp}")


 def make_privilege_registry() -> None:
     path, json_file, type_name, url = make_getter(
         "Redfish_1.5.0_PrivilegeRegistry.json",
         "privilege_registry.hpp",
         "privilege",
     )
     with open(path, "w") as registry:
         registry.write(PRIVILEGE_HEADER)

         privilege_dict: t.Dict[str, t.Tuple[t.Any, str | None]] = {}
         for mapping in json_file["Mappings"]:
             # first pass, identify all the unique privilege sets
             for operation, privilege_list in mapping["OperationMap"].items():
                 privilege_dict[
                     get_privilege_string_from_list(privilege_list)
                 ] = (
                     privilege_list,
                     None,
                 )
         for index, key in enumerate(privilege_dict):
             (privilege_list, _) = privilege_dict[key]
             name = get_variable_name_for_privilege_set(privilege_list)
             registry.write(
                 "const std::array<Privileges, {length}> "
                 "privilegeSet{name} = {key};\n".format(
                     length=len(privilege_list), name=name, key=key
                 )
             )
             privilege_dict[key] = (privilege_list, name)

         for mapping in json_file["Mappings"]:
             entity = mapping["Entity"]
             registry.write("// {}\n".format(entity))
             for operation, privilege_list in mapping["OperationMap"].items():
                 privilege_string = get_privilege_string_from_list(
                     privilege_list
                 )
                 operation = operation.lower()

                 registry.write(
                     "const static auto& {}{} = privilegeSet{};\n".format(
                         operation, entity, privilege_dict[privilege_string][1]
                     )
                 )
             registry.write("\n")
         registry.write(
             "} // namespace redfish::privileges\n// clang-format on\n"
         )


 def to_pascal_case(text: str) -> str:
     s = text.replace("_", " ")
     s1 = s.split()
     if len(text) == 0:
         return text
     return "".join(i.capitalize() for i in s1[0:])


 def main() -> None:
     dmtf_registries = OrderedDict(
         [
             ("base", "1.19.0"),
             ("composition", "1.1.2"),
             ("environmental", "1.1.0"),
             ("ethernet_fabric", "1.0.1"),
             ("fabric", "1.0.2"),
             ("heartbeat_event", "1.0.1"),
             ("job_event", "1.0.1"),
             ("license", "1.0.3"),
             ("log_service", "1.0.1"),
             ("network_device", "1.0.3"),
             ("platform", "1.0.1"),
             ("power", "1.0.1"),
             ("resource_event", "1.3.0"),
             ("sensor_event", "1.0.1"),
             ("storage_device", "1.2.1"),
             ("task_event", "1.0.3"),
             ("telemetry", "1.0.0"),
             ("update", "1.0.2"),
         ]
     )

     parser = argparse.ArgumentParser()
     parser.add_argument(
         "--registries",
         type=str,
         default="privilege,openbmc,"
         + ",".join([dmtf for dmtf in dmtf_registries]),
         help="Comma delimited list of registries to update",
     )

     args = parser.parse_args()

     registries = set(args.registries.split(","))
     registries_map: t.OrderedDict[str, RegistryInfo] = OrderedDict()

     for registry, version in dmtf_registries.items():
         if registry in registries:
             registry_pascal_case = to_pascal_case(registry)
             registries_map[registry] = make_getter(
                 f"{registry_pascal_case}.{version}.json",
                 f"{registry}_message_registry.hpp",
                 registry,
             )
     if "openbmc" in registries:
         registries_map["openbmc"] = openbmc_local_getter()

     update_registries(list(registries_map.values()))

     if "base" in registries_map:
         create_error_registry(
             registries_map["base"],
             "Base",
             "base",
             "error",
         )
     if "heartbeat_event" in registries_map:
         create_error_registry(
             registries_map["heartbeat_event"],
             "HeartbeatEvent",
             "heartbeat_event",
             "heartbeat",
         )
     if "resource_event" in registries_map:
         create_error_registry(
             registries_map["resource_event"],
             "ResourceEvent",
             "resource_event",
             "resource",
         )
     if "task_event" in registries_map:
         create_error_registry(
             registries_map["task_event"],
             "TaskEvent",
             "task_event",
             "task",
         )

     if "privilege" in registries:
         make_privilege_registry()


 if __name__ == "__main__":
     main()
	#!/usr/bin/env python3
	import argparse
	import json
	import os
	import typing as t
	from collections import OrderedDict

	import requests

	PRAGMA_ONCE: t.Final[
	str
	] = """#pragma once
	"""

	WARNING = """/****************************************************************
	* READ THIS WARNING FIRST
	* This is an auto-generated header which contains definitions
	* for Redfish DMTF defined messages.
	* DO NOT modify this registry outside of running the
	* parse_registries.py script. The definitions contained within
	* this file are owned by DMTF. Any modifications to these files
	* should be first pushed to the relevant registry in the DMTF
	* github organization.
	***************************************************************/"""

	COPYRIGHT: t.Final[
	str
	] = """// SPDX-License-Identifier: Apache-2.0
	// SPDX-FileCopyrightText: Copyright OpenBMC Authors
	"""

	INCLUDES = """
	#include "registries.hpp"

	#include <array>

	// clang-format off

	namespace redfish::registries::{}
	{{
	"""

	REGISTRY_HEADER: t.Final[str] = f"{COPYRIGHT}{PRAGMA_ONCE}{WARNING}{INCLUDES}"

	SCRIPT_DIR: t.Final[str] = os.path.dirname(os.path.realpath(__file__))

	INCLUDE_PATH: t.Final[str] = os.path.realpath(
	os.path.join(SCRIPT_DIR, "..", "redfish-core", "include", "registries")
	)

	PROXIES: t.Final[t.Dict[str, str]] = {
	"https": os.environ.get("https_proxy", "")
	}

	RegistryInfo: t.TypeAlias = t.Tuple[str, t.Dict[str, t.Any], str, str]


	def make_getter(
	dmtf_name: str, header_name: str, type_name: str
	) -> RegistryInfo:
	url = "https://redfish.dmtf.org/registries/{}".format(dmtf_name)
	dmtf = requests.get(url, proxies=PROXIES)
	dmtf.raise_for_status()
	json_file = json.loads(dmtf.text, object_pairs_hook=OrderedDict)
	path = os.path.join(INCLUDE_PATH, header_name)
	return (path, json_file, type_name, url)


	def openbmc_local_getter() -> RegistryInfo:
	url = "https://raw.githubusercontent.com/openbmc/bmcweb/refs/heads/master/redfish-core/include/registries/openbmc.json"
	with open(
	os.path.join(
	SCRIPT_DIR,
	"..",
	"redfish-core",
	"include",
	"registries",
	"openbmc.json",
	),
	"rb",
	) as json_file_fd:
	json_file = json.load(json_file_fd)

	path = os.path.join(INCLUDE_PATH, "openbmc_message_registry.hpp")
	return (path, json_file, "openbmc", url)


	def update_registries(files: t.List[RegistryInfo]) -> None:
	# Remove the old files
	for file, json_dict, namespace, url in files:
	try:
	os.remove(file)
	except BaseException:
	print("{} not found".format(file))

	with open(file, "w") as registry:
	version_split = json_dict["RegistryVersion"].split(".")

	registry.write(REGISTRY_HEADER.format(namespace))
	# Parse the Registry header info
	registry.write(
	"const Header header = {{\n"
	' "{json_dict[@Redfish.Copyright]}",\n'
	' "{json_dict[@odata.type]}",\n'
	" {version_split[0]},\n"
	" {version_split[1]},\n"
	" {version_split[2]},\n"
	' "{json_dict[Name]}",\n'
	' "{json_dict[Language]}",\n'
	' "{json_dict[Description]}",\n'
	' "{json_dict[RegistryPrefix]}",\n'
	' "{json_dict[OwningEntity]}",\n'
	"}};\n"
	"constexpr const char* url =\n"
	' "{url}";\n'
	"\n"
	"constexpr std::array registry =\n"
	"{{\n".format(
	json_dict=json_dict,
	url=url,
	version_split=version_split,
	)
	)

	messages_sorted = sorted(json_dict["Messages"].items())
	for messageId, message in messages_sorted:
	registry.write(
	" MessageEntry{{\n"
	' "{messageId}",\n'
	" {{\n"
	' "{message[Description]}",\n'
	' "{message[Message]}",\n'
	' "{message[MessageSeverity]}",\n'
	" {message[NumberOfArgs]},\n"
	" {{".format(
	messageId=messageId, message=message
	)
	)
	paramTypes = message.get("ParamTypes")
	if paramTypes:
	for paramType in paramTypes:
	registry.write(
	'\n "{}",'.format(paramType)
	)
	registry.write("\n },\n")
	else:
	registry.write("},\n")
	registry.write(
	' "{message[Resolution]}",\n }}}},\n'.format(
	message=message
	)
	)

	registry.write("\n};\n\nenum class Index\n{\n")
	for index, (messageId, message) in enumerate(messages_sorted):
	messageId = messageId[0].lower() + messageId[1:]
	registry.write(" {} = {},\n".format(messageId, index))
	registry.write(
	"}};\n}} // namespace redfish::registries::{}\n".format(
	namespace
	)
	)


	def get_privilege_string_from_list(
	privilege_list: t.List[t.Dict[str, t.Any]],
	) -> str:
	privilege_string = "{{\n"
	for privilege_json in privilege_list:
	privileges = privilege_json["Privilege"]
	privilege_string += " {"
	last_privelege = ""
	for privilege in privileges:
	last_privelege = privilege
	if privilege == "NoAuth":
	continue
	privilege_string += '"'
	privilege_string += privilege
	privilege_string += '",\n'
	if last_privelege != "NoAuth":
	privilege_string = privilege_string[:-2]
	privilege_string += "}"
	privilege_string += ",\n"
	privilege_string = privilege_string[:-2]
	privilege_string += "\n}}"
	return privilege_string


	def get_variable_name_for_privilege_set(
	privilege_list: t.List[t.Dict[str, t.Any]],
	) -> str:
	names = []
	for privilege_json in privilege_list:
	privileges = privilege_json["Privilege"]
	names.append("And".join(privileges))
	return "Or".join(names)


	PRIVILEGE_HEADER = (
	COPYRIGHT
	+ PRAGMA_ONCE
	+ WARNING
	+ """
	#include "privileges.hpp"

	#include <array>

	// clang-format off

	namespace redfish::privileges
	{
	"""
	)


	def get_response_code(entry_id: str) -> str \| None:
	codes = {
	"AccessDenied": "forbidden",
	"AccountForSessionNoLongerExists": "forbidden",
	"AccountModified": "ok",
	"AccountRemoved": "ok",
	"CouldNotEstablishConnection": "not_found",
	"Created": "created",
	"EventSubscriptionLimitExceeded": "service_unavailable",
	"GeneralError": "internal_server_error",
	"GenerateSecretKeyRequired": "forbidden",
	"InsufficientPrivilege": "forbidden",
	"InsufficientStorage": "insufficient_storage",
	"InternalError": "internal_server_error",
	"MaximumErrorsExceeded": "internal_server_error",
	"NoValidSession": "forbidden",
	"OperationFailed": "bad_gateway",
	"OperationNotAllowed": "method_not_allowed",
	"OperationTimeout": "internal_server_error",
	"PasswordChangeRequired": None,
	"PreconditionFailed": "precondition_failed",
	"PropertyNotWritable": "forbidden",
	"PropertyValueExternalConflict": "conflict",
	"PropertyValueModified": "ok",
	"PropertyValueResourceConflict": "conflict",
	"ResourceAtUriUnauthorized": "unauthorized",
	"ResourceCannotBeDeleted": "method_not_allowed",
	"ResourceExhaustion": "service_unavailable",
	"ResourceInStandby": "service_unavailable",
	"ResourceInUse": "service_unavailable",
	"ResourceNotFound": "not_found",
	"RestrictedRole": "forbidden",
	"ServiceDisabled": "service_unavailable",
	"ServiceInUnknownState": "service_unavailable",
	"ServiceShuttingDown": "service_unavailable",
	"ServiceTemporarilyUnavailable": "service_unavailable",
	"SessionLimitExceeded": "service_unavailable",
	"SessionTerminated": "ok",
	"SubscriptionTerminated": "ok",
	"Success": None,
	}

	return codes.get(entry_id, "bad_request")


	def make_error_function(
	entry_id: str,
	entry: t.Dict[str, t.Any],
	is_header: bool,
	registry_name: str,
	namespace_name: str,
	) -> str:
	arg_nonstring_types = {
	"const boost::urls::url_view_base&": {
	"AccessDenied": [1],
	"CouldNotEstablishConnection": [1],
	"GenerateSecretKeyRequired": [1],
	"InvalidObject": [1],
	"PasswordChangeRequired": [1],
	"PropertyValueResourceConflict": [3],
	"ResetRequired": [1],
	"ResourceAtUriInUnknownFormat": [1],
	"ResourceAtUriUnauthorized": [1],
	"ResourceCreationConflict": [1],
	"ResourceMissingAtURI": [1],
	"SourceDoesNotSupportProtocol": [1],
	},
	"const nlohmann::json&": {
	"ActionParameterValueError": [1],
	"ActionParameterValueFormatError": [1],
	"ActionParameterValueTypeError": [1],
	"PropertyValueExternalConflict": [2],
	"PropertyValueFormatError": [1],
	"PropertyValueIncorrect": [2],
	"PropertyValueModified": [2],
	"PropertyValueNotInList": [1],
	"PropertyValueOutOfRange": [1],
	"PropertyValueResourceConflict": [2],
	"PropertyValueTypeError": [1],
	"QueryParameterValueFormatError": [1],
	"QueryParameterValueTypeError": [1],
	},
	"uint64_t": {
	"ArraySizeTooLong": [2],
	"InvalidIndex": [1],
	"StringValueTooLong": [2],
	"TaskProgressChanged": [2],
	},
	}

	out = ""
	args = []
	argtypes = []
	for arg_index, arg in enumerate(entry.get("ParamTypes", [])):
	arg_index += 1
	typename = "std::string_view"
	for typestring, entries in arg_nonstring_types.items():
	if arg_index in entries.get(entry_id, []):
	typename = typestring

	argtypes.append(typename)
	args.append(f"{typename} arg{arg_index}")
	function_name = entry_id[0].lower() + entry_id[1:]
	arg = ", ".join(args)
	out += f"nlohmann::json {function_name}({arg})"

	if is_header:
	out += ";\n\n"
	else:
	out += "\n{\n"
	to_array_type = ""
	arg_param = "{}"
	if argtypes:
	outargs = []
	for index, typename in enumerate(argtypes):
	index += 1
	if typename == "const nlohmann::json&":
	out += f"std::string arg{index}Str = arg{index}.dump(-1, ' ', true, nlohmann::json::error_handler_t::replace);\n"
	elif typename == "uint64_t":
	out += f"std::string arg{index}Str = std::to_string(arg{index});\n"

	for index, typename in enumerate(argtypes):
	index += 1
	if typename == "const boost::urls::url_view_base&":
	outargs.append(f"arg{index}.buffer()")
	to_array_type = "<std::string_view>"
	elif typename == "const nlohmann::json&":
	outargs.append(f"arg{index}Str")
	to_array_type = "<std::string_view>"
	elif typename == "uint64_t":
	outargs.append(f"arg{index}Str")
	to_array_type = "<std::string_view>"
	else:
	outargs.append(f"arg{index}")
	argstring = ", ".join(outargs)
	arg_param = f"std::to_array{to_array_type}({{{argstring}}})"
	out += f" return getLog(redfish::registries::{namespace_name}::Index::{function_name}, {arg_param});"
	out += "\n}\n\n"
	if registry_name == "Base":
	args.insert(0, "crow::Response& res")
	if entry_id == "InternalError":
	if is_header:
	args.append(
	"std::source_location location = std::source_location::current()"
	)
	else:
	args.append("const std::source_location location")
	arg = ", ".join(args)
	out += f"void {function_name}({arg})"
	if is_header:
	out += ";\n"
	else:
	out += "\n{\n"
	if entry_id == "InternalError":
	out += """BMCWEB_LOG_CRITICAL("Internal Error {}({}:{}) `{}`: ", location.file_name(),
	location.line(), location.column(),
	location.function_name());\n"""

	if entry_id == "ServiceTemporarilyUnavailable":
	out += "res.addHeader(boost::beast::http::field::retry_after, arg1);"

	res = get_response_code(entry_id)
	if res:
	out += f" res.result(boost::beast::http::status::{res});\n"
	args_out = ", ".join([f"arg{x + 1}" for x in range(len(argtypes))])

	addMessageToJson = {
	"PropertyDuplicate": 1,
	"ResourceAlreadyExists": 2,
	"CreateFailedMissingReqProperties": 1,
	"PropertyValueFormatError": 2,
	"PropertyValueNotInList": 2,
	"PropertyValueTypeError": 2,
	"PropertyValueError": 1,
	"PropertyNotWritable": 1,
	"PropertyValueModified": 1,
	"PropertyMissing": 1,
	}

	addMessageToRoot = [
	"SessionTerminated",
	"SubscriptionTerminated",
	"AccountRemoved",
	"Created",
	"Success",
	"PasswordChangeRequired",
	]

	if entry_id in addMessageToJson:
	out += f" addMessageToJson(res.jsonValue, {function_name}({args_out}), arg{addMessageToJson[entry_id]});\n"
	elif entry_id in addMessageToRoot:
	out += f" addMessageToJsonRoot(res.jsonValue, {function_name}({args_out}));\n"
	else:
	out += f" addMessageToErrorJson(res.jsonValue, {function_name}({args_out}));\n"
	out += "}\n"
	out += "\n"
	return out


	def create_error_registry(
	registry_info: RegistryInfo,
	registry_name: str,
	namespace_name: str,
	filename: str,
	) -> None:
	file, json_dict, namespace, url = registry_info
	base_filename = filename + "_messages"

	error_messages_hpp = os.path.join(
	SCRIPT_DIR, "..", "redfish-core", "include", f"{base_filename}.hpp"
	)
	messages = json_dict["Messages"]

	with open(
	error_messages_hpp,
	"w",
	) as out:
	out.write(PRAGMA_ONCE)
	out.write(WARNING)
	out.write(
	"""
	// These generated headers are a superset of what is needed.
	// clang sees them as an error, so ignore
	// NOLINTBEGIN(misc-include-cleaner)
	#include "http_response.hpp"

	#include <boost/url/url_view_base.hpp>
	#include <nlohmann/json.hpp>

	#include <cstdint>
	#include <source_location>
	#include <string_view>
	// NOLINTEND(misc-include-cleaner)

	namespace redfish
	{

	namespace messages
	{
	"""
	)
	for entry_id, entry in messages.items():
	message = entry["Message"]
	for index in range(1, 10):
	message = message.replace(f"'%{index}'", f"<arg{index}>")
	message = message.replace(f"%{index}", f"<arg{index}>")

	if registry_name == "Base":
	out.write("/**\n")
	out.write(f"* @brief Formats {entry_id} message into JSON\n")
	out.write(f'* Message body: "{message}"\n')
	out.write("*\n")
	arg_index = 0
	for arg_index, arg in enumerate(entry.get("ParamTypes", [])):
	arg_index += 1

	out.write(
	f"* @param[in] arg{arg_index} Parameter of message that will replace %{arg_index} in its body.\n"
	)
	out.write("*\n")
	out.write(
	f"* @returns Message {entry_id} formatted to JSON */\n"
	)

	out.write(
	make_error_function(
	entry_id, entry, True, registry_name, namespace_name
	)
	)
	out.write(" }\n")
	out.write("}\n")

	error_messages_cpp = os.path.join(
	SCRIPT_DIR, "..", "redfish-core", "src", f"{base_filename}.cpp"
	)
	with open(
	error_messages_cpp,
	"w",
	) as out:
	out.write(WARNING)
	out.write(f'\n#include "{base_filename}.hpp"\n')
	headers = []

	headers.append('"registries.hpp"')
	if registry_name == "Base":
	reg_name_lower = "base"
	headers.append('"error_message_utils.hpp"')
	headers.append('"http_response.hpp"')
	headers.append('"logging.hpp"')
	headers.append("<boost/beast/http/field.hpp>")
	headers.append("<boost/beast/http/status.hpp>")
	headers.append("<boost/url/url_view_base.hpp>")
	headers.append("<source_location>")
	else:
	reg_name_lower = namespace_name.lower()
	headers.append(f'"registries/{reg_name_lower}_message_registry.hpp"')

	headers.append("<nlohmann/json.hpp>")
	headers.append("<array>")
	headers.append("<cstddef>")
	headers.append("<span>")

	if registry_name not in ("ResourceEvent", "HeartbeatEvent"):
	headers.append("<cstdint>")
	headers.append("<string>")
	headers.append("<string_view>")

	for header in headers:
	out.write(f"#include {header}\n")

	out.write(
	"""
	// Clang can't seem to decide whether this header needs to be included or not,
	// and is inconsistent. Include it for now
	// NOLINTNEXTLINE(misc-include-cleaner)
	#include <utility>

	namespace redfish
	{

	namespace messages
	{
	"""
	)
	out.write(
	"""
	static nlohmann::json getLog(redfish::registries::{namespace_name}::Index name,
	std::span<const std::string_view> args)
	{{
	size_t index = static_cast<size_t>(name);
	if (index >= redfish::registries::{namespace_name}::registry.size())
	{{
	return {{}};
	}}
	return getLogFromRegistry(redfish::registries::{namespace_name}::header,
	redfish::registries::{namespace_name}::registry, index, args);
	}}

	""".format(
	namespace_name=namespace_name
	)
	)
	for entry_id, entry in messages.items():
	out.write(
	f"""/**
	* @internal
	* @brief Formats {entry_id} message into JSON
	*
	* See header file for more information
	* @endinternal
	*/
	"""
	)
	message = entry["Message"]
	out.write(
	make_error_function(
	entry_id, entry, False, registry_name, namespace_name
	)
	)

	out.write(" }\n")
	out.write("}\n")
	os.system(f"clang-format -i {error_messages_hpp} {error_messages_cpp}")


	def make_privilege_registry() -> None:
	path, json_file, type_name, url = make_getter(
	"Redfish_1.5.0_PrivilegeRegistry.json",
	"privilege_registry.hpp",
	"privilege",
	)
	with open(path, "w") as registry:
	registry.write(PRIVILEGE_HEADER)

	privilege_dict: t.Dict[str, t.Tuple[t.Any, str \| None]] = {}
	for mapping in json_file["Mappings"]:
	# first pass, identify all the unique privilege sets
	for operation, privilege_list in mapping["OperationMap"].items():
	privilege_dict[
	get_privilege_string_from_list(privilege_list)
	] = (
	privilege_list,
	None,
	)
	for index, key in enumerate(privilege_dict):
	(privilege_list, _) = privilege_dict[key]
	name = get_variable_name_for_privilege_set(privilege_list)
	registry.write(
	"const std::array<Privileges, {length}> "
	"privilegeSet{name} = {key};\n".format(
	length=len(privilege_list), name=name, key=key
	)
	)
	privilege_dict[key] = (privilege_list, name)

	for mapping in json_file["Mappings"]:
	entity = mapping["Entity"]
	registry.write("// {}\n".format(entity))
	for operation, privilege_list in mapping["OperationMap"].items():
	privilege_string = get_privilege_string_from_list(
	privilege_list
	)
	operation = operation.lower()

	registry.write(
	"const static auto& {}{} = privilegeSet{};\n".format(
	operation, entity, privilege_dict[privilege_string][1]
	)
	)
	registry.write("\n")
	registry.write(
	"} // namespace redfish::privileges\n// clang-format on\n"
	)


	def to_pascal_case(text: str) -> str:
	s = text.replace("_", " ")
	s1 = s.split()
	if len(text) == 0:
	return text
	return "".join(i.capitalize() for i in s1[0:])


	def main() -> None:
	dmtf_registries = OrderedDict(
	[
	("base", "1.19.0"),
	("composition", "1.1.2"),
	("environmental", "1.1.0"),
	("ethernet_fabric", "1.0.1"),
	("fabric", "1.0.2"),
	("heartbeat_event", "1.0.1"),
	("job_event", "1.0.1"),
	("license", "1.0.3"),
	("log_service", "1.0.1"),
	("network_device", "1.0.3"),
	("platform", "1.0.1"),
	("power", "1.0.1"),
	("resource_event", "1.3.0"),
	("sensor_event", "1.0.1"),
	("storage_device", "1.2.1"),
	("task_event", "1.0.3"),
	("telemetry", "1.0.0"),
	("update", "1.0.2"),
	]
	)

	parser = argparse.ArgumentParser()
	parser.add_argument(
	"--registries",
	type=str,
	default="privilege,openbmc,"
	+ ",".join([dmtf for dmtf in dmtf_registries]),
	help="Comma delimited list of registries to update",
	)

	args = parser.parse_args()

	registries = set(args.registries.split(","))
	registries_map: t.OrderedDict[str, RegistryInfo] = OrderedDict()

	for registry, version in dmtf_registries.items():
	if registry in registries:
	registry_pascal_case = to_pascal_case(registry)
	registries_map[registry] = make_getter(
	f"{registry_pascal_case}.{version}.json",
	f"{registry}_message_registry.hpp",
	registry,
	)
	if "openbmc" in registries:
	registries_map["openbmc"] = openbmc_local_getter()

	update_registries(list(registries_map.values()))

	if "base" in registries_map:
	create_error_registry(
	registries_map["base"],
	"Base",
	"base",
	"error",
	)
	if "heartbeat_event" in registries_map:
	create_error_registry(
	registries_map["heartbeat_event"],
	"HeartbeatEvent",
	"heartbeat_event",
	"heartbeat",
	)
	if "resource_event" in registries_map:
	create_error_registry(
	registries_map["resource_event"],
	"ResourceEvent",
	"resource_event",
	"resource",
	)
	if "task_event" in registries_map:
	create_error_registry(
	registries_map["task_event"],
	"TaskEvent",
	"task_event",
	"task",
	)

	if "privilege" in registries:
	make_privilege_registry()


	if __name__ == "__main__":
	main()