diff --git a/src/hermes/commands/__init__.py b/src/hermes/commands/__init__.py
index d2116ef2..03539bd4 100644
--- a/src/hermes/commands/__init__.py
+++ b/src/hermes/commands/__init__.py
@@ -8,12 +8,12 @@
# "unused import" errors.
# flake8: noqa
-# from hermes.commands.base import HermesHelpCommand
-# from hermes.commands.base import HermesVersionCommand
-# from hermes.commands.clean.base import HermesCleanCommand
+from hermes.commands.base import HermesHelpCommand
+from hermes.commands.base import HermesVersionCommand
+from hermes.commands.clean.base import HermesCleanCommand
# from hermes.commands.init.base import HermesInitCommand
from hermes.commands.curate.base import HermesCurateCommand
from hermes.commands.harvest.base import HermesHarvestCommand
-# from hermes.commands.process.base import HermesProcessCommand
-# from hermes.commands.deposit.base import HermesDepositCommand
+from hermes.commands.process.base import HermesProcessCommand
+from hermes.commands.deposit.base import HermesDepositCommand
# from hermes.commands.postprocess.base import HermesPostprocessCommand
diff --git a/src/hermes/commands/base.py b/src/hermes/commands/base.py
index 2d182267..12e3c994 100644
--- a/src/hermes/commands/base.py
+++ b/src/hermes/commands/base.py
@@ -175,6 +175,7 @@ def __call__(self, command: HermesCommand) -> None:
class HermesHelpSettings(BaseModel):
+ """Intentionally empty settings class for the help command."""
pass
@@ -200,3 +201,23 @@ def __call__(self, args: argparse.Namespace) -> None:
# Otherwise, simply show the general help and exit (cleanly).
self.parser.print_help()
self.parser.exit()
+
+
+class HermesVersionSettings(BaseModel):
+ """Intentionally empty settings class for the version command."""
+ pass
+
+
+class HermesVersionCommand(HermesCommand):
+ """Show HERMES version and exit."""
+
+ command_name = "version"
+ settings_class = HermesVersionSettings
+
+ def load_settings(self, args: argparse.Namespace):
+ """Pass loading settings as not necessary for this command."""
+ pass
+
+ def __call__(self, args: argparse.Namespace) -> None:
+ self.log.info(metadata.version("hermes"))
+ self.parser.exit()
diff --git a/src/hermes/commands/cli.py b/src/hermes/commands/cli.py
index 565381fc..d725e40a 100644
--- a/src/hermes/commands/cli.py
+++ b/src/hermes/commands/cli.py
@@ -16,7 +16,9 @@
# from hermes.commands import (HermesHelpCommand, HermesVersionCommand, HermesCleanCommand,
# HermesHarvestCommand, HermesProcessCommand, HermesCurateCommand,
# HermesDepositCommand, HermesPostprocessCommand, HermesInitCommand)
-from hermes.commands import HermesCurateCommand, HermesHarvestCommand
+from hermes.commands import (HermesCurateCommand, HermesDepositCommand,
+ HermesHarvestCommand, HermesHelpCommand,
+ HermesProcessCommand, HermesVersionCommand)
from hermes.commands.base import HermesCommand
@@ -38,15 +40,15 @@ def main() -> None:
setting_types = {}
for command in (
- # HermesHelpCommand(parser),
- # HermesVersionCommand(parser),
- # HermesInitCommand(parser),
# HermesCleanCommand(parser),
- HermesHarvestCommand(parser),
- # HermesProcessCommand(parser),
HermesCurateCommand(parser),
- # HermesDepositCommand(parser),
+ HermesDepositCommand(parser),
+ HermesHarvestCommand(parser),
+ HermesHelpCommand(parser),
+ # HermesInitCommand(parser),
# HermesPostprocessCommand(parser),
+ HermesProcessCommand(parser),
+ HermesVersionCommand(parser),
):
if command.settings_class is not None:
setting_types[command.command_name] = command.settings_class
diff --git a/src/hermes/commands/deposit/base.py b/src/hermes/commands/deposit/base.py
index 75018579..6fbf3625 100644
--- a/src/hermes/commands/deposit/base.py
+++ b/src/hermes/commands/deposit/base.py
@@ -7,15 +7,13 @@
import abc
import argparse
-import json
-import sys
from pydantic import BaseModel
from hermes.commands.base import HermesCommand, HermesPlugin
-from hermes.model.context import CodeMetaContext
-from hermes.model.path import ContextPath
-from hermes.model.errors import HermesValidationError
+from hermes.model.context_manager import HermesContext
+from hermes.model import SoftwareMetadata
+from hermes.model.error import HermesValidationError
class BaseDepositPlugin(HermesPlugin):
@@ -24,26 +22,35 @@ class BaseDepositPlugin(HermesPlugin):
TODO: describe workflow... needs refactoring to be less stateful!
"""
- def __init__(self, command, ctx):
- self.command = command
- self.ctx = ctx
-
def __call__(self, command: HermesCommand) -> None:
"""Initiate the deposition process.
This calls a list of additional methods on the class, none of which need to be implemented.
"""
self.command = command
+ self.ctx = HermesContext()
+
+ self.ctx.prepare_step("curate")
+ self.metadata = SoftwareMetadata.load_from_cache(self.ctx, "result")
+ self.ctx.finalize_step("curate")
self.prepare()
- self.map_metadata()
+ deposit = self.map_metadata()
+ self.ctx.prepare_step("deposit")
+ with self.ctx[command.settings.target] as cache:
+ cache["deposit"] = deposit
+ self.ctx.finalize_step("deposit")
if self.is_initial_publication():
self.create_initial_version()
else:
self.create_new_version()
- self.update_metadata()
+ deposit = self.update_metadata()
+ self.ctx.prepare_step("deposit")
+ with self.ctx["deposit"] as cache:
+ cache["result"] = deposit
+ self.ctx.finalize_step("deposit")
self.delete_artifacts()
self.upload_artifacts()
self.publish()
@@ -58,8 +65,8 @@ def prepare(self) -> None:
pass
@abc.abstractmethod
- def map_metadata(self) -> None:
- """Map the given metadata to the target schema of the deposition platform.
+ def map_metadata(self) -> dict:
+ """Map the given metadata to the target schema of the deposition platform and return it.
When mapping metadata, make sure to add traces to the HERMES software, e.g. via
DataCite's ``relatedIdentifier`` using the ``isCompiledBy`` relation. Ideally, the value
@@ -88,8 +95,9 @@ def create_new_version(self) -> None:
"""Create a new version of an existing publication on the target platform."""
pass
- def update_metadata(self) -> None:
- """Update the metadata of the newly created version."""
+ @abc.abstractmethod
+ def update_metadata(self) -> dict:
+ """Update the metadata of the newly created version and return it even if it hasn't changed."""
pass
def delete_artifacts(self) -> None:
@@ -106,7 +114,7 @@ def publish(self) -> None:
pass
-class _DepositSettings(BaseModel):
+class DepositSettings(BaseModel):
"""Generic deposition settings."""
target: str = ""
@@ -116,7 +124,7 @@ class HermesDepositCommand(HermesCommand):
""" Deposit the curated metadata to repositories. """
command_name = "deposit"
- settings_class = _DepositSettings
+ settings_class = DepositSettings
def init_command_parser(self, command_parser: argparse.ArgumentParser) -> None:
command_parser.add_argument('--file', '-f', nargs=1, action='append',
@@ -128,26 +136,13 @@ def __call__(self, args: argparse.Namespace) -> None:
self.args = args
plugin_name = self.settings.target
- ctx = CodeMetaContext()
- codemeta_file = ctx.get_cache("curate", ctx.hermes_name)
- if not codemeta_file.exists():
- self.log.error("You must run the 'curate' command before deposit")
- sys.exit(1)
-
- codemeta_path = ContextPath("codemeta")
- with open(codemeta_file) as codemeta_fh:
- ctx.update(codemeta_path, json.load(codemeta_fh))
-
try:
- plugin_func = self.plugins[plugin_name](self, ctx)
-
+ plugin_func = self.plugins[plugin_name]()
except KeyError as e:
self.log.error("Plugin '%s' not found.", plugin_name)
self.errors.append(e)
-
try:
plugin_func(self)
-
except HermesValidationError as e:
self.log.error("Error while executing %s: %s", plugin_name, e)
self.errors.append(e)
diff --git a/src/hermes/commands/deposit/file.py b/src/hermes/commands/deposit/file.py
index 6c5d6419..ed6bd570 100644
--- a/src/hermes/commands/deposit/file.py
+++ b/src/hermes/commands/deposit/file.py
@@ -11,22 +11,23 @@
from pydantic import BaseModel
from hermes.commands.deposit.base import BaseDepositPlugin
-from hermes.model.path import ContextPath
class FileDepositSettings(BaseModel):
- filename: str = 'hermes.json'
+ filename: str = 'codemeta.json'
class FileDepositPlugin(BaseDepositPlugin):
settings_class = FileDepositSettings
- def map_metadata(self) -> None:
- self.ctx.update(ContextPath.parse('deposit.file'), self.ctx['codemeta'])
+ def map_metadata(self) -> dict:
+ return self.metadata.compact()
+
+ def update_metadata(self) -> dict:
+ return self.metadata.compact()
def publish(self) -> None:
file_config = self.command.settings.file
- output_data = self.ctx['deposit.file']
with open(file_config.filename, 'w') as deposition_file:
- json.dump(output_data, deposition_file, indent=2)
+ json.dump(self.metadata.compact(), deposition_file, indent=2)
diff --git a/src/hermes/commands/deposit/invenio.py b/src/hermes/commands/deposit/invenio.py
index 69fb87a0..ba45c146 100644
--- a/src/hermes/commands/deposit/invenio.py
+++ b/src/hermes/commands/deposit/invenio.py
@@ -6,22 +6,20 @@
# SPDX-FileContributor: Oliver Bertuch
# SPDX-FileContributor: Michael Meinel
-import json
import logging
import pathlib
-import typing as t
from datetime import date, datetime
from pathlib import Path
from urllib.parse import urlparse
import requests
from pydantic import BaseModel
+from typing import Union
-from hermes.commands.deposit.base import BaseDepositPlugin, HermesDepositCommand
+from hermes.commands.deposit.base import BaseDepositPlugin
from hermes.commands.deposit.error import DepositionUnauthorizedError
from hermes.error import MisconfigurationError
-from hermes.model.context import CodeMetaContext
-from hermes.model.path import ContextPath
+from hermes.model.error import HermesValidationError
from hermes.utils import hermes_doi, hermes_user_agent
@@ -109,7 +107,7 @@ def __init__(self, client=None):
def resolve_latest_id(
self, record_id=None, doi=None, codemeta_identifier=None
- ) -> t.Tuple[t.Optional[str], dict]:
+ ) -> tuple[Union[str, None], dict]:
"""
Using the given metadata parameters, figure out the latest record id.
@@ -167,7 +165,7 @@ def resolve_doi(self, doi) -> str:
*_, record_id = page_url.path.split('/')
return record_id
- def resolve_record_id(self, record_id: str) -> t.Tuple[str, dict]:
+ def resolve_record_id(self, record_id: str) -> tuple[str, dict]:
"""
Find the latest version of a given record.
@@ -186,7 +184,7 @@ def resolve_record_id(self, record_id: str) -> t.Tuple[str, dict]:
res_json = res.json()
return res_json['id'], res_json['metadata']
- def resolve_license_id(self, license_url: t.Optional[str]) -> t.Optional[str]:
+ def resolve_license_id(self, license_url: Union[str, None]) -> Union[str, None]:
"""Get Invenio license representation from CodeMeta.
The license to use is extracted from the ``license`` field in the
@@ -219,7 +217,7 @@ def resolve_license_id(self, license_url: t.Optional[str]) -> t.Optional[str]:
parsed_url = urlparse(license_url)
url_path = parsed_url.path.rstrip("/")
- license_id = url_path.split("/")[-1]
+ license_id = str.lower(url_path.split("/")[-1])
response = self.client.get_license(license_id)
if response.status_code == 404:
@@ -231,7 +229,8 @@ def resolve_license_id(self, license_url: t.Optional[str]) -> t.Optional[str]:
@staticmethod
def _extract_license_id_from_response(data: dict) -> str:
- return data["metadata"]["id"]
+ # TODO: find correct key, data["metadata"]["id"] did not work for me but data["id"] does
+ return data["id"]
class InvenioDepositSettings(BaseModel):
@@ -243,7 +242,7 @@ class InvenioDepositSettings(BaseModel):
access_right: str = None
embargo_date: str = None
access_conditions: str = None
- api_paths: t.Dict = {}
+ api_paths: dict = {}
auth_token: str = ''
files: list[pathlib.Path] = []
@@ -258,11 +257,13 @@ class InvenioDepositPlugin(BaseDepositPlugin):
invenio_resolver_class = InvenioResolver
settings_class = InvenioDepositSettings
- def __init__(self, command: HermesDepositCommand, ctx: CodeMetaContext, client=None, resolver=None) -> None:
- super().__init__(command, ctx)
+ def __init__(self) -> None:
+ super().__init__()
- self.invenio_context_path = ContextPath.parse(f"deposit.{self.platform_name}")
self.invenio_ctx = None
+
+ def __call__(self, command, *, client=None, resolver=None):
+ self.command = command
self.config = getattr(self.command.settings, self.platform_name)
if client is None:
@@ -292,7 +293,9 @@ def __init__(self, command: HermesDepositCommand, ctx: CodeMetaContext, client=N
self.resolver = resolver or self.invenio_resolver_class(self.client)
self.links = {}
- # TODO: Populate some data structure here? Or move more of this into __init__?
+ super().__call__(command)
+
+ # TODO: Populate some data structure here? Or move more of this into __init__.py?
def prepare(self) -> None:
"""Prepare the deposition on an Invenio-based platform.
@@ -305,49 +308,42 @@ def prepare(self) -> None:
- check access modalities (access right, access conditions, embargo data, existence
of license)
- check whether required configuration options are present
- - update ``self.ctx`` with metadata collected during the checks
+ - update ``self.metadata`` with metadata collected during the checks
"""
rec_id = self.config.record_id
doi = self.config.doi
- try:
- codemeta_identifier = self.ctx["codemeta.identifier"]
- except KeyError:
- codemeta_identifier = None
-
+ codemeta_identifier = self.metadata.get("identifier", None)
rec_id, rec_meta = self.resolver.resolve_latest_id(
record_id=rec_id, doi=doi, codemeta_identifier=codemeta_identifier
)
- version = self.ctx["codemeta"].get("version")
+ if len(self.metadata.get("version", [])) > 1:
+ raise HermesValidationError("Too many licenses for invenio deposit.")
+ if len(self.metadata.get("version", [])) == 1:
+ version = self.metadata["version"][0]
+ else:
+ version = None
if rec_meta and (version == rec_meta.get("version")):
raise ValueError(f"Version {version} already deposited.")
- self.ctx.update(self.invenio_context_path['latestRecord'], {'id': rec_id, 'metadata': rec_meta})
-
- license = self._get_license_identifier()
- self.ctx.update(self.invenio_context_path["license"], license)
-
- communities = self._get_community_identifiers()
- self.ctx.update(self.invenio_context_path["communities"], communities)
+ deposition_data = {}
+ deposition_data["latestRecord"] = {'id': rec_id, 'metadata': rec_meta}
+ deposition_data["license"] = self._get_license_identifier()
+ deposition_data["communities"] = self._get_community_identifiers()
access_right, embargo_date, access_conditions = self._get_access_modalities(license)
- self.ctx.update(self.invenio_context_path["access_right"], access_right)
- self.ctx.update(self.invenio_context_path["embargo_date"], embargo_date)
- self.ctx.update(self.invenio_context_path["access_conditions"], access_conditions)
-
- self.invenio_ctx = self.ctx[self.invenio_context_path]
+ deposition_data["access_right"] = access_right
+ deposition_data["embargo_date"] = embargo_date
+ deposition_data["access_conditions"] = access_conditions
- def map_metadata(self) -> None:
- """Map the harvested metadata onto the Invenio schema."""
+ self.invenio_ctx = deposition_data
- deposition_metadata = self._codemeta_to_invenio_deposition()
- self.ctx.update(self.invenio_context_path["depositionMetadata"], deposition_metadata)
-
- # Store a snapshot of the mapped data within the cache, useful for analysis, debugging, etc
- with open(self.ctx.get_cache("deposit", self.platform_name, create=True), 'w') as invenio_json:
- json.dump(deposition_metadata, invenio_json, indent=' ')
+ def map_metadata(self) -> dict:
+ """Map the harvested metadata onto the Invenio schema and return it."""
+ self.invenio_ctx["depositionMetadata"] = self._codemeta_to_invenio_deposition()
+ return self.invenio_ctx["depositionMetadata"]
def is_initial_publication(self) -> bool:
latest_record_id = self.invenio_ctx.get("latestRecord", {}).get("id")
@@ -406,8 +402,8 @@ def related_identifiers(self):
},
]
- def update_metadata(self) -> None:
- """Update the metadata of a draft."""
+ def update_metadata(self) -> dict:
+ """Update the metadata of a draft and return it."""
draft_url = self.links["latest_draft"]
@@ -426,8 +422,7 @@ def update_metadata(self) -> None:
self.links.update(deposit["links"])
_log.debug("Created new version deposit: %s", self.links["html"])
- with open(self.ctx.get_cache('deposit', 'deposit', create=True), 'w') as deposit_file:
- json.dump(deposit, deposit_file, indent=4)
+ return deposit
def delete_artifacts(self) -> None:
"""Delete existing file artifacts.
@@ -448,7 +443,11 @@ def upload_artifacts(self) -> None:
bucket_url = self.links["bucket"]
- files = *self.config.files, *[f[0] for f in self.command.args.file]
+ if self.command.args.file:
+ files = *self.config.files, *[f[0] for f in self.command.args.file]
+ else:
+ files = tuple(self.config.files)
+
for path_arg in files:
path = Path(path_arg)
@@ -505,14 +504,41 @@ def _codemeta_to_invenio_deposition(self) -> dict:
differences between Invenio-based platforms.
"""
- metadata = self.ctx["codemeta"]
+ metadata = self.metadata
license = self.invenio_ctx["license"]
communities = self.invenio_ctx["communities"]
access_right = self.invenio_ctx["access_right"]
embargo_date = self.invenio_ctx["embargo_date"]
access_conditions = self.invenio_ctx["access_conditions"]
- creators = [
+ creators = []
+ for author in metadata.get("author", []):
+ if "Person" not in author.get("@type", []):
+ continue
+ creator = {}
+ if len(
+ affils := [
+ name for affil in author.get("affiliation", []) for name in affil.get("legalname", [])
+ ]
+ ) != 0:
+ creator["affiliation"] = affils
+
+ if len(author.get("familyName", [])) > 1:
+ raise HermesValidationError(f"Author has too many family names: {author}")
+ if len(author.get("familyName", [])) == 1:
+ given_names_str = " ".join(author.get("givenName", []))
+ name = f"{author['familyName'][0]}, {given_names_str}"
+ elif len(author.get("name", [])) != 1:
+ raise HermesValidationError(f"Author has too many or no names: {author}")
+ else:
+ name = author["name"][0]
+ creator["name"] = name
+ if (id := author.get("@id", None)) is not None:
+ creator["orcid"] = id.replace("https://orcid.org/", "")
+ if creator:
+ creators.append(creator)
+
+ """creators = [
# TODO: Distinguish between @type "Person" and others
{
k: v for k, v in {
@@ -520,15 +546,16 @@ def _codemeta_to_invenio_deposition(self) -> dict:
"affiliation": author.get("affiliation", {"legalName": None}).get("legalName"),
# Invenio wants "family, given". author.get("name") might not have this format.
"name": f"{author.get('familyName')}, {author.get('givenName')}"
- if author.get("familyName") and author.get("givenName")
+ if "familyName" in author and "givenName" in author
else author.get("name"),
# Invenio expects the ORCID without the URL part
"orcid": author.get("@id", "").replace("https://orcid.org/", "") or None,
}.items() if v is not None
}
for author in metadata["author"]
- ]
+ ]"""
+ # TODO: reimplement with new api
# This is not used at the moment. See comment below in `deposition_metadata` dict.
contributors = [ # noqa: F841
# TODO: Distinguish between @type "Person" and others
@@ -538,7 +565,7 @@ def _codemeta_to_invenio_deposition(self) -> dict:
"affiliation": contributor.get("affiliation", {"legalName": None}).get("legalName"),
# Invenio wants "family, given". contributor.get("name") might not have this format.
"name": f"{contributor.get('familyName')}, {contributor.get('givenName')}"
- if contributor.get("familyName") and contributor.get("givenName")
+ if "familyName" in contributor and "givenName" in contributor
else contributor.get("name"),
# Invenio expects the ORCID without the URL part
"orcid": contributor.get("@id", "").replace("https://orcid.org/", "") or None,
@@ -550,6 +577,33 @@ def _codemeta_to_invenio_deposition(self) -> dict:
for contributor in metadata.get("contributor", []) if contributor.get("name") != "GitHub"
]
+ if len(metadata.get("name", [])) != 1:
+ _log.error("More than one or zero names for the Software are given.")
+ raise HermesValidationError("More than one or zerno names for the Software.")
+ name = metadata["name"][0]
+
+ if len(metadata.get("schema:description", [])) > 1:
+ _log.error("More than one descriptions of the Software are given.")
+ raise HermesValidationError("More than one descriptions of the Software are given.")
+ if len(metadata.get("schema:description", [])) == 1:
+ description = metadata["schema:description"][0]
+ else:
+ description = None
+
+ if len(metadata.get("schema:version", [])) > 1:
+ _log.error("More than one version of the Software are given.")
+ raise HermesValidationError("More than one version of the Software are given.")
+ if len(metadata.get("schema:version", [])) == 1:
+ version = metadata["schema:version"][0]
+ else:
+ version = None
+
+ keywords = metadata.get("schema:keywords", [])
+ if len(keywords) == 0:
+ keywords = None
+ else:
+ keywords = keywords.to_python()
+
# TODO: Use the fields currently set to `None`.
# Some more fields are available but they most likely don't relate to software
# publications targeted by hermes.
@@ -563,12 +617,9 @@ def _codemeta_to_invenio_deposition(self) -> dict:
# TODO: Maybe we want a different date? Then make this configurable. If not,
# this can be removed as it defaults to today.
"publication_date": date.today().isoformat(),
- "title": metadata["name"],
+ "title": name,
"creators": creators,
- # TODO: Use a real description here. Possible sources could be
- # `tool.poetry.description` from pyproject.toml or `abstract` from
- # CITATION.cff. This should then be stored in codemeta description field.
- "description": metadata["name"],
+ "description": description,
"access_right": access_right,
"license": license,
"embargo_date": embargo_date,
@@ -581,8 +632,8 @@ def _codemeta_to_invenio_deposition(self) -> dict:
# them.
# TODO: Use the DOI we get back from this.
"prereserve_doi": True,
- # TODO: A good source for this could be `tool.poetry.keywords` in pyproject.toml.
- "keywords": None,
+ "keywords": keywords,
+ # TODO: Is there a good codemeta/ schema field?
"notes": None,
"related_identifiers": self.related_identifiers(),
# TODO: Use `contributors`. In the case of the hermes workflow itself, the
@@ -594,17 +645,21 @@ def _codemeta_to_invenio_deposition(self) -> dict:
"communities": communities,
"grants": None,
"subjects": None,
- "version": metadata.get('version'),
+ "version": version,
}.items() if v is not None}
return deposition_metadata
- def _get_license_identifier(self) -> t.Optional[str]:
+ def _get_license_identifier(self) -> Union[str, None]:
"""Get Invenio license identifier that matches the given license URL.
If no license is configured, ``None`` will be returned.
"""
- license_url = self.ctx["codemeta"].get("license")
+ if "license" not in self.metadata:
+ raise HermesValidationError("No license is given.")
+ if len(self.metadata["license"]) > 1:
+ raise HermesValidationError("Too many licenses for invenio deposit.")
+ license_url = self.metadata["license"][0]
return self.resolver.resolve_license_id(license_url)
def _get_community_identifiers(self):
@@ -612,7 +667,7 @@ def _get_community_identifiers(self):
This function gets the communities to be used for the deposition on an Invenio-based
site from the config and checks their validity against the site's API. If one of the
- identifiers can not be found on the site, a :class:`HermesMisconfigurationError` is
+ identifiers can not be found on the site, a :class:`MisconfigurationError` is
raised.
"""
diff --git a/src/hermes/commands/deposit/invenio_rdm.py b/src/hermes/commands/deposit/invenio_rdm.py
index a381db90..01e08371 100644
--- a/src/hermes/commands/deposit/invenio_rdm.py
+++ b/src/hermes/commands/deposit/invenio_rdm.py
@@ -6,9 +6,8 @@
# SPDX-FileContributor: Oliver Bertuch
# SPDX-FileContributor: Michael Meinel
-import typing as t
-
from requests import HTTPError
+from typing import Union
from hermes.commands.deposit.invenio import InvenioClient, InvenioDepositPlugin, InvenioResolver
@@ -27,7 +26,7 @@ def get_licenses(self):
class InvenioRDMResolver(InvenioResolver):
invenio_client_class = InvenioRDMClient
- def resolve_license_id(self, license_url: t.Optional[str]) -> t.Optional[dict]:
+ def resolve_license_id(self, license_url: Union[str, None]) -> Union[dict, None]:
"""Deliberately try to resolve the license URL to a valid InvenioRDM license information record from the
vocabulary.
@@ -47,6 +46,12 @@ def resolve_license_id(self, license_url: t.Optional[str]) -> t.Optional[dict]:
except HTTPError:
pass
+ # FIXME: Why not get all license_cross_refs and then use a query parameter like this:
+ # ?q=props.url:("license_url" OR "license_cross_ref[1]" OR ...)&size=1000
+ # That would be able to replace _search_license_info.
+ # FIXME: Some licenses in valid_licenses["hits"]["hits"]["props"]["url"] are only http although
+ # https://spdx.org/licenses/license.json lists them in crossRef as https
+
# If the easy "mapping" did not work, we really need to "search" for the correct license ID.
response = self.client.get_licenses()
response.raise_for_status()
@@ -65,6 +70,7 @@ def resolve_license_id(self, license_url: t.Optional[str]) -> t.Optional[dict]:
if license_info is not None:
break
else:
+ # FIXME: Why is this only raised here and not always when license_info is None?
raise RuntimeError(f"Could not resolve license URL {license_url} to a valid identifier.")
return license_info
@@ -73,7 +79,7 @@ def resolve_license_id(self, license_url: t.Optional[str]) -> t.Optional[dict]:
def _extract_license_id_from_response(data: dict) -> str:
return data["id"]
- def _search_license_info(self, _url: str, valid_licenses: dict) -> t.Optional[dict]:
+ def _search_license_info(self, _url: str, valid_licenses: dict) -> Union[dict, None]:
for license_info in valid_licenses['hits']['hits']:
try:
if license_info['props']['url'] == _url:
diff --git a/src/hermes/commands/process/base.py b/src/hermes/commands/process/base.py
index 9e29d1e6..1aae0dab 100644
--- a/src/hermes/commands/process/base.py
+++ b/src/hermes/commands/process/base.py
@@ -5,13 +5,14 @@
# SPDX-FileContributor: Michael Meinel
import argparse
-import json
-import sys
from pydantic import BaseModel
from hermes.commands.base import HermesCommand, HermesPlugin
-from hermes.model.context import HermesHarvestContext, CodeMetaContext
+from hermes.model.api import SoftwareMetadata
+from hermes.model.context_manager import HermesContext
+from hermes.model.error import HermesContextError
+from hermes.model.merge.container import ld_merge_dict
class HermesProcessPlugin(HermesPlugin):
@@ -33,42 +34,27 @@ class HermesProcessCommand(HermesCommand):
def __call__(self, args: argparse.Namespace) -> None:
self.args = args
- ctx = CodeMetaContext()
-
- if not (ctx.hermes_dir / "harvest").exists():
- self.log.error("You must run the harvest command before process")
- sys.exit(1)
+ ctx = HermesContext()
+ merged_doc = ld_merge_dict([{}])
# Get all harvesters
harvester_names = self.root_settings.harvest.sources
- harvester_names.reverse() # Switch order for priority handling
+ ctx.prepare_step('harvest')
for harvester in harvester_names:
self.log.info("## Process data from %s", harvester)
-
- harvest_context = HermesHarvestContext(ctx, harvester, {})
try:
- harvest_context.load_cache()
- # when the harvest step ran, but there is no cache file, this is a serious flaw
- except FileNotFoundError:
- self.log.warning("No output data from harvester %s found, skipping", harvester)
+ metadata = SoftwareMetadata.load_from_cache(ctx, harvester)
+ except HermesContextError as e:
+ self.log.error("Error while trying to load data from harvest plugin '%s': %s", harvester, e)
+ self.errors.append(e)
continue
-
- ctx.merge_from(harvest_context)
- ctx.merge_contexts_from(harvest_context)
-
- if ctx._errors:
- self.log.error('Errors during merge')
- self.errors.extend(ctx._errors)
-
- for ep, error in ctx._errors:
- self.log.info(" - %s: %s", ep.name, error)
-
- tags_path = ctx.get_cache('process', 'tags', create=True)
- with tags_path.open('w') as tags_file:
- json.dump(ctx.tags, tags_file, indent=2)
-
- ctx.prepare_codemeta()
-
- with open(ctx.get_cache("process", ctx.hermes_name, create=True), 'w') as codemeta_file:
- json.dump(ctx._data, codemeta_file, indent=2)
+ merged_doc.update(metadata)
+ ctx.finalize_step("harvest")
+
+ ctx.prepare_step("process")
+ with ctx["result"] as result_ctx:
+ result_ctx["codemeta"] = merged_doc.compact()
+ result_ctx["context"] = {"@context": merged_doc.full_context}
+ result_ctx["expanded"] = merged_doc.ld_value
+ ctx.finalize_step("process")
diff --git a/src/hermes/error.py b/src/hermes/error.py
index e56c2499..1669ed39 100644
--- a/src/hermes/error.py
+++ b/src/hermes/error.py
@@ -4,5 +4,5 @@
# SPDX-FileContributor: David Pape
-class HermesMisconfigurationError(Exception):
+class MisconfigurationError(Exception):
pass
diff --git a/src/hermes/model/api.py b/src/hermes/model/api.py
index 8b079544..db582656 100644
--- a/src/hermes/model/api.py
+++ b/src/hermes/model/api.py
@@ -1,6 +1,14 @@
-from hermes.model.types import ld_dict
+# SPDX-FileCopyrightText: 2026 German Aerospace Center (DLR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: Michael Fritzsche
+# SPDX-FileContributor: Stephan Druskat
+from hermes.model.context_manager import HermesContext, HermesContexError
+from hermes.model.types import ld_dict
from hermes.model.types.ld_context import ALL_CONTEXTS
+from hermes.model.types.ld_dict import bundled_loader
class SoftwareMetadata(ld_dict):
@@ -8,3 +16,21 @@ class SoftwareMetadata(ld_dict):
def __init__(self, data: dict = None, extra_vocabs: dict[str, str] = None) -> None:
ctx = ALL_CONTEXTS + [{**extra_vocabs}] if extra_vocabs is not None else ALL_CONTEXTS
super().__init__([ld_dict.from_dict(data, context=ctx).data_dict if data else {}], context=ctx)
+
+ @classmethod
+ def load_from_cache(cls, ctx: HermesContext, source: str) -> "SoftwareMetadata":
+ with ctx[source] as cache:
+ try:
+ return SoftwareMetadata(cache["codemeta"])
+ except Exception:
+ pass
+ try:
+ context = cache["context"]["@context"]
+ data = SoftwareMetadata()
+ data.active_ctx = data.ld_proc.initial_ctx(context, {"documentLoader": bundled_loader})
+ data.context = context
+ for key, value in cache["expanded"][0]:
+ data[key] = value
+ return data
+ except Exception as e:
+ raise HermesContexError("There is no (valid) data stored in the cache.") from e
diff --git a/src/hermes/model/merge/__init__.py b/src/hermes/model/merge/__init__.py
new file mode 100644
index 00000000..faf5a2f5
--- /dev/null
+++ b/src/hermes/model/merge/__init__.py
@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2022 German Aerospace Center (DLR)
+#
+# SPDX-License-Identifier: Apache-2.0
diff --git a/src/hermes/model/merge/action.py b/src/hermes/model/merge/action.py
new file mode 100644
index 00000000..b9b516ef
--- /dev/null
+++ b/src/hermes/model/merge/action.py
@@ -0,0 +1,264 @@
+# SPDX-FileCopyrightText: 2025 German Aerospace Center (DLR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: Michael Meinel
+# SPDX-FileContributor: Michael Fritzsche
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Callable, Union
+from typing_extensions import Self
+
+from ..types import ld_dict, ld_list
+from ..types.ld_container import BASIC_TYPE, JSON_LD_VALUE, TIME_TYPE
+
+if TYPE_CHECKING:
+ from .container import ld_merge_dict, ld_merge_list
+
+
+class MergeError(ValueError):
+ """ Class for any error while merging. """
+ pass
+
+
+class MergeAction:
+ """ Base class for the different actions occuring druing a merge. """
+ def merge(
+ self: Self,
+ target: ld_merge_dict,
+ key: list[Union[str, int]],
+ value: ld_merge_list,
+ update: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ ) -> Union[JSON_LD_VALUE, BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]:
+ """
+ An abstract method that needs to be implemented by all subclasses
+ to have a generic way to use the merge actions.
+
+ :param target: The ld_merge_dict inside of which the items are merged.
+ :type target: ld_merge_dict
+ :param key: The "path" of keys so that parent[key[-1]] is value and
+ for the outermost parent of target out_parent out_parent[key[0]]...[key[-1]] results in value.
+ :type key: list[str | int]
+ :param value: The value inside target that is to be merged with update.
+ :type value: ld_merge_list
+ :param update: The value that is to be merged into target with value.
+ :type update: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return: The merged value in an arbitrary format that is supported by :meth:`ld_dict.__setitem__`.
+ :rtype: JSON_LD_VALUE | BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+ """
+ raise NotImplementedError()
+
+
+class Reject(MergeAction):
+ def merge(
+ self: Self,
+ target: ld_merge_dict,
+ key: list[Union[str, int]],
+ value: ld_merge_list,
+ update: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ ) -> ld_merge_list:
+ """
+ Rejects the new data ``update`` and lets target add an entry to itself documenting what data has been rejected.
+
+ :param target: The ld_merge_dict inside of which the items are merged.
+ :type target: ld_merge_dict
+ :param key: The "path" of keys so that parent[key[-1]] is value and
+ for the outermost parent of target out_parent out_parent[key[0]]...[key[-1]] results in value.
+ :type key: list[str | int]
+ :param value: The value inside target that is to be merged with update.
This value won't be changed.
+ :type value: ld_merge_list
+ :param update: The value that is to be merged into target with value.
This value will be rejected.
+ :type update: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return: The merged value.
+ This value will always be value.
+ :rtype: ld_merge_list
+ """
+ # Add the entry that data has been rejected.
+ target.reject(key, update)
+ # Return value unchanged.
+ return value
+
+
+class Replace(MergeAction):
+ def merge(
+ self: Self,
+ target: ld_merge_dict,
+ key: list[Union[str, int]],
+ value: ld_merge_list,
+ update: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ ) -> Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]:
+ """
+ Replaces the old data ``value`` with the new data ``update``
+ and lets target add an entry to itself documenting what data has been replaced.
+
+ :param target: The ld_merge_dict inside of which the items are merged.
+ :type target: ld_merge_dict
+ :param key: The "path" of keys so that parent[key[-1]] is value and
+ for the outermost parent of target out_parent out_parent[key[0]]...[key[-1]] results in value.
+ :type key: list[str | int]
+ :param value: The value inside target that is to be merged with update.
This value will bew replaced.
+ :type value: ld_merge_list
+ :param update: The value that is to be merged into target with value.
+ This value will be used instead of value.
+ :type update: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return: The merged value.
+ This value will be update.
+ :rtype: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+ """
+ # If necessary, add the entry that data has been replaced.
+ target.replace(key, value)
+ # Return the new value.
+ return update
+
+
+class Concat(MergeAction):
+ def merge(
+ self: Self,
+ target: ld_merge_dict,
+ key: list[Union[str, int]],
+ value: ld_merge_list,
+ update: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ ) -> ld_merge_list:
+ """
+ Concatenates the new data ``update`` to the old data ``value``.
+
+ :param target: The ld_merge_dict inside of which the items are merged.
+ :type target: ld_merge_dict
+ :param key: The "path" of keys so that parent[key[-1]] is value and
+ for the outermost parent of target out_parent out_parent[key[0]]...[key[-1]] results in value.
+ :type key: list[str | int]
+ :param value: The value inside target that is to be merged with update.
+ :type value: ld_merge_list
+ :param update: The value that is to be merged into target with value.
+ :type update: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return: The merged value.
+ ``value`` concatenated with ``update``.
+ :rtype: ld_merge_list
+ """
+ # Concatenate the items and return the result.
+ if isinstance(update, (list, ld_list)):
+ value.extend(update)
+ else:
+ value.append(update)
+ return value
+
+
+class Collect(MergeAction):
+ def __init__(self: Self, match: Callable[[Any, Any], bool], reject_incoming: bool = True) -> None:
+ """
+ Set the match function for this collect merge action. And the behaivior for matches.
+
+ :param match: The function used to evaluate equality while merging.
+ :type match: Callable[[Any, Any], bool]
+ :param reject_incoming: If an incoming item matches an already collected one, if ``reject_incoming`` True,
+ the incoming item gets rejected, if ``reject_incoming`` False, the match of the incoming item gets replaced.
+ :type reject_incoming: bool
+
+ :return:
+ :rtype: None
+ """
+ self.match = match
+ self.reject_incoming = reject_incoming
+
+ def merge(
+ self: Self,
+ target: ld_merge_dict,
+ key: list[Union[str, int]],
+ value: ld_merge_list,
+ update: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ ) -> ld_merge_list:
+ """
+ Collects the unique items (according to :attr:`match`) from ``value`` and ``update``.
+
+ :param target: The ld_merge_dict inside of which the items are merged.
+ :type target: ld_merge_dict
+ :param key: The "path" of keys so that parent[key[-1]] is value and
+ for the outermost parent of target out_parent out_parent[key[0]]...[key[-1]] results in value.
+ :type key: list[str | int]
+ :param value: The value inside target that is to be merged with update.
+ :type value: ld_merge_list
+ :param update: The value that is to be merged into target with value.
+ :type update: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return: The merged value.
+ :rtype: ld_merge_list
+ """
+ if not isinstance(update, (list, ld_list)):
+ update = [update]
+
+ # iterate over all new items
+ for update_item in update:
+ # Iterate over all items in value and if a match is found replace the first one or reject update_item.
+ for index, item in enumerate(value):
+ if self.match(item, update_item):
+ if not self.reject_incoming:
+ value[index] = update_item
+ break
+ else:
+ # If the current new item has no occurence in value (according to self.match) add it to value.
+ value.append(update_item)
+
+ return value
+
+
+class MergeSet(MergeAction):
+ def __init__(self: Self, match: Callable[[Any, Any], bool]) -> None:
+ """
+ Set the match function for this collect merge action.
+
+ :param match: The function used to evaluate equality while merging.
+ :type match: Callable[[ANy, Any], bool]
+
+ :return:
+ :rtype: None
+ """
+ self.match = match
+
+ def merge(
+ self: Self,
+ target: ld_merge_dict,
+ key: list[Union[str, int]],
+ value: ld_merge_list,
+ update: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ ) -> ld_merge_list:
+ """
+ Merges similar items (according to :attr:`match`) from ``value`` and ``update``.
+
+ :param target: The ld_merge_dict inside of which the items are merged.
+ :type target: ld_merge_dict
+ :param key: The "path" of keys so that parent[key[-1]] is value and
+ for the outermost parent of target out_parent out_parent[key[0]]...[key[-1]] results in value.
+ :type key: list[str | int]
+ :param value: The value inside target that is to be merged with update.
+ :type value: ld_merge_list
+ :param update: The value that is to be merged into target with value.
+ :type update: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return: The merged value.
+ :rtype: ld_merge_list
+ """
+ if not isinstance(update, (list, ld_list)):
+ update = [update]
+
+ for update_item in update:
+ # For each new item merge it into a similar item (according to match) inside target[key[-1]]
+ # (aka inside value) if such an item exists.
+ # Otherwise append it to target[key[-1]] (aka to value).
+ for index, item in enumerate(value):
+ if self.match(item, update_item):
+ if isinstance(item, ld_dict) and isinstance(update_item, ld_dict):
+ item.update(update_item)
+ elif isinstance(item, ld_list) and isinstance(update_item, ld_list):
+ self.merge(target, [*key, index], item, update_item)
+ elif isinstance(item, (ld_dict, ld_list)) or isinstance(update_item, (ld_dict, ld_list)):
+ """ FIXME: log error """
+ break
+ else:
+ value.append(update_item)
+ # Return the merged values.
+ return value
diff --git a/src/hermes/model/merge/container.py b/src/hermes/model/merge/container.py
new file mode 100644
index 00000000..2be14694
--- /dev/null
+++ b/src/hermes/model/merge/container.py
@@ -0,0 +1,349 @@
+# SPDX-FileCopyrightText: 2025 German Aerospace Center (DLR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: Michael Meinel
+# SPDX-FileContributor: Michael Fritzsche
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Callable, Union
+from typing_extensions import Self
+
+from ..types import ld_container, ld_context, ld_dict, ld_list
+from ..types.ld_container import (
+ BASIC_TYPE, EXPANDED_JSON_LD_VALUE, JSON_LD_CONTEXT_DICT, JSON_LD_VALUE, TIME_TYPE
+)
+from ..types.pyld_util import bundled_loader
+from .strategy import CODEMETA_STRATEGY, PROV_STRATEGY
+
+if TYPE_CHECKING:
+ from .action import MergeAction
+
+
+class _ld_merge_container:
+ """
+ Abstract base class for ld_merge_dict and ld_merge_list,
+ providing the merge containers with overrides of ld_container._to_python().
+ See also :class:`ld_dict`, :class:`ld_list` and :class:`ld_container`.
+ """
+
+ def _to_python(
+ self: Self,
+ full_iri: str,
+ ld_value: Union[EXPANDED_JSON_LD_VALUE, dict[str, EXPANDED_JSON_LD_VALUE], list[str], str]
+ ) -> Union["ld_merge_dict", "ld_merge_list", BASIC_TYPE, TIME_TYPE]:
+ """
+ Returns a pythonized version of the given value pretending the value is in self and full_iri its key.
+
+ :param self: the ld_container ld_value is considered to be in.
+ :type self: Self
+ :param full_iri: The expanded iri of the key of ld_value / self (later if self is not a dictionary).
+ :type full_iri: str
+ :param ld_value: The value thats pythonized value is requested. ld_value has to be valid expanded JSON-LD if it
+ was embeded in self._data.
+ :type ld_value: EXPANDED_JSON_LD_VALUE | dict[str, EXPANDED_JSON_LD_VALUE] | list[str] | str
+
+ :return: The pythonized value of the ld_value.
+ :rtype: ld_merge_dict | ld_merge_list | BASIC_TYPE | TIME_TYPE
+ """
+ value = super()._to_python(full_iri, ld_value)
+ # replace ld_dicts with ld_merge_dicts
+ if isinstance(value, ld_dict) and not isinstance(value, ld_merge_dict):
+ value = ld_merge_dict(
+ value.ld_value,
+ parent=value.parent,
+ key=value.key,
+ index=value.index,
+ context=value.context
+ )
+ # replace ld_lists with ld_merge_lists
+ if isinstance(value, ld_list) and not isinstance(value, ld_merge_list):
+ value = ld_merge_list(
+ value.ld_value,
+ parent=value.parent,
+ key=value.key,
+ index=value.index,
+ context=value.context
+ )
+ return value
+
+
+class ld_merge_list(_ld_merge_container, ld_list):
+ """
+ ld_list wrapper to ensure the 'merge_container'-property does not get lost, while merging.
+ See also :class:`ld_list` and :class:`ld_merge_container`.
+ """
+
+ def __init__(
+ self: "ld_merge_list",
+ data: Union[list[str], list[dict[str, EXPANDED_JSON_LD_VALUE]]],
+ *,
+ parent: Union[ld_container, None] = None,
+ key: Union[str, None] = None,
+ index: Union[int, None] = None,
+ context: Union[list[Union[str, JSON_LD_CONTEXT_DICT]], None] = None
+ ) -> None:
+ """
+ Create a new ld_merge_list.
+ For further information on this function and the errors it throws see :meth:`ld_list.__init__`.
+
+ :param self: The instance of ld_merge_list to be initialized.
+ :type self: Self
+ :param data: The expanded json-ld data that is mapped (must be valid for @set, @list or @graph)
+ :type data: list[str] | list[dict[str, BASIC_TYPE | EXPANDED_JSON_LD_VALUE]]
+ :param parent: parent node of this container.
+ :type parent: ld_container | None
+ :param key: key into the parent container.
+ :type key: str | None
+ :param index: index into the parent container.
+ :type index: int | None
+ :param context: local context for this container.
+ :type context: list[str | JSON_LD_CONTEXT_DICT] | None
+
+ :return:
+ :rtype: None
+ """
+ super().__init__(data, parent=parent, key=key, index=index, context=context)
+
+
+class ld_merge_dict(_ld_merge_container, ld_dict):
+ """
+ ld_dict wrapper providing methods to merge an object of this class with an ld_dict object.
+ See also :class:`ld_dict` and :class:`ld_merge_container`.
+
+ :ivar strategies: The strategies for merging different types of values in the ld_dicts.
+ :ivartype strategies: dict[str | None, dict[str | None, MergeAction]]
+ """
+
+ def __init__(
+ self: Self,
+ data: list[dict[str, EXPANDED_JSON_LD_VALUE]],
+ *,
+ parent: Union[ld_dict, ld_list, None] = None,
+ key: Union[str, None] = None,
+ index: Union[int, None] = None,
+ context: Union[list[Union[str, JSON_LD_CONTEXT_DICT]], None] = None
+ ) -> None:
+ """
+ Create a new instance of an ld_merge_dict.
+ See also :meth:`ld_dict.__init__`.
+
+ :param self: The instance of ld_container to be initialized.
+ :type self: Self
+ :param data: The expanded json-ld data that is mapped.
+ :type data: EXPANDED_JSON_LD_VALUE
+ :param parent: parent node of this container.
+ :type parent: ld_dict | ld_list | None
+ :param key: key into the parent container.
+ :type key: str | None
+ :param index: index into the parent container.
+ :type index: int | None
+ :param context: local context for this container.
+ :type context: list[str | JSON_LD_CONTEXT_DICT] | None
+
+ :return:
+ :rtype: None
+
+ :raises ValueError: If the given data doesn't represent an ld_dict.
+ """
+ super().__init__(data, parent=parent, key=key, index=index, context=context)
+
+ # add provernance context
+ self.update_context(ld_context.HERMES_PROV_CONTEXT)
+
+ # add strategies
+ self.strategies = {**CODEMETA_STRATEGY}
+ self.add_strategy(PROV_STRATEGY)
+
+ def update_context(
+ self: Self, other_context: Union[list[Union[str, JSON_LD_CONTEXT_DICT]], None]
+ ) -> None:
+ """
+ Updates selfs context with other_context.
+ JSON-LD processing prioritizes the context values in order (first least important, last most important).
+
+ :param self: The instance of the ld_merge_dict context is added to.
+ :type self: Self
+ :param other_context: The context object that is added to selfs context.
+ :type other_context: list[str | JSON_LD_CONTEXT_DICT] | None
+
+ :return:
+ :rtype: None
+ """
+ if other_context:
+ if not isinstance(self.context, list):
+ self.context = [self.context]
+ if isinstance(other_context, list):
+ self.context = [*other_context, *self.context]
+ else:
+ self.context = [other_context, *self.context]
+
+ # update the active context that is used for compaction/ expansion
+ self.active_ctx = self.ld_proc.initial_ctx(self.context, {"documentLoader": bundled_loader})
+
+ def update(self: Self, other: ld_dict) -> None:
+ """
+ Updates/ Merges this ld_merge dict with the given ld_dict other.
+ This overwrites :meth:`ld_dict.update`, and may cause unexpected behavior if not used carefully.
+
+ :param self: The ld_merge_dict that is updated with other.
+ :type self: Self
+ :param other: The ld_container that is merged into self.
+ :type other: ld_dict
+
+ :return:
+ :rtype: None
+ """
+ # update add all new context
+ if isinstance(other, ld_dict):
+ self.update_context(other.context)
+
+ # add the acutal values based on the MergeAction strategies
+ # this works implicitly because ld_dict.update invokes self.__setitem__ which is overwritten by ld_merge_dict
+ super().update(other)
+
+ def add_strategy(self: Self, strategy: dict[Union[str, None], dict[Union[str, None], MergeAction]]) -> None:
+ """
+ Adds the given strategy to the self.strategies.
+
+ :param self: The ld_merge_dict the strategy is added to.
+ :type self: Self
+ :param strategy: The object describing how which object types are supposed to be merged.
+ :type strategy: dict[str | None, dict[str | None, MergeAction]]
+ """
+ for key, value in strategy.items():
+ self.strategies[key] = {**value, **self.strategies.get(key, {})}
+
+ def __setitem__(self: Self, key: str, value: Union[JSON_LD_VALUE, BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]):
+ """
+ Creates the new entry for self[key] using self.strategies on the values in self[key] and value.
+ Wraps :meth:`ld_dict.__setitem__`, and may cause unexpected behavior if not used carefully.
+
+ :param self: The ld_merge_dict whose value at key gets updated/ merged with value.
+ :type self: Self
+ :param key: The key at whicht the value is updated/ merged at in self.
+ :type key: str
+ :param value: The value that is merged into self[key].
+ :type value: JSON_LD_VALUE | BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+ """
+ # create the new item if self[key] and value have to be merged.
+ if key in self:
+ value = self._merge_item(key, value)
+ # update the entry of self[key]
+ super().__setitem__(key, value)
+
+ def match(
+ self: Self,
+ key: str,
+ value: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list],
+ match: Callable[[Any, Any], bool]
+ ) -> Union[BASIC_TYPE, TIME_TYPE, "ld_merge_dict", ld_merge_list]:
+ """
+ Returns the first item in self[key] for which match(item, value) returns true.
+ If no such item is found None is returned instead.
+
+ :param self: The ld_merge_dict in whose entry for key a match for value is searched.
+ :type self: Self
+ :param key: The key to the items in self in which a match for value is searched.
+ :type key: str
+ :param value: The value a match is searched for in self[key].
+ :type value: Union[JSON_LD_VALUE, BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ :param match: The method defining if two objects are a match.
+ :type match: Callable[[Any, Any], bool]
+
+ :return: The item in self[key] that is a match to value if one exists else None
+ :rtype: BASIC_TYPE | TIME_TYPE | ld_merge_dict | ld_merge_list
+ """
+ # iterate over all items in self[key] and return the first that is a match
+ for item in self[key]:
+ if match(item, value):
+ return item
+
+ def _merge_item(
+ self: Self, key: str, value: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ ) -> Union[BASIC_TYPE, TIME_TYPE, "ld_merge_dict", ld_merge_list]:
+ """
+ Applies the most suitable merge strategy to merge self[key] and value and then returns the result.
+
+ :param self: The ld_merge_dict whose entry at key is to be merged with value.
+ :type self: Self
+ :param key: The key to the entry in self that is to be merged with value.
+ :type key: str
+ :param value: The value that is to be merged with self[key].
+ :type value: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return: The result of the merge from self[key] with value.
+ :rtype: BASIC_TYPE | TIME_TYPE | ld_merge_dict | ld_merge_list
+ """
+ # search for all applicable strategies
+ strategy = {**self.strategies[None]}
+ ld_types = self.data_dict.get('@type', [])
+ for ld_type in ld_types:
+ strategy.update(self.strategies.get(ld_type, {}))
+
+ # choose one merge strategy and return the item returned by following the merge startegy
+ merger = strategy.get(key, strategy[None])
+ return merger.merge(self, [*self.path, key], self[key], value)
+
+ def _add_related(
+ self: Self, rel: str, key: str, value: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]
+ ) -> None:
+ """
+ Adds an entry for rel to self containing which key and value is affected.
+
+ :param self: The ld_merge_container the special entry is added to.
+ :type self: Self
+ :param rel: The "type" of the special entry (used as the key).
+ :type rel: str
+ :param key: The key of the affected key, value pair in self.
+ :type key: str
+ :param value: The value of the affected key, value pair in self.
+ :type value: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return:
+ :rtype: None
+ """
+ # FIXME: key not only string
+ # make sure appending is possible
+ self.emplace(rel)
+ # append the new entry
+ self[rel].append({"@type": "schema:PropertyValue", "schema:name": str(key), "schema:value": str(value)})
+
+ def reject(self: Self, key: str, value: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]) -> None:
+ """
+ Adds an entry to self containing containing information that the key, value pair
+ key, value has been rejected in the merge.
+ For further information see :meth:`ld_merge_dict._add_related`.
+
+ :param self: The ld_merge_container the special entry is added to.
+ :type self: Self
+ :param key: The key of the rejected key, value pair in self.
+ :type key: str
+ :param value: The value of the rejected key, value pair in self.
+ :type value: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return:
+ :rtype: None
+ """
+ # FIXME: key not only string
+ self._add_related("hermes-rt:reject", key, value)
+
+ def replace(self: Self, key: str, value: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]) -> None:
+ """
+ Adds an entry to self containing containing information that the key, value pair
+ key, value was replaced in the merge.
+ For further information see :meth:`ld_merge_dict._add_related`.
+
+ :param self: The ld_merge_container the special entry is added to.
+ :type self: Self
+ :param key: The key of the old key, value pair in self.
+ :type key: str
+ :param value: The value of the old key, value pair in self.
+ :type value: BASIC_TYPE | TIME_TYPE | ld_dict | ld_list
+
+ :return:
+ :rtype: None
+ """
+ # FIXME: key not only string
+ self._add_related("hermes-rt:replace", key, value)
diff --git a/src/hermes/model/merge/match.py b/src/hermes/model/merge/match.py
new file mode 100644
index 00000000..cbcad94d
--- /dev/null
+++ b/src/hermes/model/merge/match.py
@@ -0,0 +1,134 @@
+# SPDX-FileCopyrightText: 2025 German Aerospace Center (DLR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: Michael Meinel
+# SPDX-FileContributor: Michael Fritzsche
+
+from typing import Any, Callable
+
+from ..types import ld_dict
+
+
+def match_keys(*keys: list[str], fall_back_to_equals: bool = False) -> Callable[[Any, Any], bool]:
+ """
+ Creates a function taking to parameters that returns true
+ if both given parameter have at least one common key in the given list of keys
+ and for all common keys in the given list of keys the values of both objects are the same.
+ If fall_back_to_equals is True, the returned function returns the value of normal == comparison
+ if no key from keys is in both objects.
+
+ :param keys: The list of important keys for the comparison method.
+ :type keys: list[str]
+ :param fall_back_to_equals: Whether or not a fall back option should be used.
+ :type fall_back_to_equals: bool
+
+ :return: A function comparing two given objects values for the keys in keys.
+ :rtype: Callable[[ld_merge_dict, ld_dict], bool]
+ """
+
+ # create and return the match function using the given keys
+ def match_func(left: Any, right: Any) -> bool:
+ """
+ Compares left to right by checking if a) they have at least one common key in a predetermined list of keys and
+ b) testing if both objects have equal values for all common keys in the predetermined key list.
+ It may fall back on == if no common key in the predetermined list of keys exists.
+
+ :param left: The first object for the comparison.
+ :type left: ld_merge_dict
+ :param right: The second object for the comparison.
+ :type right: ld_dict
+
+ :return: The result of the comparison.
+ :rtype: bool
+ """
+ if not (isinstance(left, ld_dict) and isinstance(right, ld_dict)):
+ return fall_back_to_equals and (left == right)
+ # create a list of all common important keys
+ active_keys = [key for key in keys if key in left and key in right]
+ # fall back to == if no active keys
+ if fall_back_to_equals and not active_keys:
+ return left == right
+ # check if both objects have the same values for all active keys
+ pairs = [(left[key] == right[key]) for key in active_keys]
+ # return whether or not both objects had the same values for all active keys
+ # and there was at least one active key
+ return len(active_keys) > 0 and all(pairs)
+ return match_func
+
+
+def match_person(left: Any, right: Any) -> bool:
+ """
+ Compares two objects assuming they are representing schema:Person's
+ if they are not ld_dicts, == is used as a fallback.
+ If both objects have an @id value, the truth value returned by this function is the comparison of both ids.
+ If either other has no @id value and both objects have at least one email value,
+ they are considered equal if they have one common email.
+ If the equality of the objects is not yet decided, == comparison of the objects is returned.
+
+ :param left: The first object for the comparison.
+ :type left: ld_merge_dict
+ :param right: The second object for the comparison.
+ :type right: ld_dict
+
+ :return: The result of the comparison.
+ :rtype: bool
+ """
+ if not (isinstance(left, ld_dict) and isinstance(right, ld_dict)):
+ return left == right
+ if "@id" in left and "@id" in right:
+ return left["@id"] == right["@id"]
+ if "schema:email" in left and "schema:email" in right:
+ if len(left["schema:email"]) > 0 and len(right["schema:email"]) > 0:
+ mails_right = right["schema:email"]
+ return any((mail in mails_right) for mail in left["schema:email"])
+ return left == right
+
+
+def match_multiple_types(
+ *functions_for_types: list[tuple[str, Callable[[Any, Any], bool]]],
+ fall_back_function: Callable[[Any, Any], bool] = match_keys("@id", fall_back_to_equals=True)
+) -> Callable[[Any, Any], bool]:
+ """
+ Returns a function that compares two objects using the given functions.
+
+ :param functions_for_types: Tuples of type and match_function.
+ The returned function will compare two objects of a the same, given type with the specified function.
+ :type functions_for_types: list[tuple[str, Callable[[Any, Any], bool]]]
+ :param fall_back_function: The fallback for comparison if the objects that are being compared don't have a common
+ type with specified compare function or at least one object is not a JSON-LD dictionary.
+ :type fall_back_function: Callable[[Any, Any], bool]
+
+ :return: The function that compares the two given objects using the given functions.
+ :rtype: Callable[[Any, Any], bool]
+ """
+
+ # create and return the match function using the given keys
+ def match_func(left: Any, right: Any) -> bool:
+ """
+ Compares two objects using a predetermined function if either objects is not an ld_dict
+ or they don't have a common type in a predetermined list of types.
+ If the objects are ld_dicts and have the same type with a known comparison function this is used instead.
+
+ :param left: The first object for the comparison.
+ :type left: ld_merge_dict
+ :param right: The second object for the comparison.
+ :type right: ld_dict
+
+ :return: The result of the comparison.
+ :rtype: bool
+ """
+ # If at least one of the objects is not an ld_dict or contains no value for the key "@type", use the fallback.
+ if not (isinstance(left, ld_dict) and isinstance(right, ld_dict) and "@type" in left and "@type" in right):
+ return fall_back_function(left, right)
+ # Extract the list of types
+ types_left = left["@type"]
+ types_right = right["@type"]
+ # Iterate over all known type, match_function pairs.
+ # If one type is in both objects return the result of the comparison with the match_function.
+ for ld_type, func in functions_for_types:
+ if ld_type in types_left and ld_type in types_right:
+ return func(left, right)
+ # No common type with known match_function: Fallback
+ return fall_back_function(left, right)
+ return match_func
diff --git a/src/hermes/model/merge/strategy.py b/src/hermes/model/merge/strategy.py
new file mode 100644
index 00000000..ac78545c
--- /dev/null
+++ b/src/hermes/model/merge/strategy.py
@@ -0,0 +1,708 @@
+# SPDX-FileCopyrightText: 2025 German Aerospace Center (DLR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: Michael Meinel
+# SPDX-FileContributor: Michael Fritzsche
+
+from ..types.ld_context import iri_map as iri
+from .action import Concat, MergeSet
+from .match import match_keys, match_person, match_multiple_types
+
+
+DEFAULT_MATCH = match_keys("@id", fall_back_to_equals=True)
+
+MATCH_FUNCTION_FOR_TYPE = {"schema:Person": match_person}
+
+ACTIONS = {
+ "default": MergeSet(DEFAULT_MATCH),
+ "concat": Concat(),
+ "Person": MergeSet(MATCH_FUNCTION_FOR_TYPE["schema:Person"]),
+ **{
+ "Or".join(types): MergeSet(match_multiple_types(
+ *(("schema:" + type, MATCH_FUNCTION_FOR_TYPE.get("schema:" + type, DEFAULT_MATCH)) for type in types)
+ ))
+ for types in [
+ ("AboutPage", "CreativeWork"),
+ ("AdministrativeArea", "GeoShape", "Place"),
+ ("AggregateOffer", "CreativeWork", "Event", "MenuItem", "Product", "Service", "Trip"),
+ ("AnatomicalStructure", "AnatomicalSystem"),
+ ("AnatomicalStructure", "AnatomicalSystem", "BioChemEntity", "DefinedTerm"),
+ ("AnatomicalStructure", "AnatomicalSystem", "SuperficialAnatomy"),
+ ("AudioObject", "Clip", "MusicRecording"),
+ ("BioChemEntity", "CreativeWork", "Event", "MedicalEntity", "Organization", "Person", "Product"),
+ ("Brand", "Organization"),
+ ("CategoryCode", "Thing"),
+ ("Class", "Enumeration"),
+ ("Class", "Enumeration", "Property"),
+ ("Clip", "VideoObject"),
+ ("Comment", "CreativeWork"),
+ ("ContactPoint", "Place"),
+ ("CreativeWork", "HowToSection", "HowToStep"),
+ ("CreativeWork", "Product"),
+ ("CreditCard", "MonetaryAmount", "UnitPriceSpecification"),
+ ("DataFeedItem", "Thing"),
+ ("Demand", "Offer"),
+ ("DefinedTerm", "Enumeration", "PropertyValue", "QualitativeValue", "QuantitativeValue", "StructuredValue"),
+ ("DefinedTerm", "PropertyValue"),
+ ("DefinedTerm", "QuantitativeValue", "SizeSpecification"),
+ ("DefinedTerm", "StructuredValue"),
+ ("DefinedTerm", "Taxon"),
+ ("Distance", "QuantitativeValue"),
+ ("Drug", "DrugClass", "LifestyleModification", "MedicalTherapy"),
+ ("Duration", "QuantitativeValue"),
+ ("EducationalOrganization", "Organization"),
+ ("GeoCoordinates", "GeoShape"),
+ ("GeoShape", "Place"),
+ ("GeospatialGeometry", "Place"),
+ ("ImageObject", "Photograph"),
+ ("ItemList", "ListItem", "WebContent"),
+ ("ItemList", "MusicRecording"),
+ ("ListItem", "Thing"),
+ ("LoanOrCredit", "PaymentMethod"),
+ ("Mass", "QuantitativeValue"),
+ ("MedicalCondition", "PropertyValue"),
+ ("MemberProgramTier", "Organization", "ProgramMembership"),
+ ("MenuItem", "MenuSection"),
+ ("MonetaryAmount", "MonetaryAmountDistribution"),
+ ("MonetaryAmount", "PriceSpecification"),
+ ("MonetaryAmount", "ShippingRateSettings"),
+ ("MusicGroup", "Person"),
+ ("Organization", "Person"),
+ ("PerformingGroup", "Person"),
+ ("Place", "PostalAddress", "VirtualLocation"),
+ ("ProductGroup", "ProductModel"),
+ ("Property", "PropertyValue", "StatisticalVariable"),
+ ("Product", "Service"),
+ ("QuantitativeValue", "ServicePeriod"),
+ ("SoftwareApplication", "WebSite")
+ ]
+ }
+}
+
+
+PROV_STRATEGY = {
+ None: {
+ iri["hermes-rt:graph"]: ACTIONS["concat"],
+ iri["hermes-rt:replace"]: ACTIONS["concat"],
+ iri["hermes-rt:reject"]: ACTIONS["concat"]
+ }
+}
+
+
+# Filled with entries for every schema-type that can be found inside an JSON-LD dict of type
+# SoftwareSourceCode or SoftwareApplication using schema and CodeMeta as Context.
+CODEMETA_STRATEGY = {None: {None: ACTIONS["default"]}}
+CODEMETA_STRATEGY[iri["schema:Thing"]] = {iri["schema:owner"]: ACTIONS["OrganizationOrPerson"]}
+
+
+CODEMETA_STRATEGY[iri["schema:Action"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Thing"]],
+ iri["schema:agent"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:location"]: ACTIONS["PlaceOrPostalAddressOrVirtualLocation"],
+ iri["schema:participant"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:provider"]: ACTIONS["OrganizationOrPerson"]
+}
+
+
+CODEMETA_STRATEGY[iri["schema:BioChemEntity"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Thing"]],
+ iri["schema:associatedDisease"]: ACTIONS["MedicalConditionOrPropertyValue"],
+ iri["schema:hasMolecularFunction"]: ACTIONS["DefinedTermOrPropertyValue"],
+ iri["schema:isInvolvedInBiologicalProcess"]: ACTIONS["DefinedTermOrPropertyValue"],
+ iri["schema:isLocatedInSubcellularLocation"]: ACTIONS["DefinedTermOrPropertyValue"],
+ iri["schema:taxonomicRange"]: ACTIONS["DefinedTermOrTaxon"]
+}
+
+CODEMETA_STRATEGY[iri["schema:Gene"]] = {
+ **CODEMETA_STRATEGY[iri["schema:BioChemEntity"]],
+ iri["schema:expressedIn"]: ACTIONS["AnatomicalStructureOrAnatomicalSystemOrBioChemEntityOrDefinedTerm"]
+}
+
+
+CODEMETA_STRATEGY[iri["schema:CreativeWork"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Thing"]],
+ iri["schema:accountablePerson"]: ACTIONS["Person"],
+ iri["schema:audio"]: ACTIONS["AudioObjectOrClipOrMusicRecording"],
+ iri["schema:author"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:character"]: ACTIONS["Person"],
+ iri["schema:contributor"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:copyrightHolder"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:creator"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:editor"]: ACTIONS["Person"],
+ iri["schema:funder"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:isBasedOn"]: ACTIONS["CreativeWorkOrProduct"],
+ iri["schema:maintainer"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:offers"]: ACTIONS["DemandOrOffer"],
+ iri["schema:producer"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:provider"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:publisher"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:sdPublisher"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:size"]: ACTIONS["DefinedTermOrQuantitativeValueOrSizeSpecification"],
+ iri["schema:sponsor"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:translator"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:video"]: ACTIONS["ClipOrVideoObject"]
+}
+
+CODEMETA_STRATEGY[iri["schema:Article"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:NewsArticle"]] = {**CODEMETA_STRATEGY[iri["schema:Article"]]}
+CODEMETA_STRATEGY[iri["schema:ScholarlyArticle"]] = {**CODEMETA_STRATEGY[iri["schema:Article"]]}
+CODEMETA_STRATEGY[iri["schema:Certification"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:Claim"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:claimInterpreter"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:Clip"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:actor"]: ACTIONS["PerformingGroupOrPerson"],
+ iri["schema:dircetor"]: ACTIONS["Person"],
+ iri["schema:musicBy"]: ACTIONS["MusicGroupOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:Comment"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:parentItem"]: ACTIONS["CommentOrCreativeWork"]
+}
+CODEMETA_STRATEGY[iri["schema:CorrectionComment"]] = {**CODEMETA_STRATEGY[iri["schema:Comment"]]}
+CODEMETA_STRATEGY[iri["schema:CreativeWorkSeason"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:actor"]: ACTIONS["PerformingGroupOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:DataCatalog"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:Dataset"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:variableMeasured"]: ACTIONS["PropertyOrPropertyValueOrStatisticalVariable"]
+}
+CODEMETA_STRATEGY[iri["schema:DataFeed"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Dataset"]],
+ iri["schema:dataFeedElement"]: ACTIONS["DataFeedItemOrThing"]
+}
+CODEMETA_STRATEGY[iri["schema:DefinedTermSet"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:CategoryCodeSet"]] = {**CODEMETA_STRATEGY[iri["schema:DefinedTermSet"]]}
+CODEMETA_STRATEGY[iri["schema:EducationalOccupationalCredential"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:Episode"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:actor"]: ACTIONS["PerformingGroupOrPerson"],
+ iri["schema:dircetor"]: ACTIONS["Person"],
+ iri["schema:duration"]: ACTIONS["DurationOrQuantitativeValue"],
+ iri["schema:musicBy"]: ACTIONS["MusicGroupOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:HowTo"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:step"]: ACTIONS["CreativeWorkOrHowToSectionOrHowToStep"]
+}
+CODEMETA_STRATEGY[iri["schema:HyperTocEntry"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:Map"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:MediaObject"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:duration"]: ACTIONS["DurationOrQuantitativeValue"],
+ iri["schema:height"]: ACTIONS["DistanceOrQuantitativeValue"],
+ iri["schema:ineligibleRegion"]: ACTIONS["GeoShapeOrPlace"],
+ iri["schema:width"]: ACTIONS["DistanceOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:AudioObject"]] = {**CODEMETA_STRATEGY[iri["schema:MediaObject"]]}
+CODEMETA_STRATEGY[iri["schema:DataDownload"]] = {**CODEMETA_STRATEGY[iri["schema:MediaObject"]]}
+CODEMETA_STRATEGY[iri["schema:ImageObject"]] = {**CODEMETA_STRATEGY[iri["schema:MediaObject"]]}
+CODEMETA_STRATEGY[iri["schema:VideoObject"]] = {
+ **CODEMETA_STRATEGY[iri["schema:MediaObject"]],
+ iri["schema:actor"]: ACTIONS["PerformingGroupOrPerson"],
+ iri["schema:dircetor"]: ACTIONS["Person"],
+ iri["schema:musicBy"]: ACTIONS["MusicGroupOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:MenuSection"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:MusicComposition"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:composer"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:lyricist"]: ACTIONS["Person"]
+}
+CODEMETA_STRATEGY[iri["schema:MusicPlaylist"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:track"]: ACTIONS["ItemListOrMusicRecording"]
+}
+CODEMETA_STRATEGY[iri["schema:MusicAlbum"]] = {
+ **CODEMETA_STRATEGY[iri["schema:MusicPlaylist"]],
+ iri["schema:byArtist"]: ACTIONS["MusicGroupOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:MusicRelease"]] = {
+ **CODEMETA_STRATEGY[iri["schema:MusicPlaylist"]],
+ iri["schema:creditedTo"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:duration"]: ACTIONS["DurationOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:MusicRecording"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:byArtist"]: ACTIONS["MusicGroupOrPerson"],
+ iri["schema:duration"]: ACTIONS["DurationOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:Photograph"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:Review"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:negativeNotes"]: ACTIONS["ItemListOrListItemOrWebContent"],
+ iri["schema:positiveNotes"]: ACTIONS["ItemListOrListItemOrWebContent"]
+}
+CODEMETA_STRATEGY[iri["schema:SoftwareApplication"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:OperatingSystem"]] = {**CODEMETA_STRATEGY[iri["schema:SoftwareApplication"]]}
+CODEMETA_STRATEGY[iri["schema:RuntimePlatform"]] = {**CODEMETA_STRATEGY[iri["schema:SoftwareApplication"]]}
+CODEMETA_STRATEGY[iri["schema:SoftwareSourceCode"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["maintainer"]: ACTIONS["Person"]
+}
+CODEMETA_STRATEGY[iri["schema:WebContent"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:WebPage"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ iri["schema:reviewedBy"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:AboutPage"]] = {**CODEMETA_STRATEGY[iri["schema:WebPage"]]}
+CODEMETA_STRATEGY[iri["schema:WebPageElement"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+CODEMETA_STRATEGY[iri["schema:WebSite"]] = {**CODEMETA_STRATEGY[iri["schema:CreativeWork"]]}
+
+
+CODEMETA_STRATEGY[iri["schema:Event"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Thing"]],
+ iri["schema:actor"]: ACTIONS["PerformingGroupOrPerson"],
+ iri["schema:attendee"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:composer"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:contributor"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:dircetor"]: ACTIONS["Person"],
+ iri["schema:duration"]: ACTIONS["DurationOrQuantitativeValue"],
+ iri["schema:funder"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:location"]: ACTIONS["PlaceOrPostalAddressOrVirtualLocation"],
+ iri["schema:offers"]: ACTIONS["DemandOrOffer"],
+ iri["schema:organizer"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:performer"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:sponsor"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:translator"]: ACTIONS["OrganizationOrPerson"]
+}
+
+CODEMETA_STRATEGY[iri["schema:PublicationEvent"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Event"]],
+ iri["schema:publishedBy"]: ACTIONS["OrganizationOrPerson"]
+}
+
+
+CODEMETA_STRATEGY[iri["schema:Intangible"]] = {**CODEMETA_STRATEGY[iri["schema:Thing"]]}
+
+CODEMETA_STRATEGY[iri["schema:AlignmentObject"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:Audience"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:Brand"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:BroadcastChannel"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:BroadcastFrequencySpecification"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:Class"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:supersededBy"]: ACTIONS["ClassOrEnumeration"]
+}
+CODEMETA_STRATEGY[iri["schema:ComputerLanguage"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:ConstraintNode"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:StatisticalVariable"]] = {**CODEMETA_STRATEGY[iri["schema:ConstraintNode"]]}
+CODEMETA_STRATEGY[iri["schema:DefinedTerm"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:CategoryCode"]] = {**CODEMETA_STRATEGY[iri["schema:DefinedTerm"]]}
+CODEMETA_STRATEGY[iri["schema:Demand"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:acceptedPaymentMethod"]: ACTIONS["LoanOrCreditOrPaymentMethod"],
+ iri["schema:areaServed"]: ACTIONS["AdministrativeAreaOrGeoShapeOrPlace"],
+ iri["schema:eligibleRegion"]: ACTIONS["GeoShapeOrPlace"],
+ iri["schema:ineligibleRegion"]: ACTIONS["GeoShapeOrPlace"],
+ iri["schema:itemOffered"]: ACTIONS["AggregateOfferOrCreativeWorkOrEventOrMenuItemOrProductOrServiceOrTrip"],
+ iri["schema:seller"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:EnergyConsumptionDetails"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:EntryPoint"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:Enumeration"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:supersededBy"]: ACTIONS["ClassOrEnumeration"]
+}
+CODEMETA_STRATEGY[iri["schema:QualitativeValue"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Enumeration"]],
+ iri[
+ "schema:valueReference"
+ ]: ACTIONS["DefinedTermOrEnumerationOrPropertyValueOrQualitativeValueOrQuantitativeValueOrStructuredValue"]
+}
+CODEMETA_STRATEGY[iri["schema:SizeSpecification"]] = {**CODEMETA_STRATEGY[iri["schema:QualitativeValue"]]}
+CODEMETA_STRATEGY[iri["schema:GeospatialGeometry"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:geoContains"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoCoveredBy"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoCovers"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoCrosses"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoDisjoint"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoEquals"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoIntersects"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoOverlaps"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoTouches"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoWithin"]: ACTIONS["GeospatialGeometryOrPlace"]
+}
+CODEMETA_STRATEGY[iri["schema:Grant"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri[
+ "schema:fundedItem"
+ ]: ACTIONS["BioChemEntityOrCreativeWorkOrEventOrMedicalEntityOrOrganizationOrPersonOrProduct"],
+ iri["schema:funder"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:sponsor"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:HealthInsurancePlan"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:HealthPlanCostSharingSpecification"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:HealthPlanFormulary"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:HealthPlanNetwork"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:ItemList"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:itemListElement"]: ACTIONS["ListItemOrThing"]
+}
+CODEMETA_STRATEGY[iri["schema:OfferCatalog"]] = {**CODEMETA_STRATEGY[iri["schema:ItemList"]]}
+CODEMETA_STRATEGY[iri["schema:BreadcrumbList"]] = {**CODEMETA_STRATEGY[iri["schema:ItemList"]]}
+CODEMETA_STRATEGY[iri["schema:Language"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:ListItem"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:HowToItem"]] = {**CODEMETA_STRATEGY[iri["schema:ListItem"]]}
+CODEMETA_STRATEGY[iri["schema:HowToSupply"]] = {**CODEMETA_STRATEGY[iri["schema:HowToItem"]]}
+CODEMETA_STRATEGY[iri["schema:HowToTool"]] = {**CODEMETA_STRATEGY[iri["schema:HowToItem"]]}
+CODEMETA_STRATEGY[iri["schema:MediaSubscription"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:MemberProgram"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:MemberProgramTier"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:hasTierRequirement"]: ACTIONS["CreditCardOrMonetaryAmountOrUnitPriceSpecification"]
+}
+CODEMETA_STRATEGY[iri["schema:MenuItem"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:menuAddOn"]: ACTIONS["MenuItemOrMenuSection"],
+ iri["schema:offers"]: ACTIONS["DemandOrOffer"]
+}
+CODEMETA_STRATEGY[iri["schema:MerchantReturnPolicy"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:MerchantReturnPolicySeasonalOverride"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:Occupation"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:estimatedSalary"]: ACTIONS["MonetaryAmountOrMonetaryAmountDistribution"]
+}
+CODEMETA_STRATEGY[iri["schema:OccupationalExperienceRequirements"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:Offer"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:acceptedPaymentMethod"]: ACTIONS["LoanOrCreditOrPaymentMethod"],
+ iri["schema:areaServed"]: ACTIONS["AdministrativeAreaOrGeoShapeOrPlace"],
+ iri["schema:category"]: ACTIONS["CategoryCodeOrThing"],
+ iri["schema:eligibleRegion"]: ACTIONS["GeoShapeOrPlace"],
+ iri["schema:ineligibleRegion"]: ACTIONS["GeoShapeOrPlace"],
+ iri["schema:itemOffered"]: ACTIONS["AggregateOfferOrCreativeWorkOrEventOrMenuItemOrProductOrServiceOrTrip"],
+ iri["schema:leaseLength"]: ACTIONS["DurationOrQuantitativeValue"],
+ iri["schema:offeredBy"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:seller"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:AggregateOffer"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Offer"]],
+ iri["schema:offers"]: ACTIONS["DemandOrOffer"]
+}
+CODEMETA_STRATEGY[iri["schema:PaymentMethod"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:ProgramMembership"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:member"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:Property"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:supersededBy"]: ACTIONS["ClassOrEnumerationOrProperty"]
+}
+CODEMETA_STRATEGY[iri["schema:Quantity"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:Duration"]] = {**CODEMETA_STRATEGY[iri["schema:Quantity"]]}
+CODEMETA_STRATEGY[iri["schema:Energy"]] = {**CODEMETA_STRATEGY[iri["schema:Quantity"]]}
+CODEMETA_STRATEGY[iri["schema:Mass"]] = {**CODEMETA_STRATEGY[iri["schema:Quantity"]]}
+CODEMETA_STRATEGY[iri["schema:Rating"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:author"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:AggregateRating"]] = {**CODEMETA_STRATEGY[iri["schema:Rating"]]}
+CODEMETA_STRATEGY[iri["schema:Schedule"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:duration"]: ACTIONS["DurationOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:Series"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:Service"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Intangible"]],
+ iri["schema:areaServed"]: ACTIONS["AdministrativeAreaOrGeoShapeOrPlace"],
+ iri["schema:brand"]: ACTIONS["BrandOrOrganization"],
+ iri["schema:broker"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:category"]: ACTIONS["CategoryCodeOrThing"],
+ iri["schema:isRelatedTo"]: ACTIONS["ProductOrService"],
+ iri["schema:isSimilarTo"]: ACTIONS["ProductOrService"],
+ iri["schema:offers"]: ACTIONS["DemandOrOffer"],
+ iri["schema:provider"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:BroadcastService"]] = {**CODEMETA_STRATEGY[iri["schema:Service"]]}
+CODEMETA_STRATEGY[iri["schema:CableOrSatelliteService"]] = {**CODEMETA_STRATEGY[iri["schema:Service"]]}
+CODEMETA_STRATEGY[iri["schema:FinancialProduct"]] = {**CODEMETA_STRATEGY[iri["schema:Service"]]}
+CODEMETA_STRATEGY[iri["schema:LoanOrCredit"]] = {**CODEMETA_STRATEGY[iri["schema:FinancialProduct"]]}
+CODEMETA_STRATEGY[iri["schema:ServiceChannel"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:SpeakableSpecification"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:StructuredValue"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+CODEMETA_STRATEGY[iri["schema:ContactPoint"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:areaServed"]: ACTIONS["AdministrativeAreaOrGeoShapeOrPlace"]
+}
+CODEMETA_STRATEGY[iri["schema:PostalAddress"]] = {**CODEMETA_STRATEGY[iri["schema:ContactPoint"]]}
+CODEMETA_STRATEGY[iri["schema:Distance"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:GeoCoordinates"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:GeoShape"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:InteractionCounter"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:interactionService"]: ACTIONS["SoftwareApplicationOrWebSite"],
+ iri["schema:location"]: ACTIONS["PlaceOrPostalAddressOrVirtualLocation"]
+}
+CODEMETA_STRATEGY[iri["schema:MonetaryAmount"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:NutritionInformation"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:OfferShippingDetails"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:depth"]: ACTIONS["DistanceOrQuantitativeValue"],
+ iri["schema:height"]: ACTIONS["DistanceOrQuantitativeValue"],
+ iri["schema:shippingRate"]: ACTIONS["MonetaryAmountOrShippingRateSettings"],
+ iri["schema:weight"]: ACTIONS["MassOrQuantitativeValue"],
+ iri["schema:width"]: ACTIONS["DistanceOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:OpeningHoursSpecification"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:PostalCodeRangeSpecification"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:PriceSpecification"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:DeliveryChargeSpecification"]] = {
+ **CODEMETA_STRATEGY[iri["schema:PriceSpecification"]],
+ iri["schema:areaServed"]: ACTIONS["AdministrativeAreaOrGeoShapeOrPlace"],
+ iri["schema:eligibleRegion"]: ACTIONS["GeoShapeOrPlace"],
+ iri["schema:ineligibleRegion"]: ACTIONS["GeoShapeOrPlace"]
+}
+CODEMETA_STRATEGY[iri["schema:UnitPriceSpecification"]] = {
+ **CODEMETA_STRATEGY[iri["schema:PriceSpecification"]],
+ iri["schema:billingDuration"]: ACTIONS["DurationOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:PropertyValue"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri[
+ "schema:valueReference"
+ ]: ACTIONS["DefinedTermOrEnumerationOrPropertyValueOrQualitativeValueOrQuantitativeValueOrStructuredValue"]
+}
+CODEMETA_STRATEGY[iri["schema:LocationFeatureSpecification"]] = {**CODEMETA_STRATEGY[iri["schema:PropertyValue"]]}
+CODEMETA_STRATEGY[iri["schema:QuantitativeValue"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri[
+ "schema:valueReference"
+ ]: ACTIONS["DefinedTermOrEnumerationOrPropertyValueOrQualitativeValueOrQuantitativeValueOrStructuredValue"]
+}
+CODEMETA_STRATEGY[iri["schema:QuantitativeValueDistribution"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:duration"]: ACTIONS["DurationOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:MonetaryAmountDistribution"]] = {
+ **CODEMETA_STRATEGY[iri["schema:QuantitativeValueDistribution"]]
+}
+CODEMETA_STRATEGY[iri["schema:RepaymentSpecification"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:ServicePeriod"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:duration"]: ACTIONS["DurationOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:ShippingConditions"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:depth"]: ACTIONS["DistanceOrQuantitativeValue"],
+ iri["schema:height"]: ACTIONS["DistanceOrQuantitativeValue"],
+ iri["schema:shippingRate"]: ACTIONS["MonetaryAmountOrShippingRateSettings"],
+ iri["schema:transitTime"]: ACTIONS["QuantitativeValueOrServicePeriod"],
+ iri["schema:weight"]: ACTIONS["MassOrQuantitativeValue"],
+ iri["schema:width"]: ACTIONS["DistanceOrQuantitativeValue"]
+}
+CODEMETA_STRATEGY[iri["schema:ShippingDeliveryTime"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:handlingTime"]: ACTIONS["QuantitativeValueOrServicePeriod"],
+ iri["schema:transitTime"]: ACTIONS["QuantitativeValueOrServicePeriod"]
+}
+CODEMETA_STRATEGY[iri["schema:ShippingRateSettings"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:shippingRate"]: ACTIONS["MonetaryAmountOrShippingRateSettings"]
+}
+CODEMETA_STRATEGY[iri["schema:ShippingService"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:handlingTime"]: ACTIONS["QuantitativeValueOrServicePeriod"]
+}
+CODEMETA_STRATEGY[iri["schema:TypeAndQuantityNode"]] = {
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]],
+ iri["schema:typeOfGood"]: ACTIONS["ProductOrService"]
+}
+CODEMETA_STRATEGY[iri["schema:WarrantyPromise"]] = {**CODEMETA_STRATEGY[iri["schema:StructuredValue"]]}
+CODEMETA_STRATEGY[iri["schema:VirtualLocation"]] = {**CODEMETA_STRATEGY[iri["schema:Intangible"]]}
+
+
+CODEMETA_STRATEGY[iri["schema:MedicalEntity"]] = {**CODEMETA_STRATEGY[iri["schema:Thing"]]}
+
+CODEMETA_STRATEGY[iri["schema:AnatomicalStructure"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:AnatomicalSystem"]] = {
+ **CODEMETA_STRATEGY[iri["schema:MedicalEntity"]],
+ iri["schema:comprisedOf"]: ACTIONS["AnatomicalStructureOrAnatomicalSystem"]
+}
+CODEMETA_STRATEGY[iri["schema:DrugClass"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:LifestyleModification"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalCause"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalCondition"]] = {
+ **CODEMETA_STRATEGY[iri["schema:MedicalEntity"]],
+ iri["schema:associatedAnatomy"]: ACTIONS["AnatomicalStructureOrAnatomicalSystemOrSuperficialAnatomy"],
+ iri["schema:possibleTreatment"]: ACTIONS["DrugOrDrugClassOrLifestyleModificationOrMedicalTherapy"],
+ iri["schema:secondaryPrevention"]: ACTIONS["DrugOrDrugClassOrLifestyleModificationOrMedicalTherapy"]
+}
+CODEMETA_STRATEGY[iri["schema:MedicalSignOrSymptom"]] = {
+ **CODEMETA_STRATEGY[iri["schema:MedicalCondition"]],
+ iri["schema:possibleTreatment"]: ACTIONS["DrugOrDrugClassOrLifestyleModificationOrMedicalTherapy"]
+}
+CODEMETA_STRATEGY[iri["schema:MedicalSign"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalSignOrSymptom"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalContraindication"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalDevice"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalGuideline"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalIntangible"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:DDxElement"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalIntangible"]]}
+CODEMETA_STRATEGY[iri["schema:DrugLegalStatus"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalIntangible"]]}
+CODEMETA_STRATEGY[iri["schema:DoseSchedule"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalIntangible"]]}
+CODEMETA_STRATEGY[iri["schema:DrugStrength"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalIntangible"]]}
+CODEMETA_STRATEGY[iri["schema:MaximumDoseSchedule"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalIntangible"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalConditionStage"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalIntangible"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalProcedure"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:TherapeuticProcedure"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalProcedure"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalTherapy"]] = {**CODEMETA_STRATEGY[iri["schema:TherapeuticProcedure"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalRiskFactor"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:MedicalStudy"]] = {
+ **CODEMETA_STRATEGY[iri["schema:MedicalEntity"]],
+ iri["schema:sponsor"]: ACTIONS["OrganizationOrPerson"]
+}
+CODEMETA_STRATEGY[iri["schema:MedicalTest"]] = {**CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]}
+CODEMETA_STRATEGY[iri["schema:SuperficialAnatomy"]] = {
+ **CODEMETA_STRATEGY[iri["schema:MedicalEntity"]],
+ iri["schema:relatedAnatomy"]: ACTIONS["AnatomicalStructureOrAnatomicalSystem"]
+}
+
+
+CODEMETA_STRATEGY[iri["schema:Organization"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Thing"]],
+ iri["schema:acceptedPaymentMethod"]: ACTIONS["LoanOrCreditOrPaymentMethod"],
+ iri["schema:alumni"]: ACTIONS["Person"],
+ iri["schema:areaServed"]: ACTIONS["AdministrativeAreaOrGeoShapeOrPlace"],
+ iri["schema:brand"]: ACTIONS["BrandOrOrganization"],
+ iri["schema:employee"]: ACTIONS["Person"],
+ iri["schema:founder"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:funder"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:legalRepresentative"]: ACTIONS["Person"],
+ iri["schema:location"]: ACTIONS["PlaceOrPostalAddressOrVirtualLocation"],
+ iri["schema:member"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:memberOf"]: ACTIONS["MemberProgramTierOrOrganizationOrProgramMembership"],
+ iri["schema:ownershipFundingInfo"]: ACTIONS["AboutPageOrCreativeWork"],
+ iri["schema:sponsor"]: ACTIONS["OrganizationOrPerson"]
+}
+
+CODEMETA_STRATEGY[iri["schema:PerformingGroup"]] = {**CODEMETA_STRATEGY[iri["schema:Organization"]]}
+CODEMETA_STRATEGY[iri["schema:MusicGroup"]] = {
+ **CODEMETA_STRATEGY[iri["schema:PerformingGroup"]],
+ iri["schema:musicGroupMember"]: ACTIONS["Person"],
+ iri["schema:track"]: ACTIONS["ItemListOrMusicRecording"]
+}
+
+
+CODEMETA_STRATEGY[iri["schema:Person"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Thing"]],
+ iri["schema:alumniOf"]: ACTIONS["EducationalOrganizationOrOrganization"],
+ iri["schema:brand"]: ACTIONS["BrandOrOrganization"],
+ iri["schema:children"]: ACTIONS["Person"],
+ iri["schema:colleague"]: ACTIONS["Person"],
+ iri["schema:follows"]: ACTIONS["Person"],
+ iri["schema:funder"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:height"]: ACTIONS["DistanceOrQuantitativeValue"],
+ iri["schema:homeLocation"]: ACTIONS["ContactPointOrPlace"],
+ iri["schema:knows"]: ACTIONS["Person"],
+ iri["schema:memberOf"]: ACTIONS["MemberProgramTierOrOrganizationOrProgramMembership"],
+ iri["schema:netWorth"]: ACTIONS["MonetaryAmountOrPriceSpecification"],
+ iri["schema:parent"]: ACTIONS["Person"],
+ iri["schema:pronouns"]: ACTIONS["DefinedTermOrStructuredValue"],
+ iri["schema:relatedTo"]: ACTIONS["Person"],
+ iri["schema:sibling"]: ACTIONS["Person"],
+ iri["schema:sponsor"]: ACTIONS["OrganizationOrPerson"],
+ iri["schema:spouse"]: ACTIONS["Person"],
+ iri["schema:weight"]: ACTIONS["MassOrQuantitativeValue"],
+ iri["schema:workLocation"]: ACTIONS["ContactPointOrPlace"]
+}
+
+
+CODEMETA_STRATEGY[iri["schema:Place"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Thing"]],
+ iri["schema:geo"]: ACTIONS["GeoCoordinatesOrGeoShape"],
+ iri["schema:geoContains"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoCoveredBy"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoCovers"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoCrosses"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoDisjoint"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoEquals"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoIntersects"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoOverlaps"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoTouches"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:geoWithin"]: ACTIONS["GeospatialGeometryOrPlace"],
+ iri["schema:photo"]: ACTIONS["ImageObjectOrPhotograph"]
+}
+
+CODEMETA_STRATEGY[iri["schema:AdministrativeArea"]] = {**CODEMETA_STRATEGY[iri["schema:Place"]]}
+CODEMETA_STRATEGY[iri["schema:Country"]] = {**CODEMETA_STRATEGY[iri["schema:AdministrativeArea"]]}
+CODEMETA_STRATEGY[iri["schema:CivicStructure"]] = {**CODEMETA_STRATEGY[iri["schema:Place"]]}
+
+
+CODEMETA_STRATEGY[iri["schema:Product"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Thing"]],
+ iri["schema:brand"]: ACTIONS["BrandOrOrganization"],
+ iri["schema:category"]: ACTIONS["CategoryCodeOrThing"],
+ iri["schema:depth"]: ACTIONS["DistanceOrQuantitativeValue"],
+ iri["schema:height"]: ACTIONS["DistanceOrQuantitativeValue"],
+ iri["schema:isRelatedTo"]: ACTIONS["ProductOrService"],
+ iri["schema:isSimilarTo"]: ACTIONS["ProductOrService"],
+ iri["schema:isVariantOf"]: ACTIONS["ProductGroupOrProductModel"],
+ iri["schema:negativeNotes"]: ACTIONS["ItemListOrListItemOrWebContent"],
+ iri["schema:offers"]: ACTIONS["DemandOrOffer"],
+ iri["schema:positiveNotes"]: ACTIONS["ItemListOrListItemOrWebContent"],
+ iri["schema:size"]: ACTIONS["DefinedTermOrQuantitativeValueOrSizeSpecification"],
+ iri["schema:weight"]: ACTIONS["MassOrQuantitativeValue"],
+ iri["schema:width"]: ACTIONS["DistanceOrQuantitativeValue"]
+}
+
+CODEMETA_STRATEGY[iri["schema:ProductGroup"]] = {**CODEMETA_STRATEGY[iri["schema:Product"]]}
+CODEMETA_STRATEGY[iri["schema:ProductModel"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Product"]],
+ iri["schema:isVariantOf"]: ACTIONS["ProductGroupOrProductModel"]
+}
+
+
+CODEMETA_STRATEGY[iri["schema:Taxon"]] = {**CODEMETA_STRATEGY[iri["schema:Thing"]]}
+
+
+CODEMETA_STRATEGY[iri["schema:CreativeWorkSeries"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ **CODEMETA_STRATEGY[iri["schema:Series"]]
+}
+
+CODEMETA_STRATEGY[iri["schema:DefinedRegion"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Place"]],
+ **CODEMETA_STRATEGY[iri["schema:StructuredValue"]]
+}
+
+CODEMETA_STRATEGY[iri["schema:Drug"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Product"]],
+ **CODEMETA_STRATEGY[iri["schema:MedicalEntity"]]
+}
+
+CODEMETA_STRATEGY[iri["schema:EducationalOrganization"]] = {
+ **CODEMETA_STRATEGY[iri["schema:Organization"]],
+ **CODEMETA_STRATEGY[iri["schema:CivicStructure"]]
+}
+
+CODEMETA_STRATEGY[iri["schema:HowToSection"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ **CODEMETA_STRATEGY[iri["schema:ItemList"]],
+ **CODEMETA_STRATEGY[iri["schema:ListItem"]]
+}
+
+CODEMETA_STRATEGY[iri["schema:HowToStep"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CreativeWork"]],
+ **CODEMETA_STRATEGY[iri["schema:ItemList"]],
+ **CODEMETA_STRATEGY[iri["schema:ListItem"]]
+}
+
+CODEMETA_STRATEGY[iri["schema:MedicalCode"]] = {
+ **CODEMETA_STRATEGY[iri["schema:CategoryCode"]],
+ **CODEMETA_STRATEGY[iri["schema:MedicalIntangible"]]
+}
+
+CODEMETA_STRATEGY[iri["schema:PaymentCard"]] = {
+ **CODEMETA_STRATEGY[iri["schema:FinancialProduct"]],
+ **CODEMETA_STRATEGY[iri["schema:PaymentMethod"]]
+}
+CODEMETA_STRATEGY[iri["schema:CreditCard"]] = {
+ **CODEMETA_STRATEGY[iri["schema:LoanOrCredit"]],
+ **CODEMETA_STRATEGY[iri["schema:PaymentCard"]]
+}
diff --git a/src/hermes/model/types/__init__.py b/src/hermes/model/types/__init__.py
index df6aed71..3b2089b9 100644
--- a/src/hermes/model/types/__init__.py
+++ b/src/hermes/model/types/__init__.py
@@ -20,7 +20,6 @@
(lambda c: isinstance(c, list), {"ld_container": ld_list}),
# pythonize items from lists (expanded set is already handled above)
- (ld_container.is_json_id, {"python": lambda c, **_: c["@id"]}),
(ld_container.is_typed_json_value, {"python": lambda c, **kw: ld_container.typed_ld_to_py([c], **kw)}),
(ld_container.is_json_value, {"python": lambda c, **_: c["@value"]}),
(ld_list.is_container, {"ld_container": lambda c, **kw: ld_list([c], **kw)}),
diff --git a/src/hermes/model/types/ld_container.py b/src/hermes/model/types/ld_container.py
index c8ab051f..49343954 100644
--- a/src/hermes/model/types/ld_container.py
+++ b/src/hermes/model/types/ld_container.py
@@ -105,17 +105,9 @@ def __init__(
self.context = context or []
- # Create active context (to use with pyld) depending on the initial variables
- # Re-use active context from parent if available
- if self.parent:
- if self.context:
- self.active_ctx = self.ld_proc.process_context(
- self.parent.active_ctx, self.context, {"documentLoader": bundled_loader}
- )
- else:
- self.active_ctx = parent.active_ctx
- else:
- self.active_ctx = self.ld_proc.initial_ctx(self.full_context, {"documentLoader": bundled_loader})
+ # Create active context (to use with pyld) depending on the initial variables.
+ # Don't re-use active context from parent (created some weird in the process step when context is often added).
+ self.active_ctx = self.ld_proc.initial_ctx(self.full_context, {"documentLoader": bundled_loader})
def add_context(self: Self, context: list[Union[str | JSON_LD_CONTEXT_DICT]]) -> None:
"""
@@ -185,8 +177,10 @@ def ld_value(self: Self) -> EXPANDED_JSON_LD_VALUE:
return self._data
def _to_python(
- self: Self, full_iri: str, ld_value: Union[list, dict, str]
- ) -> Union[ld_dict, ld_list, BASIC_TYPE, TIME_TYPE]:
+ self: Self,
+ full_iri: str,
+ ld_value: Union[EXPANDED_JSON_LD_VALUE, dict[str, EXPANDED_JSON_LD_VALUE], list[str], str]
+ ) -> Union["ld_container", BASIC_TYPE, TIME_TYPE]:
"""
Returns a pythonized version of the given value pretending the value is in self and full_iri its key.
@@ -196,7 +190,7 @@ def _to_python(
:type full_iri: str
:param ld_value: The value thats pythonized value is requested. ld_value has to be valid expanded JSON-LD if it
was embeded in self._data.
- :type ld_value: list | dict | str
+ :type ld_value: EXPANDED_JSON_LD_VALUE | dict[str, EXPANDED_JSON_LD_VALUE] | list[str] | str
:return: The pythonized value of the ld_value.
:rtype: ld_dict | ld_list | BASIC_TYPE | TIME_TYPE
@@ -246,7 +240,7 @@ def _to_expanded_json(
# while searching build a path such that it leads from the found ld_dicts ld_value to selfs data_dict/ item_list
parent = self
path = []
- while parent.__class__.__name__ not in {"ld_dict", "SoftwareMetadata"}:
+ while "ld_dict" not in [sub_cls.__name__ for sub_cls in type(parent).mro()]:
if parent.container_type == "@list":
path.extend(["@list", 0])
elif parent.container_type == "@graph":
@@ -259,7 +253,7 @@ def _to_expanded_json(
# if neither self nor any of its parents is a ld_dict:
# create a dict with the key of the outer most parent of self and this parents ld_value as a value
# this dict is stored in an ld_container and simulates the most minimal JSON-LD object possible
- if parent.__class__.__name__ not in {"ld_dict", "SoftwareMetadata"}:
+ if "ld_dict" not in [sub_cls.__name__ for sub_cls in type(parent).mro()]:
key = self.ld_proc.expand_iri(parent.active_ctx, parent.key)
parent = ld_container([{key: parent._data}])
path.append(0)
@@ -286,7 +280,7 @@ def _to_expanded_json(
[(new_key, temp) for new_key in temp.keys() if isinstance(temp[new_key], special_types)]
)
elif isinstance(temp, ld_container):
- if temp.__class__.__name__ == "ld_list" and temp.container_type == "@set":
+ if "ld_list" in [sub_cls.__name__ for sub_cls in type(temp).mro()] and temp.container_type == "@set":
ref[key] = temp._data
else:
ref[key] = temp._data[0]
diff --git a/src/hermes/model/types/ld_dict.py b/src/hermes/model/types/ld_dict.py
index a6b9de09..6f38041b 100644
--- a/src/hermes/model/types/ld_dict.py
+++ b/src/hermes/model/types/ld_dict.py
@@ -90,10 +90,7 @@ def __getitem__(self: Self, key: str) -> ld_list:
:rtype: ld_list
"""
full_iri = self.ld_proc.expand_iri(self.active_ctx, key)
- if full_iri not in self.data_dict:
- self[full_iri] = []
- ld_value = self.data_dict[full_iri]
- return self._to_python(full_iri, ld_value)
+ return self._to_python(full_iri, self.data_dict[full_iri])
def __setitem__(self: Self, key: str, value: Union[JSON_LD_VALUE, BASIC_TYPE, TIME_TYPE, ld_dict, ld_list]) -> None:
"""
@@ -232,6 +229,18 @@ def __ne__(
return NotImplemented
return not x
+ def __bool__(self):
+ return bool(self.data_dict)
+
+ def setdefault(self, key, default):
+ if key not in self:
+ self[key] = default
+ return self[key]
+
+ def emplace(self, key):
+ if key not in self:
+ self[key] = []
+
def get(
self: Self, key: str, default: Any = _NO_DEFAULT
) -> Union[ld_list, Any]:
@@ -381,7 +390,7 @@ def from_dict(
# expand value and create an ld_dict from it
ld_value = cls.ld_proc.expand(ld_data, {"expandContext": full_context, "documentLoader": bundled_loader})
- ld_value = cls(ld_value, parent=parent, key=key, context=merged_contexts)
+ ld_value = ld_dict(ld_value, parent=parent, key=key, context=merged_contexts)
return ld_value
@@ -419,7 +428,4 @@ def is_json_dict(cls: type[Self], ld_value: Any) -> bool:
if any(k in ld_value for k in ["@set", "@graph", "@list", "@value"]):
return False
- if ['@id'] == [*ld_value.keys()]:
- return False
-
return True
diff --git a/src/hermes/model/types/ld_list.py b/src/hermes/model/types/ld_list.py
index 23ebe4d0..5dd0669f 100644
--- a/src/hermes/model/types/ld_list.py
+++ b/src/hermes/model/types/ld_list.py
@@ -29,7 +29,7 @@
class ld_list(ld_container):
"""
An JSON-LD container resembling a list ("@set", "@list" or "@graph").
- See also :class:`ld_container`
+ See also :class:`ld_container`.
:ivar container_type: The type of JSON-LD container the list is representing. ("@set", "@list", "graph")
:ivartype container_type: str
diff --git a/test/hermes_test/model/test_api.py b/test/hermes_test/model/test_api.py
index b4bec276..906203b5 100644
--- a/test/hermes_test/model/test_api.py
+++ b/test/hermes_test/model/test_api.py
@@ -1,3 +1,10 @@
+# SPDX-FileCopyrightText: 2026 German Aerospace Center (DLR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: Michael Fritzsche
+# SPDX-FileContributor: Stephan Druskat
+
import pytest
from hermes.model import SoftwareMetadata
@@ -53,23 +60,25 @@ def test_init_nested_object():
def test_append():
data = SoftwareMetadata()
- data["schema:foo"].append("a")
- assert type(data["schema:foo"]) is ld_list
- assert data["schema:foo"][0] == "a" and data["schema:foo"].item_list == [{"@value": "a"}]
- data["schema:foo"].append("b")
- assert type(data["schema:foo"]) is ld_list
- assert data["schema:foo"] == [{"@value": "a"}, {"@value": "b"}]
- data["schema:foo"].append("c")
- assert data["schema:foo"] == [{"@value": "a"}, {"@value": "b"}, {"@value": "c"}]
+ data.emplace("schema:name")
+ data["schema:name"].append("a")
+ assert type(data["schema:name"]) is ld_list
+ assert data["schema:name"][0] == "a" and data["schema:name"] == ["a"]
+ data["schema:name"].append("b")
+ assert type(data["schema:name"]) is ld_list and data["schema:name"] == ["a", "b"]
+ data.emplace("schema:name")
+ data["schema:name"].append("c")
+ assert data["schema:name"] == ["a", "b", "c"]
+
data = SoftwareMetadata()
- data["schema:foo"].append({"schema:name": "bar"})
- assert type(data["schema:foo"]) is ld_list and type(data["schema:foo"][0]) is ld_dict
- assert data["schema:foo"] == [{"http://schema.org/name": [{"@value": "bar"}]}]
- data["schema:foo"].append({"schema:name": "bar"})
- assert type(data["schema:foo"]) is ld_list
- assert data["schema:foo"] == 2 * [{"http://schema.org/name": [{"@value": "bar"}]}]
- data["schema:foo"].append({"schema:name": "bar"})
- assert data["schema:foo"] == 3 * [{"http://schema.org/name": [{"@value": "bar"}]}]
+ data.setdefault("schema:Person", []).append({"schema:name": "foo"})
+ assert type(data["schema:Person"]) is ld_list and type(data["schema:Person"][0]) is ld_dict
+ assert data["schema:Person"][0] == {"http://schema.org/name": ["foo"]}
+ data["schema:Person"].append({"schema:name": "foo"})
+ assert type(data["schema:Person"]) is ld_list
+ assert data["schema:Person"] == 2 * [{"http://schema.org/name": ["foo"]}]
+ data["schema:Person"].append({"schema:name": "foo"})
+ assert data["schema:Person"] == 3 * [{"http://schema.org/name": ["foo"]}]
def test_iterative_assignment():
@@ -95,7 +104,7 @@ def test_usage():
data["author"][0]["email"].append("foo@baz.com")
assert len(data["author"]) == 2
assert len(data["author"][0]["email"]) == 2
- assert len(data["author"][1]["email"]) == 0
+ assert len(data["author"][1].get("email", [])) == 0
harvest = {
"authors": [
{"name": "Foo", "affiliation": ["Uni A", "Lab B"], "kw": ["a", "b", "c"]},
@@ -104,15 +113,19 @@ def test_usage():
]
}
for author in harvest["authors"]:
- for exist_author in data["author"]:
- if author["name"] in exist_author["name"]:
- exist_author["affiliation"] = author.get("affiliation", [])
- exist_author["email"].extend(email if isinstance((email := author.get("email", [])), list) else [email])
- exist_author["schema:knowsAbout"].extend(kw if isinstance((kw := author.get("kw", [])), list) else [kw])
+ for exist_author in data.get("author", []):
+ if author["name"] in exist_author.get("name", []):
+ exist_author["affiliation"] = author["affiliation"]
+ if "email" in author:
+ exist_author.emplace("email")
+ exist_author["email"].append(author["email"])
+ if "kw" in author:
+ exist_author.emplace("schema:knowsAbout")
+ exist_author["schema:knowsAbout"].extend(author["kw"])
break
else:
- data["author"].append(author)
- assert len(data["author"]) == 3
+ data.setdefault("author", []).append(author)
+ assert len(data.get("author", [])) == 3
foo, bar, baz = data["author"]
assert foo["name"][0] == "Foo"
assert foo["affiliation"] == ["Uni A", "Lab B"]
@@ -123,14 +136,13 @@ def test_usage():
assert bar["email"] == ["bar@c.edu"]
assert baz["name"][0] == "Baz"
assert baz["affiliation"] == ["Lab E"]
- assert len(baz["schema:knowsAbout"]) == 0
- assert len(baz["email"]) == 0
+ assert len(baz.get("schema:knowsAbout", [])) == 0
+ assert len(baz.get("email", [])) == 0
for author in data["author"]:
assert "name" in author
- assert "email" in author
- if author["schema:knowsAbout"] == ["egg"]:
+ if "Baz" not in author["name"]:
+ assert "email" in author
+ if "schema:knowsAbout" not in author:
# FIXME: None has to be discussed
- # json-ld processor just removes it in expansion
author["schema:knowsAbout"] = None
author["schema:pronouns"] = "they/them"
- assert len(bar["schema:knowsAbout"]) == 0
diff --git a/test/hermes_test/model/test_api_e2e.py b/test/hermes_test/model/test_api_e2e.py
index 43ce0be8..b85a07a6 100644
--- a/test/hermes_test/model/test_api_e2e.py
+++ b/test/hermes_test/model/test_api_e2e.py
@@ -4,11 +4,22 @@
# SPDX-FileContributor: Michael Fritzsche
+from datetime import date
+import json
import pytest
import sys
-from hermes.model import SoftwareMetadata
-from hermes.model.context_manager import HermesContext
+from hermes.model import context_manager, SoftwareMetadata
from hermes.commands import cli
+from pathlib import Path
+
+
+@pytest.fixture
+def sandbox_auth():
+ path = Path("./../auth.txt")
+ if not path.exists():
+ pytest.skip("Local auth token file does not exist.")
+ with path.open() as f:
+ yield f.read()
@pytest.mark.parametrize(
@@ -162,7 +173,7 @@
"http://schema.org/license": [{"@id": "https://spdx.org/licenses/Apache-2.0"}],
"http://schema.org/name": [{"@value": "Test"}],
"http://schema.org/url": [
- {"@id": 'https://arxiv.org/abs/2201.09015'},
+ {"@id": "https://arxiv.org/abs/2201.09015"},
{"@id": "https://docs.software-metadata.pub/en/latest"}
],
"http://schema.org/version": [{"@value": "9.0.1"}]
@@ -182,21 +193,19 @@ def test_cff_harvest(tmp_path, monkeypatch, cff, res):
sys.argv = ["hermes", "harvest", "--path", str(tmp_path), "--config", str(config_file)]
result = {}
try:
- monkeypatch.setattr(HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
+ monkeypatch.setattr(context_manager.HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
cli.main()
- except SystemExit:
- print("TODO: Delete when package is working again or mock cli")
+ except SystemExit as e:
+ if e.code != 0:
+ raise e
finally:
- manager = HermesContext()
+ manager = context_manager.HermesContext()
manager.prepare_step("harvest")
- with manager["cff"] as cache:
- result = SoftwareMetadata(cache["codemeta"])
+ result = SoftwareMetadata.load_from_cache(manager, "cff")
manager.finalize_step("harvest")
sys.argv = orig_argv
- # FIXME: update to compare the SoftwareMetadata objects instead of the data_dicts (in multiple places)
- # after merge with refactor/data-model and/or refactor/423-implement-public-api
- assert result.data_dict == res.data_dict
+ assert result == res
@pytest.mark.parametrize(
@@ -343,19 +352,19 @@ def test_codemeta_harvest(tmp_path, monkeypatch, codemeta, res):
sys.argv = ["hermes", "harvest", "--path", str(tmp_path), "--config", str(config_file)]
result = {}
try:
- monkeypatch.setattr(HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
+ monkeypatch.setattr(context_manager.HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
cli.main()
- except SystemExit:
- print("TODO: Delete when package is working again or mock cli")
+ except SystemExit as e:
+ if e.code != 0:
+ raise e
finally:
- manager = HermesContext()
+ manager = context_manager.HermesContext()
manager.prepare_step("harvest")
- with manager["codemeta"] as cache:
- result = SoftwareMetadata(cache["codemeta"])
+ result = SoftwareMetadata.load_from_cache(manager, "codemeta")
manager.finalize_step("harvest")
sys.argv = orig_argv
- assert result.data_dict == res.data_dict
+ assert result == res
@pytest.mark.parametrize(
@@ -431,7 +440,7 @@ def test_codemeta_harvest(tmp_path, monkeypatch, codemeta, res):
def test_do_nothing_curate(tmp_path, monkeypatch, process_result, res):
monkeypatch.chdir(tmp_path)
- manager = HermesContext(tmp_path)
+ manager = context_manager.HermesContext(tmp_path)
manager.prepare_step("process")
with manager["result"] as cache:
cache["expanded"] = process_result.ld_value
@@ -445,14 +454,301 @@ def test_do_nothing_curate(tmp_path, monkeypatch, process_result, res):
sys.argv = ["hermes", "curate", "--path", str(tmp_path), "--config", str(config_file)]
result = {}
try:
- monkeypatch.setattr(HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
+ monkeypatch.setattr(context_manager.HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
cli.main()
- except SystemExit:
+ except SystemExit as e:
+ if e.code != 0:
+ raise e
+ finally:
manager.prepare_step("curate")
- with manager["result"] as cache:
- result = SoftwareMetadata(cache["expanded"][0], cache["context"]["@context"][1])
+ result = SoftwareMetadata.load_from_cache(manager, "result")
manager.finalize_step("curate")
- finally:
sys.argv = orig_argv
assert result.data_dict == res.data_dict
+
+
+@pytest.mark.parametrize(
+ "metadata",
+ [
+ SoftwareMetadata({
+ "@type": ["http://schema.org/SoftwareSourceCode"],
+ "http://schema.org/description": [{"@value": "for testing"}],
+ "http://schema.org/name": [{"@value": "Test"}]
+ }),
+ ]
+)
+def test_file_deposit(tmp_path, monkeypatch, metadata):
+ monkeypatch.chdir(tmp_path)
+
+ manager = context_manager.HermesContext(tmp_path)
+ manager.prepare_step("curate")
+ with manager["result"] as cache:
+ cache["codemeta"] = metadata.compact()
+ manager.finalize_step("curate")
+
+ config_file = tmp_path / "hermes.toml"
+ config_file.write_text("[deposit]\ntarget = \"file\"")
+
+ orig_argv = sys.argv[:]
+ sys.argv = ["hermes", "deposit", "--path", str(tmp_path), "--config", str(config_file)]
+ result = {}
+ try:
+ monkeypatch.setattr(context_manager.HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
+ cli.main()
+ except SystemExit as e:
+ if e.code != 0:
+ raise e
+ finally:
+ with open("codemeta.json", "r") as cache:
+ result = SoftwareMetadata(json.load(cache))
+ sys.argv = orig_argv
+
+ assert result == metadata
+
+
+@pytest.mark.parametrize(
+ "metadata, invenio_metadata",
+ [
+ (
+ SoftwareMetadata({
+ "@type": ["http://schema.org/SoftwareSourceCode"],
+ "http://schema.org/description": [{"@value": "for testing"}],
+ "http://schema.org/name": [{"@value": "Test"}],
+ "http://schema.org/author": [{
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Test"}],
+ "http://schema.org/givenName": [{"@value": "Testi"}]
+ }],
+ "http://schema.org/license": [{"@id": "https://spdx.org/licenses/Apache-2.0"}]
+ }),
+ {
+ "upload_type": "software",
+ "publication_date": date.today().isoformat(),
+ "title": "Test",
+ "creators": [{"name": "Test, Testi"}],
+ "description": "for testing",
+ "access_right": "closed",
+ "license": "apache-2.0",
+ "prereserve_doi": True,
+ "related_identifiers": [
+ {"identifier": "10.5281/zenodo.13311079", "relation": "isCompiledBy", "scheme": "doi"}
+ ]
+ }
+ )
+ ]
+)
+def test_invenio_deposit(tmp_path, monkeypatch, sandbox_auth, metadata, invenio_metadata):
+ monkeypatch.chdir(tmp_path)
+
+ manager = context_manager.HermesContext(tmp_path)
+ manager.prepare_step("curate")
+ with manager["result"] as cache:
+ cache["codemeta"] = metadata.compact()
+ manager.finalize_step("curate")
+
+ (tmp_path / "test.txt").write_text("Test, oh wonderful test!\n")
+
+ config_file = tmp_path / "hermes.toml"
+ config_file.write_text(f"""[deposit]
+target = "invenio"
+[deposit.invenio]
+site_url = "https://sandbox.zenodo.org"
+access_right = "closed"
+auth_token = "{sandbox_auth}"
+files = ["test.txt"]
+[deposit.invenio.api_paths]
+licenses = "api/vocabularies/licenses"
+""")
+
+ orig_argv = sys.argv[:]
+ sys.argv = ["hermes", "deposit", "--path", str(tmp_path), "--config", str(config_file), "--initial"]
+ result = {}
+ try:
+ monkeypatch.setattr(context_manager.HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
+ cli.main()
+ except SystemExit as e:
+ if e.code != 0:
+ raise e
+ finally:
+ manager.prepare_step("deposit")
+ with manager["invenio"] as cache:
+ result = cache["deposit"]
+ manager.finalize_step("deposit")
+ sys.argv = orig_argv
+
+ # TODO: compare to actually expected value
+ assert result == invenio_metadata
+
+
+@pytest.mark.parametrize(
+ "metadata_in, metadata_out",
+ [
+ (
+ {
+ "cff": SoftwareMetadata({
+ "@type": ["http://schema.org/SoftwareSourceCode"],
+ "http://schema.org/description": [{"@value": "for testing"}],
+ "http://schema.org/name": [{"@value": "Test"}],
+ "http://schema.org/author": [{
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Test"}],
+ "http://schema.org/givenName": [{"@value": "Testi"}]
+ }],
+ "http://schema.org/license": [{"@id": "https://spdx.org/licenses/Apache-2.0"}]
+ })
+ },
+ SoftwareMetadata({
+ "@type": ["http://schema.org/SoftwareSourceCode"],
+ "http://schema.org/description": [{"@value": "for testing"}],
+ "http://schema.org/name": [{"@value": "Test"}],
+ "http://schema.org/author": [{
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Test"}],
+ "http://schema.org/givenName": [{"@value": "Testi"}]
+ }],
+ "http://schema.org/license": [{"@id": "https://spdx.org/licenses/Apache-2.0"}]
+ })
+ )
+ ]
+)
+def test_process(tmp_path, monkeypatch, metadata_in, metadata_out):
+ monkeypatch.chdir(tmp_path)
+
+ manager = context_manager.HermesContext(tmp_path)
+ manager.prepare_step("harvest")
+ for harvester, result in metadata_in.items():
+ with manager[harvester] as cache:
+ cache["codemeta"] = result.compact()
+ cache["context"] = {"@context": result.full_context}
+ cache["expanded"] = result.ld_value
+ manager.finalize_step("harvest")
+
+ config_file = tmp_path / "hermes.toml"
+ config_file.write_text(f"[harvest]\nsources = [{', '.join(f'\"{harvester}\"' for harvester in metadata_in)}]")
+
+ orig_argv = sys.argv[:]
+ sys.argv = ["hermes", "process", "--path", str(tmp_path), "--config", str(config_file)]
+ result = {}
+ try:
+ monkeypatch.setattr(context_manager.HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
+ cli.main()
+ except SystemExit as e:
+ if e.code != 0:
+ raise e
+ finally:
+ manager.prepare_step("process")
+ result = SoftwareMetadata.load_from_cache(manager, "result")
+ manager.finalize_step("process")
+ sys.argv = orig_argv
+
+ assert result == metadata_out
+
+
+@pytest.mark.parametrize(
+ "metadata_in, metadata_out",
+ [
+ (
+ {
+ "cff": SoftwareMetadata({
+ "@type": ["http://schema.org/SoftwareSourceCode"],
+ "http://schema.org/name": [{"@value": "Test"}],
+ "http://schema.org/author": [
+ {
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Test"}],
+ "http://schema.org/email": [{"@value": "test.testi@testis.tests"}]
+ },
+ {
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Testers"}]
+ },
+ {
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Tester"}],
+ "http://schema.org/email": [{"@value": "test@tester.tests"}]
+ }
+ ],
+ "http://schema.org/license": [{"@id": "https://spdx.org/licenses/Apache-2.0"}]
+ }),
+ "codemeta": SoftwareMetadata({
+ "@type": ["http://schema.org/SoftwareSourceCode"],
+ "http://schema.org/description": [{"@value": "for testing"}],
+ "http://schema.org/name": [{"@value": "Test"}, {"@value": "Testis Test"}],
+ "http://schema.org/author": [
+ {
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Test"}],
+ "http://schema.org/givenName": [{"@value": "Testi"}],
+ "http://schema.org/email": [
+ {"@value": "test.testi@testis.tests"},
+ {"@value": "test.testi@testis.tests2"}
+ ]
+ },
+ {
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Testers"}]
+ }
+ ]
+ })
+ },
+ SoftwareMetadata({
+ "@type": ["http://schema.org/SoftwareSourceCode"],
+ "http://schema.org/description": [{"@value": "for testing"}],
+ "http://schema.org/name": [{"@value": "Test"}, {"@value": "Testis Test"}],
+ "http://schema.org/author": [
+ {
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Test"}],
+ "http://schema.org/givenName": [{"@value": "Testi"}],
+ "http://schema.org/email": [
+ {"@value": "test.testi@testis.tests"},
+ {"@value": "test.testi@testis.tests2"}
+ ]
+ },
+ {
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Testers"}]
+ },
+ {
+ "@type": "http://schema.org/Person",
+ "http://schema.org/familyName": [{"@value": "Tester"}],
+ "http://schema.org/email": [{"@value": "test@tester.tests"}]
+ }
+ ],
+ "http://schema.org/license": [{"@id": "https://spdx.org/licenses/Apache-2.0"}]
+ })
+ )
+ ]
+)
+def test_process_complex(tmp_path, monkeypatch, metadata_in, metadata_out):
+ monkeypatch.chdir(tmp_path)
+
+ manager = context_manager.HermesContext(tmp_path)
+ manager.prepare_step("harvest")
+ for harvester, result in metadata_in.items():
+ with manager[harvester] as cache:
+ cache["codemeta"] = result.compact()
+ cache["context"] = {"@context": result.full_context}
+ cache["expanded"] = result.ld_value
+ manager.finalize_step("harvest")
+
+ config_file = tmp_path / "hermes.toml"
+ config_file.write_text(f"[harvest]\nsources = [{', '.join(f'\"{harvester}\"' for harvester in metadata_in)}]")
+
+ orig_argv = sys.argv[:]
+ sys.argv = ["hermes", "process", "--path", str(tmp_path), "--config", str(config_file)]
+ result = {}
+ try:
+ monkeypatch.setattr(context_manager.HermesContext.__init__, "__defaults__", (tmp_path.cwd(),))
+ cli.main()
+ except SystemExit as e:
+ if e.code != 0:
+ raise e
+ finally:
+ manager.prepare_step("process")
+ result = SoftwareMetadata.load_from_cache(manager, "result")
+ manager.finalize_step("process")
+ sys.argv = orig_argv
+
+ assert result == metadata_out
diff --git a/test/hermes_test/model/types/test_ld_container.py b/test/hermes_test/model/types/test_ld_container.py
index f73fdcd9..f0844ecd 100644
--- a/test/hermes_test/model/types/test_ld_container.py
+++ b/test/hermes_test/model/types/test_ld_container.py
@@ -107,7 +107,7 @@ def test_to_python_id_value(self, mock_context):
assert cont._to_python("http://spam.eggs/ham",
[{"@id": "http://spam.eggs/spam"}]) == [{"@id": "http://spam.eggs/spam"}]
assert cont._to_python("http://spam.eggs/ham",
- {"@id": "http://spam.eggs/identifier"}) == "http://spam.eggs/identifier"
+ {"@id": "http://spam.eggs/identifier"}) == {"@id": "http://spam.eggs/identifier"}
def test_to_python_basic_value(self, mock_context):
cont = ld_container([{}], context=[mock_context])
diff --git a/test/hermes_test/model/types/test_ld_dict.py b/test/hermes_test/model/types/test_ld_dict.py
index 66095295..66ce44cb 100644
--- a/test/hermes_test/model/types/test_ld_dict.py
+++ b/test/hermes_test/model/types/test_ld_dict.py
@@ -27,14 +27,17 @@ def test_malformed_input():
def test_build_in_get():
- di = ld_dict([{"http://xmlns.com/foaf/0.1/name": [{"@value": "Manu Sporny"}],
- "http://xmlns.com/foaf/0.1/homepage": [{"@id": "http://manu.sporny.org/"}],
- "http://xmlns.com/foaf/0.1/foo": [{"http://xmlns.com/foaf/0.1/foobar": [{"@value": "bar"}],
- "http://xmlns.com/foaf/0.1/barfoo": [{"@value": "foo"}]}]}],
- context=[{"xmlns": "http://xmlns.com/foaf/0.1/"}])
- assert di["xmlns:name"] == ["Manu Sporny"]
- assert di["xmlns:homepage"] == ["http://manu.sporny.org/"]
- assert di["xmlns:foo"] == [{"xmlns:foobar": ["bar"], "xmlns:barfoo": ["foo"]}]
+ di = ld_dict([{"http://schema.org/name": [{"@value": "Manu Sporny"}],
+ "http://schema.org/homepage": [{"@id": "http://manu.sporny.org/"}],
+ "http://schema.org/foo": [{"http://schema.org/foobar": [{"@value": "bar"}],
+ "http://schema.org/barfoo": [{"@value": "foo"}]}]}],
+ context=[{"schema": "http://schema.org/"}])
+ assert isinstance(di["schema:name"], ld_list) and di["schema:name"].item_list == [{"@value": "Manu Sporny"}]
+ assert isinstance(di["schema:homepage"], ld_list)
+ assert di["schema:homepage"].item_list == [{"@id": "http://manu.sporny.org/"}]
+ assert isinstance(di["http://schema.org/foo"], ld_list) and isinstance(di["http://schema.org/foo"][0], ld_dict)
+ assert di["http://schema.org/foo"][0].data_dict == {"http://schema.org/foobar": [{"@value": "bar"}],
+ "http://schema.org/barfoo": [{"@value": "foo"}]}
with pytest.raises(KeyError):
di["bar"]
@@ -189,10 +192,28 @@ def test_get():
context=[{"schema": "https://schema.org/"}])
assert di.get("https://schema.org/name") == ["Manu Sporny"]
assert di.get("schema:name") == ["Manu Sporny"]
- assert di.get("bar", None) is None # invalid key
+ assert di.get("bar", None) is None
with pytest.raises(KeyError):
di.get("bar")
- assert isinstance(di.get("schema:bar", None), ld_list) and len(di.get("schema:bar", None)) == 0
+
+
+def test_setdefault():
+ di = ld_dict([{"https://schema.org/name": [{"@value": "Manu Sporny"}]}],
+ context=[{"schema": "https://schema.org/"}])
+ assert di.setdefault("schema:name", []) == [{"@value": "Manu Sporny"}]
+ assert di.setdefault("schema:email", []) == []
+ assert di["schema:email"] == []
+
+
+def test_emplace():
+ di = ld_dict([{"https://schema.org/name": [{"@value": "Manu Sporny"}]}],
+ context=[{"schema": "https://schema.org/"}])
+ di.emplace("schema:name")
+ assert di["schema:name"] == [{"@value": "Manu Sporny"}]
+ with pytest.raises(KeyError):
+ di["schema:email"]
+ di.emplace("schema:email")
+ assert di["schema:email"] == []
def test_update():
@@ -255,11 +276,12 @@ def test_items():
inner_di = ld_dict([{}], parent=di, key="http://xmlns.com/foaf/0.1/foo")
inner_di.update({"xmlns:foobar": "bar", "http://xmlns.com/foaf/0.1/barfoo": {"@id": "foo"}})
di.update({"http://xmlns.com/foaf/0.1/name": "foo", "xmlns:homepage": {"@id": "bar"}, "xmlns:foo": inner_di})
- assert [*di.items()][0:2] == [("http://xmlns.com/foaf/0.1/name", ["foo"]),
- ("http://xmlns.com/foaf/0.1/homepage", ["bar"])]
- assert [*di.items()][2][0] == "http://xmlns.com/foaf/0.1/foo"
- assert [*di.items()][2][1][0] == {"http://xmlns.com/foaf/0.1/foobar": [{"@value": "bar"}],
- "http://xmlns.com/foaf/0.1/barfoo": [{"@id": "foo"}]}
+ items = [*di.items()]
+ assert (items[0][0], items[1][0]) == ("http://xmlns.com/foaf/0.1/name", "http://xmlns.com/foaf/0.1/homepage")
+ assert (items[0][1].item_list, items[1][1].item_list) == ([{"@value": "foo"}], [{"@id": "bar"}])
+ assert items[2][0] == "http://xmlns.com/foaf/0.1/foo" and isinstance(items[2][1], ld_list)
+ assert items[2][1][0].data_dict == {"http://xmlns.com/foaf/0.1/foobar": [{"@value": "bar"}],
+ "http://xmlns.com/foaf/0.1/barfoo": [{"@id": "foo"}]}
def test_ref():
@@ -277,13 +299,13 @@ def test_to_python():
inner_di = ld_dict([{}], parent=di)
inner_di.update({"xmlns:foobar": "bar", "http://xmlns.com/foaf/0.1/barfoo": {"@id": "foo"}})
di.update({"http://xmlns.com/foaf/0.1/name": "foo", "xmlns:homepage": {"@id": "bar"}, "xmlns:foo": inner_di})
- assert di.to_python() == {"xmlns:name": ["foo"], "xmlns:homepage": ["bar"],
- "xmlns:foo": [{"xmlns:foobar": ["bar"], "xmlns:barfoo": ["foo"]}]}
+ assert di.to_python() == {"xmlns:name": ["foo"], "xmlns:homepage": [{"@id": "bar"}],
+ "xmlns:foo": [{"xmlns:foobar": ["bar"], "xmlns:barfoo": [{"@id": "foo"}]}]}
di.update({"http://spam.eggs/eggs": {
"@value": "2022-02-22T00:00:00", "@type": "https://schema.org/DateTime"
}})
- assert di.to_python() == {"xmlns:name": ["foo"], "xmlns:homepage": ["bar"],
- "xmlns:foo": [{"xmlns:foobar": ["bar"], "xmlns:barfoo": ["foo"]}],
+ assert di.to_python() == {"xmlns:name": ["foo"], "xmlns:homepage": [{"@id": "bar"}],
+ "xmlns:foo": [{"xmlns:foobar": ["bar"], "xmlns:barfoo": [{"@id": "foo"}]}],
"http://spam.eggs/eggs": ["2022-02-22T00:00:00"]}
@@ -354,13 +376,16 @@ def test_from_dict():
def test_is_ld_dict():
assert not any(ld_dict.is_ld_dict(item) for item in [{}, {"foo": "bar"}, {"@id": "foo"}])
- assert not any(ld_dict.is_ld_dict(item) for item in [[{"@id": "foo"}], [{"@set": "foo"}], [{}, {}], [], [""]])
- assert all(ld_dict.is_ld_dict([item]) for item in [{"@id": "foo", "foobar": "bar"}, {"foo": "bar"}])
+ assert not any(ld_dict.is_ld_dict(item) for item in [[{"@set": "foo"}], [{}, {}], [], [""]])
+ assert all(
+ ld_dict.is_ld_dict([item])
+ for item in [{"@id": "foo"}, {"@id": "foo", "foobar": "bar"}, {"foo": "bar"}]
+ )
def test_is_json_dict():
assert not any(ld_dict.is_json_dict(item) for item in [1, "", [], {""}, ld_dict([{}])])
assert not any(ld_dict.is_json_dict({key: [], "foo": "bar"}) for key in ["@set", "@graph", "@list", "@value"])
- assert not ld_dict.is_json_dict({"@id": "foo"})
+ assert ld_dict.is_json_dict({"@id": "foo"})
assert ld_dict.is_json_dict({"@id": "foo", "foobar": "bar"})
assert ld_dict.is_json_dict({"foo": "bar"})