# Source code for prov.model

    """Python implementation of the W3C Provenance Data Model (PROV-DM), including
    support for PROV-JSON import/export

    References:

    PROV-DM: http://www.w3.org/TR/prov-dm/
    PROV-JSON: https://openprovenance.org/prov-json/
    """

    from __future__ import annotations  # needed for | type annotations in Python < 3.10

    from collections import defaultdict
    import datetime
    import io
    import itertools
    import logging
    import os
    import shutil
    import tempfile
    from io import IOBase
    from typing import (
        Any,
        Callable,
        Iterable,
        Optional,
        Union,
    )
    import typing  # to use typing.TypeAlias in comments for compatibility with Python 3.9
    from urllib.parse import urlparse

    import dateutil.parser
    from prov import Error, serializers
    from prov.constants import *
    from prov.identifier import Identifier, QualifiedName, Namespace

    __author__ = "Trung Dong Huynh"
    __email__ = "trungdong@donggiang.com"

    logger = logging.getLogger(__name__)

    # Type aliases for convenience
    QualifiedNameCandidate = Union[QualifiedName, str, Identifier]  # type: typing.TypeAlias
    OptionalID = Optional[QualifiedNameCandidate]  # type: typing.TypeAlias
    EntityRef = Union["ProvEntity", QualifiedNameCandidate]  # type: typing.TypeAlias
    ActivityRef = Union["ProvActivity", QualifiedNameCandidate]  # type: typing.TypeAlias
    AgentRef = Union[
        "ProvAgent", "ProvEntity", "ProvActivity", QualifiedNameCandidate
    ]  # type: typing.TypeAlias
    GenrationRef = Union["ProvGeneration", QualifiedNameCandidate]  # type: typing.TypeAlias
    UsageRef = Union["ProvUsage", QualifiedNameCandidate]  # type: typing.TypeAlias
    RecordAttributesArg = Union[
        dict[QualifiedNameCandidate, Any],
        Iterable[tuple[QualifiedNameCandidate, Any]],
    ]  # type: typing.TypeAlias
    NameValuePair = tuple[QualifiedName, Any]  # type: typing.TypeAlias
    DatetimeOrStr = Union[datetime.datetime, str]  # type: typing.TypeAlias
    NSCollection = Union[dict[str, str], Iterable[Namespace]]  # type: typing.TypeAlias
    PathLike = Union[str, bytes, os.PathLike]  # type: typing.TypeAlias

    # Data Types
    def _ensure_datetime(value: Optional[DatetimeOrStr]) -> Optional[datetime.datetime]:
        if isinstance(value, str):
            return dateutil.parser.parse(value)
        else:
            return value

    def parse_xsd_datetime(value: str) -> Optional[datetime.datetime]:
        try:
            return dateutil.parser.parse(value)
        except ValueError:
            pass
        return None

    def parse_boolean(value: str) -> Optional[bool]:
        if value.lower() in ("false", "0"):
            return False
        elif value.lower() in ("true", "1"):
            return True
        else:
            return None

    DATATYPE_PARSERS = {
        datetime.datetime: parse_xsd_datetime,
    }

    # Mappings for XSD datatypes to Python standard types
    SupportedXSDParsedTypes = Union[
        str, datetime.datetime, float, int, bool, Identifier, None
    ]  # type: typing.TypeAlias
    XSD_DATATYPE_PARSERS: dict[QualifiedName, Callable[[str], SupportedXSDParsedTypes]] = {
        XSD_STRING: str,
        XSD_DOUBLE: float,
        XSD_LONG: int,
        XSD_INT: int,
        XSD_BOOLEAN: parse_boolean,
        XSD_DATETIME: parse_xsd_datetime,
        XSD_ANYURI: Identifier,
    }

    def parse_xsd_types(value: str, datatype: QualifiedName) -> SupportedXSDParsedTypes:
        return (
            XSD_DATATYPE_PARSERS[datatype](value)
            if datatype in XSD_DATATYPE_PARSERS
            else None
        )

    def first(a_set: set[Any]) -> Any | None:
        return next(iter(a_set), None)

    def _ensure_multiline_string_triple_quoted(value: str) -> str:
        # converting the value to a string
        s = str(value)
        # Escaping any double quote
        s = s.replace('"', '\\"')
        if "\n" in s:
            return '"""%s"""' % s
        else:
            return '"%s"' % s

    def encoding_provn_value(
        value: str | datetime.datetime | float | bool | QualifiedName,
    ) -> str:
        if isinstance(value, str):
            return _ensure_multiline_string_triple_quoted(value)
        elif isinstance(value, datetime.datetime):
            return '"{0}" %% xsd:dateTime'.format(value.isoformat())
        elif isinstance(value, float):
            return '"%g" %%%% xsd:float' % value
        elif isinstance(value, bool):
            return '"%i" %%%% xsd:boolean' % value
        else:
            # TODO: QName export
            return str(value)

    class Literal(object):
        def __init__(
            self,
            value: Any,
            datatype: Optional[QualifiedName] = None,
            langtag: Optional[str] = None,
        ):
            self._value: str = str(value)  # value is always a string
            if langtag:
                if datatype is None:
                    logger.debug(
                        "Assuming prov:InternationalizedString as the type of "
                        '"%s"@%s' % (value, langtag)
                    )
                    datatype = PROV_INTERNATIONALIZEDSTRING
                # PROV JSON states that the type field must not be set when
                # using the lang attribute and PROV XML requires it to be an
                # internationalized string.
                elif datatype != PROV_INTERNATIONALIZEDSTRING:
                    logger.warning(
                        'Invalid data type (%s) for "%s"@%s, overridden as '
                        "prov:InternationalizedString." % (datatype, value, langtag)
                    )
                    datatype = PROV_INTERNATIONALIZEDSTRING
            self._datatype: Optional[QualifiedName] = datatype
            # langtag is always a string
            self._langtag: Optional[str] = str(langtag) if langtag is not None else None

        def __str__(self) -> str:
            return self.provn_representation()

        def __repr__(self) -> str:
            return "<Literal: %s>" % self.provn_representation()

        def __eq__(self, other: Any) -> bool:
            return (
                (
                    self._value == other.value
                    and self._datatype == other.datatype
                    and self._langtag == other.langtag
                )
                if isinstance(other, Literal)
                else False
            )

        def __ne__(self, other: Any) -> bool:
            return not (self == other)

        def __hash__(self) -> int:
            return hash((self._value, self._datatype, self._langtag))

        @property
        def value(self) -> str:
            return self._value

        @property
        def datatype(self) -> QualifiedName | None:
            return self._datatype

        @property
        def langtag(self) -> str | None:
            return self._langtag

        def has_no_langtag(self) -> bool:
            return self._langtag is None

        def provn_representation(self) -> str:
            if self._langtag:
                # a language tag can only go with prov:InternationalizedString
                return "%s@%s" % (
                    _ensure_multiline_string_triple_quoted(self._value),
                    str(self._langtag),
                )
            else:
                return "%s %%%% %s" % (
                    _ensure_multiline_string_triple_quoted(self._value),
                    str(self._datatype),
                )

    # Exceptions and warnings

    class ProvException(Error):
        """Base class for PROV model exceptions."""

        pass

    class ProvWarning(Warning):
        """Base class for PROV model warnings."""

        pass

    class ProvExceptionInvalidQualifiedName(ProvException):
        """Exception for an invalid qualified identifier name."""

        qname = None
        """Intended qualified name."""

        def __init__(self, qname: Any):
            """
            Constructor.

            :param qname: Invalid qualified name.
            """
            self.qname = qname

        def __str__(self) -> str:
            return "Invalid Qualified Name: %s" % self.qname

    class ProvElementIdentifierRequired(ProvException):
        """Exception for a missing element identifier."""

        def __str__(self) -> str:
            return (
                "An identifier is missing. All PROV elements require a valid " "identifier."
            )

    #  PROV records

    class ProvRecord(object):
        """Base class for PROV records."""

        FORMAL_ATTRIBUTES = ()  # type: tuple[QualifiedName, ...]
        """Formal attributes names of this record type, in the expected order."""

        _prov_type: Optional[QualifiedName] = None
        """PROV type of record."""

        def __init__(
            self,
            bundle: ProvBundle,
            identifier: Optional[QualifiedName],
            attributes: Optional[RecordAttributesArg] = None,
        ):
            """
            Constructor.

            :param bundle: Bundle for the PROV record.
            :param identifier: (Unique) identifier of the record.
            :param attributes: Attributes to associate with the record (default: None).
            """
            self._bundle = bundle
            self._identifier = identifier
            self._attributes: dict[QualifiedName, set] = defaultdict(set)
            if attributes:
                self.add_attributes(attributes)

        def __hash__(self) -> int:
            return hash((self.get_type(), self._identifier, frozenset(self.attributes)))

        def copy(self) -> "ProvRecord":
            """
            Return an exact copy of this record.
            """
            return PROV_REC_CLS[self.get_type()](
                self._bundle, self.identifier, self.attributes
            )

        def get_type(self) -> QualifiedName:
            """Returns the PROV type of the record."""
            if self._prov_type is not None:
                return self._prov_type
            else:
                raise NotImplementedError("Type not defined for this record.")

        def get_asserted_types(self) -> set[QualifiedName]:
            """Returns the set of all asserted PROV types of this record."""
            return self._attributes[PROV_TYPE]

        def add_asserted_type(self, type_identifier: QualifiedName) -> None:
            """
            Adds a PROV type assertion to the record.

            :param type_identifier: PROV namespace identifier to add.
            """
            self._attributes[PROV_TYPE].add(type_identifier)

        def get_attribute(self, attr_name: QualifiedNameCandidate) -> set:
            """
            Returns the attribute values (if any) for the specified attribute name).

            :param attr_name: Name of the attribute.
            :return: Set of value(s) of the specified attribute.
            :rtype: set
            """
            attr_name_qn = self._bundle.mandatory_valid_qname(attr_name)
            return self._attributes[attr_name_qn]

        @property
        def identifier(self) -> QualifiedName | None:
            """Record's identifier."""
            return self._identifier

        @property
        def attributes(self) -> list[tuple[QualifiedName, Any]]:
            """
            All record attributes.

            :return: List of tuples (name, value)
            """
            return [
                (attr_name, value)
                for attr_name, values in self._attributes.items()
                for value in values
            ]

        @property
        def args(self) -> tuple:
            """
            All values of the record's formal attributes.

            :return: Tuple
            """
            return tuple(
                first(self._attributes[attr_name]) for attr_name in self.FORMAL_ATTRIBUTES
            )

        @property
        def formal_attributes(self) -> tuple[tuple[QualifiedName, Any], ...]:
            """
            All names and values of the record's formal attributes.

            :return: Tuple of tuples (name, value)
            """
            return tuple(
                (attr_name, first(self._attributes[attr_name]))
                for attr_name in self.FORMAL_ATTRIBUTES
            )

        @property
        def extra_attributes(self) -> tuple[tuple[QualifiedName, Any], ...]:
            """
            All names and values of the record's attributes that are not formal
            attributes.

            :return: Tuple of tuples (name, value)
            """
            return tuple(
                (attr_name, attr_value)
                for attr_name, attr_value in self.attributes
                if attr_name not in self.FORMAL_ATTRIBUTES
            )

        @property
        def bundle(self) -> ProvBundle:
            """
            Bundle of the record.

            :return: :py:class:`ProvBundle`
            """
            return self._bundle

        @property
        def label(self) -> str:
            """Identifying label of the record."""
            return str(
                first(self._attributes[PROV_LABEL])
                if self._attributes[PROV_LABEL]
                else self._identifier
            )

        @property
        def value(self) -> Any:
            """Value of the record."""
            return self._attributes[PROV_VALUE]

        # Handling attributes
        def _auto_literal_conversion(self, literal: Any) -> Any:
            # This method normalise datatype for literals

            if isinstance(literal, ProvRecord):
                # Use the QName of the record as the literal
                literal = literal.identifier

            if isinstance(literal, str):
                return str(literal)
            elif isinstance(literal, QualifiedName):
                return self._bundle.valid_qualified_name(literal)
            elif isinstance(literal, Literal) and literal.has_no_langtag():
                if literal.datatype:
                    # try to convert a generic Literal object to Python standard type
                    # to match the JSON decoding's literal conversion
                    value = parse_xsd_types(literal.value, literal.datatype)
                else:
                    # A literal with no datatype nor langtag defined
                    # try auto-converting the value
                    value = self._auto_literal_conversion(literal.value)
                if value is not None:
                    return value

            # No conversion possible, return the original value
            return literal

        def add_attributes(self, attributes: RecordAttributesArg) -> None:
            """
            Add attributes to the record.

            :param attributes: Dictionary of attributes, with keys being qualified
                identifiers. Alternatively, an iterable of tuples (key, value) with the
                keys satisfying the same condition.
            """
            if attributes:
                if isinstance(attributes, dict):
                    # Converting the dictionary into a list of tuples
                    # (i.e. attribute-value pairs)
                    attributes = attributes.items()

                # Check if one of the attributes specifies that the current type
                # is a collection. In that case multiple attributes of the same
                # type are allowed.
                if PROV_ATTR_COLLECTION in [_i[0] for _i in attributes]:
                    is_collection = True
                else:
                    is_collection = False

                for attr_name, original_value in attributes:
                    if original_value is None:
                        continue

                    # make sure the attribute name is valid
                    attr = self._bundle.mandatory_valid_qname(attr_name)

                    if attr in PROV_ATTRIBUTE_QNAMES:
                        # Expecting a qualified name
                        if isinstance(original_value, ProvRecord):
                            # Use the identifier of the record, which must exist, as the value for this attribute
                            qname = original_value.identifier
                            if qname is None:
                                raise ProvException(
                                    f"Invalid value for attribute {attr}: {original_value}."
                                    f" The record has no identifier."
                                )
                        else:
                            qname = original_value
                        value = self._bundle.mandatory_valid_qname(qname)  # type: Any
                    elif attr in PROV_ATTRIBUTE_LITERALS:
                        # Expecting a datetime object or a string that can be parsed as a datetime
                        if isinstance(original_value, str):
                            value = parse_xsd_datetime(original_value)
                        else:
                            value = original_value
                        if not isinstance(value, datetime.datetime):
                            raise ProvException(
                                f"Invalid value for attribute {attr}: {original_value}. "
                                f"Expected a datetime object or a string that can be parsed"
                                f" as a datetime."
                            )
                    else:
                        value = self._auto_literal_conversion(original_value)

                    if value is None:
                        raise ProvException(
                            "Invalid value for attribute %s: %s" % (attr, original_value)
                        )

                    if (
                        not is_collection
                        and attr in PROV_ATTRIBUTES
                        and self._attributes[attr]
                    ):
                        existing_value = first(self._attributes[attr])
                        is_not_same_value = True
                        try:
                            is_not_same_value = value != existing_value
                        except TypeError:
                            # Cannot compare them
                            pass  # consider them different values

                        if is_not_same_value:
                            raise ProvException(
                                "Cannot have more than one value for attribute %s" % attr
                            )
                        else:
                            # Same value, ignore it
                            continue

                    self._attributes[attr].add(value)

        def __eq__(self, other: Any) -> bool:
            if not isinstance(other, ProvRecord):
                return False
            if self.get_type() != other.get_type():
                return False
            if self._identifier and not (self._identifier == other._identifier):
                return False

            return set(self.attributes) == set(other.attributes)

        def __str__(self) -> str:
            return self.get_provn()

        def get_provn(self) -> str:
            """
            Returns the PROV-N representation of the record.

            :return: String
            """
            items = []

            # Generating identifier
            relation_id = ""  # default blank
            if self._identifier:
                identifier = str(self._identifier)  # TODO: QName export
                if self.is_element():
                    items.append(identifier)
                else:
                    # this is a relation, which relation uses a semicolon to separate identifiers
                    relation_id = identifier + "; "

            # Writing out the formal attributes
            for attr in self.FORMAL_ATTRIBUTES:
                if attr in self._attributes and self._attributes[attr]:
                    # Formal attributes always have single values
                    value = first(self._attributes[attr])
                    # TODO: QName export
                    items.append(
                        value.isoformat()
                        if isinstance(value, datetime.datetime)
                        else str(value)
                    )
                else:
                    items.append("-")

            # Writing out the remaining attributes
            extra = []
            for attr in self._attributes:
                if attr not in self.FORMAL_ATTRIBUTES:
                    for value in self._attributes[attr]:
                        try:
                            # try if there is a prov-n representation defined
                            provn_represenation = value.provn_representation()
                        except AttributeError:
                            provn_represenation = encoding_provn_value(value)
                        # TODO: QName export
                        extra.append("%s=%s" % (str(attr), provn_represenation))

            if extra:
                items.append("[%s]" % ", ".join(extra))
            prov_n = "%s(%s%s)" % (
                PROV_N_MAP[self.get_type()],
                relation_id,
                ", ".join(items),
            )
            return prov_n

        def is_element(self) -> bool:
            """
            True, if the record is an element, False otherwise.

            :return: bool
            """
            return False

        def is_relation(self) -> bool:
            """
            True, if the record is a relation, False otherwise.

            :return: bool
            """
            return False

    #  Abstract classes for elements and relations

    class ProvElement(ProvRecord):
        """Provenance Element (nodes in the provenance graph)."""

        def __init__(
            self,
            bundle: ProvBundle,
            identifier: Optional[QualifiedName],
            attributes: Optional[RecordAttributesArg] = None,
        ):
            if identifier is None:
                # All types of PROV elements require a valid identifier
                raise ProvElementIdentifierRequired()

            super(ProvElement, self).__init__(bundle, identifier, attributes)

        def is_element(self) -> bool:
            """
            True, if the record is an element, False otherwise.

            :return: bool
            """
            return True

        def __repr__(self) -> str:
            return "<%s: %s>" % (self.__class__.__name__, self._identifier)

    class ProvRelation(ProvRecord):
        """Provenance Relationship (edge between nodes)."""

        def is_relation(self) -> bool:
            """
            True, if the record is a relation, False otherwise.

            :return: bool
            """
            return True

        def __repr__(self) -> str:
            identifier = " %s" % self._identifier if self._identifier else ""
            element_1, element_2 = [qname for _, qname in self.formal_attributes[:2]]
            return "<%s:%s (%s, %s)>" % (
                self.__class__.__name__,
                identifier,
                element_1,
                element_2,
            )

    # Component 1: Entities and Activities

    class ProvEntity(ProvElement):
        """Provenance Entity element"""

        _prov_type = PROV_ENTITY

        # Convenient assertions that take the current ProvEntity as the first
        # (formal) argument

        def wasGeneratedBy(
            self,
            activity: Optional[ActivityRef] = None,
            time: Optional[DatetimeOrStr] = None,
            attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvEntity:
            """
            Creates a new generation record to this entity.

            :param activity: Activity or string identifier of the activity involved in
                the generation (default: None).
            :param time: Optional time for the generation (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.generation(self, activity, time, other_attributes=attributes)
            return self

        def wasInvalidatedBy(
            self,
            activity: Optional[ActivityRef],
            time: Optional[DatetimeOrStr] = None,
            attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvEntity:
            """
            Creates a new invalidation record for this entity.

            :param activity: Activity or string identifier of the activity involved in
                the invalidation (default: None).
            :param time: Optional time for the invalidation (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.invalidation(self, activity, time, other_attributes=attributes)
            return self

        def wasDerivedFrom(
            self,
            usedEntity: EntityRef,
            activity: Optional[ActivityRef] = None,
            generation: Optional[GenrationRef] = None,
            usage: Optional[UsageRef] = None,
            attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvEntity:
            """
            Creates a new derivation record for this entity from a used entity.

            :param usedEntity: Entity or a string identifier for the used entity.
            :param activity: Activity or string identifier of the activity involved in
                the derivation (default: None).
            :param generation: Optional generation record to state qualified derivation
                through an internal generation (default: None).
            :param usage: Optional usage record to state qualified derivation through
                an internal usage (default: None).
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.derivation(
                self, usedEntity, activity, generation, usage, other_attributes=attributes
            )
            return self

        def wasAttributedTo(
            self, agent: AgentRef, attributes: Optional[RecordAttributesArg] = None
        ) -> ProvEntity:
            """
            Creates a new attribution record between this entity and an agent.

            :param agent: Agent or string identifier of the agent involved in the
                attribution.
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.attribution(self, agent, other_attributes=attributes)
            return self

        def alternateOf(self, alternate2: EntityRef) -> ProvEntity:
            """
            Creates a new alternate record between this and another entity.

            :param alternate2: Entity or a string identifier for the second entity.
            """
            self._bundle.alternate(self, alternate2)
            return self

        def specializationOf(self, generalEntity: EntityRef) -> ProvEntity:
            """
            Creates a new specialisation record for this from a general entity.

            :param generalEntity: Entity or a string identifier for the general entity.
            """
            self._bundle.specialization(self, generalEntity)
            return self

        def hadMember(self, entity: EntityRef) -> ProvEntity:
            """
            Creates a new membership record to an entity for a collection.

            :param entity: Entity to be added to the collection.
            """
            self._bundle.membership(self, entity)
            return self

    class ProvActivity(ProvElement):
        """Provenance Activity element."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_STARTTIME, PROV_ATTR_ENDTIME)

        _prov_type = PROV_ACTIVITY

        #  Convenient methods

        def set_time(
            self,
            startTime: Optional[datetime.datetime] = None,
            endTime: Optional[datetime.datetime] = None,
        ) -> None:
            """
            Sets the time this activity took place.

            :param startTime: Start time for the activity.
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param endTime: Start time for the activity.
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            """
            if startTime is not None:
                self._attributes[PROV_ATTR_STARTTIME] = {startTime}
            if endTime is not None:
                self._attributes[PROV_ATTR_ENDTIME] = {endTime}

        def get_startTime(self) -> datetime.datetime | None:
            """
            Returns the time the activity started.

            :return: :py:class:`datetime.datetime`
            """
            values = self._attributes[PROV_ATTR_STARTTIME]
            return first(values) if values else None

        def get_endTime(self) -> datetime.datetime | None:
            """
            Returns the time the activity ended.

            :return: :py:class:`datetime.datetime`
            """
            values = self._attributes[PROV_ATTR_ENDTIME]
            return first(values) if values else None

        # Convenient assertions that take the current ProvActivity as the first
        # (formal) argument

        def used(
            self,
            entity: EntityRef,
            time: Optional[DatetimeOrStr] = None,
            attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvActivity:
            """
            Creates a new usage record for this activity.

            :param entity: Entity or string identifier of the entity involved in
                the usage relationship (default: None).
            :param time: Optional time for the usage (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.usage(self, entity, time, other_attributes=attributes)
            return self

        def wasInformedBy(
            self, informant: ActivityRef, attributes: Optional[RecordAttributesArg] = None
        ) -> ProvActivity:
            """
            Creates a new communication record for this activity.

            :param informant: The informing activity (relationship source).
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.communication(self, informant, other_attributes=attributes)
            return self

        def wasStartedBy(
            self,
            trigger: Optional[EntityRef],
            starter: Optional[ActivityRef] = None,
            time: Optional[DatetimeOrStr] = None,
            attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvActivity:
            """
            Creates a new start record for this activity. The activity did not exist
            before the start by the trigger.

            :param trigger: Entity triggering the start of this activity.
            :param starter: Optional extra activity to state a qualified start
                through which the trigger entity for the start is generated
                (default: None).
            :param time: Optional time for the start (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.start(self, trigger, starter, time, other_attributes=attributes)
            return self

        def wasEndedBy(
            self,
            trigger: Optional[EntityRef],
            ender: Optional[ActivityRef] = None,
            time: Optional[DatetimeOrStr] = None,
            attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvActivity:
            """
            Creates a new end record for this activity.

            :param trigger: Entity triggering the end of this activity.
            :param ender: Optionally extra activity to state a qualified end through
                which the trigger entity for the end is generated (default: None).
            :param time: Optional time for the end (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.end(self, trigger, ender, time, other_attributes=attributes)
            return self

        def wasAssociatedWith(
            self,
            agent: AgentRef,
            plan: Optional[EntityRef] = None,
            attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvActivity:
            """
            Creates a new association record for this activity.

            :param agent: Agent or string identifier of the agent involved in the
                association (default: None).
            :param plan: Optionally extra entity to state qualified association through
                an internal plan (default: None).
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.association(self, agent, plan, other_attributes=attributes)
            return self

    class ProvGeneration(ProvRelation):
        """Provenance Generation relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_ENTITY, PROV_ATTR_ACTIVITY, PROV_ATTR_TIME)

        _prov_type = PROV_GENERATION

    class ProvUsage(ProvRelation):
        """Provenance Usage relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_ACTIVITY, PROV_ATTR_ENTITY, PROV_ATTR_TIME)

        _prov_type = PROV_USAGE

    class ProvCommunication(ProvRelation):
        """Provenance Communication relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_INFORMED, PROV_ATTR_INFORMANT)

        _prov_type = PROV_COMMUNICATION

    class ProvStart(ProvRelation):
        """Provenance Start relationship."""

        FORMAL_ATTRIBUTES = (
            PROV_ATTR_ACTIVITY,
            PROV_ATTR_TRIGGER,
            PROV_ATTR_STARTER,
            PROV_ATTR_TIME,
        )

        _prov_type = PROV_START

    class ProvEnd(ProvRelation):
        """Provenance End relationship."""

        FORMAL_ATTRIBUTES = (
            PROV_ATTR_ACTIVITY,
            PROV_ATTR_TRIGGER,
            PROV_ATTR_ENDER,
            PROV_ATTR_TIME,
        )

        _prov_type = PROV_END

    class ProvInvalidation(ProvRelation):
        """Provenance Invalidation relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_ENTITY, PROV_ATTR_ACTIVITY, PROV_ATTR_TIME)

        _prov_type = PROV_INVALIDATION

    # Component 2: Derivations

    class ProvDerivation(ProvRelation):
        """Provenance Derivation relationship."""

        FORMAL_ATTRIBUTES = (
            PROV_ATTR_GENERATED_ENTITY,
            PROV_ATTR_USED_ENTITY,
            PROV_ATTR_ACTIVITY,
            PROV_ATTR_GENERATION,
            PROV_ATTR_USAGE,
        )

        _prov_type = PROV_DERIVATION

    # Component 3: Agents, Responsibility, and Influence

    class ProvAgent(ProvElement):
        """Provenance Agent element."""

        _prov_type = PROV_AGENT

        # Convenient assertions that take the current ProvAgent as the first
        # (formal) argument

        def actedOnBehalfOf(
            self,
            responsible: AgentRef,
            activity: Optional[ActivityRef] = None,
            attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvAgent:
            """
            Creates a new delegation record on behalf of this agent.

            :param responsible: Agent the responsibility is delegated to.
            :param activity: Optionally extra activity to state qualified delegation
                internally (default: None).
            :param attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            self._bundle.delegation(
                self, responsible, activity, other_attributes=attributes
            )
            return self

    class ProvAttribution(ProvRelation):
        """Provenance Attribution relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_ENTITY, PROV_ATTR_AGENT)

        _prov_type = PROV_ATTRIBUTION

    class ProvAssociation(ProvRelation):
        """Provenance Association relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_ACTIVITY, PROV_ATTR_AGENT, PROV_ATTR_PLAN)

        _prov_type = PROV_ASSOCIATION

    class ProvDelegation(ProvRelation):
        """Provenance Delegation relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_DELEGATE, PROV_ATTR_RESPONSIBLE, PROV_ATTR_ACTIVITY)

        _prov_type = PROV_DELEGATION

    class ProvInfluence(ProvRelation):
        """Provenance Influence relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_INFLUENCEE, PROV_ATTR_INFLUENCER)

        _prov_type = PROV_INFLUENCE

    # Component 5: Alternate Entities

    class ProvSpecialization(ProvRelation):
        """Provenance Specialization relationship."""

        FORMAL_ATTRIBUTES = (
            PROV_ATTR_SPECIFIC_ENTITY,
            PROV_ATTR_GENERAL_ENTITY,
        )  # type: tuple[QualifiedName, ...]

        _prov_type = PROV_SPECIALIZATION

    class ProvAlternate(ProvRelation):
        """Provenance Alternate relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_ALTERNATE1, PROV_ATTR_ALTERNATE2)

        _prov_type = PROV_ALTERNATE

    class ProvMention(ProvSpecialization):
        """Provenance Mention relationship (specific Specialization)."""

        FORMAL_ATTRIBUTES = (
            PROV_ATTR_SPECIFIC_ENTITY,
            PROV_ATTR_GENERAL_ENTITY,
            PROV_ATTR_BUNDLE,
        )

        _prov_type = PROV_MENTION

    # Component 6: Collections

    class ProvMembership(ProvRelation):
        """Provenance Membership relationship."""

        FORMAL_ATTRIBUTES = (PROV_ATTR_COLLECTION, PROV_ATTR_ENTITY)

        _prov_type = PROV_MEMBERSHIP

    #  Class mappings from PROV record type
    PROV_REC_CLS = {
        PROV_ENTITY: ProvEntity,
        PROV_ACTIVITY: ProvActivity,
        PROV_GENERATION: ProvGeneration,
        PROV_USAGE: ProvUsage,
        PROV_COMMUNICATION: ProvCommunication,
        PROV_START: ProvStart,
        PROV_END: ProvEnd,
        PROV_INVALIDATION: ProvInvalidation,
        PROV_DERIVATION: ProvDerivation,
        PROV_AGENT: ProvAgent,
        PROV_ATTRIBUTION: ProvAttribution,
        PROV_ASSOCIATION: ProvAssociation,
        PROV_DELEGATION: ProvDelegation,
        PROV_INFLUENCE: ProvInfluence,
        PROV_SPECIALIZATION: ProvSpecialization,
        PROV_ALTERNATE: ProvAlternate,
        PROV_MENTION: ProvMention,
        PROV_MEMBERSHIP: ProvMembership,
    }

    DEFAULT_NAMESPACES = {"prov": PROV, "xsd": XSD, "xsi": XSI}

    #  Bundle

    class NamespaceManager(dict):
        """Manages namespaces for PROV documents and bundles."""

        parent = None  # type: Optional[NamespaceManager]
        """Parent :py:class:`NamespaceManager` this manager one is a child of."""

        def __init__(
            self,
            namespaces: Optional[NSCollection] = None,
            default: Optional[str] = None,
            parent: Optional[NamespaceManager] = None,
        ):
            """
            Constructor.

            :param namespaces: Optional namespaces to add to the manager
                (default: None).
            :param default: Optional default namespace to use (default: None).
            :param parent: Optional parent :py:class:`NamespaceManager` to make this
                namespace manager a child of (default: None).
            """
            dict.__init__(self)
            self._default_namespaces = DEFAULT_NAMESPACES
            self.update(self._default_namespaces)
            self._namespaces = {}  # type: dict[str, Namespace]

            if default is not None:
                self.set_default_namespace(default)
            else:
                self._default = None  # type: Optional[Namespace]
            self.parent = parent
            #  TODO check if default is in the default namespaces
            self._anon_id_count = 0
            self._uri_map = dict()  # type: dict[str, Namespace]
            self._rename_map = dict()  # type: dict[Namespace, Namespace]
            self._prefix_renamed_map = dict()  # type: dict[str, Namespace]
            if namespaces is not None:
                self.add_namespaces(namespaces)

        def get_namespace(self, uri: str) -> Namespace | None:
            """
            Returns the namespace prefix for the given URI.

            :param uri: Namespace URI.
            :return: :py:class:`~prov.identifier.Namespace`.
            """
            for namespace in self.values():
                if uri == namespace._uri:
                    return namespace
            return None

        def get_registered_namespaces(self) -> Iterable[Namespace]:
            """
            Returns all registered namespaces.

            :return: Iterable of :py:class:`~prov.identifier.Namespace`.
            """
            return self._namespaces.values()

        def set_default_namespace(self, uri: str) -> None:
            """
            Sets the default namespace to the one of a given URI.

            :param uri: Namespace URI.
            """
            self._default = Namespace("", uri)
            self[""] = self._default

        def get_default_namespace(self) -> Namespace | None:
            """
            Returns the default namespace.

            :return: :py:class:`~prov.identifier.Namespace`
            """
            return self._default

        def add_namespace(self, namespace: Namespace) -> Namespace:
            """
            Adds a namespace (if not available, yet).

            :param namespace: :py:class:`~prov.identifier.Namespace` to add.
            """
            if namespace in self.values():
                #  no need to do anything
                return namespace
            if namespace in self._rename_map:
                #  already renamed and added
                return self._rename_map[namespace]

            # Checking if the URI has been defined and use the existing namespace
            # instead
            uri = namespace.uri
            prefix = namespace.prefix

            if uri in self._uri_map:
                existing_ns = self._uri_map[uri]
                self._rename_map[namespace] = existing_ns
                self._prefix_renamed_map[prefix] = existing_ns
                return existing_ns

            if prefix in self:
                #  Conflicting prefix
                new_prefix = self._get_unused_prefix(prefix)
                new_namespace = Namespace(new_prefix, namespace.uri)
                self._rename_map[namespace] = new_namespace
                # TODO: What if the prefix is already in the map and point to a
                # different Namespace? Raise an exception?
                self._prefix_renamed_map[prefix] = new_namespace
                prefix = new_prefix
                namespace = new_namespace

            # Only now add the namespace to the registry
            self._namespaces[prefix] = namespace
            self[prefix] = namespace
            self._uri_map[uri] = namespace

            return namespace

        def add_namespaces(self, namespaces: NSCollection) -> None:
            """
            Add multiple namespaces into this manager.

            :param namespaces: A collection of namespace(s) to add.
            :type namespaces: List of :py:class:`~prov.identifier.Namespace` or
                dict of {prefix: uri}.
            :returns: None
            """
            if isinstance(namespaces, dict):
                # expecting a dictionary of {prefix: uri},
                # convert it to a list of Namespace
                namespaces = [Namespace(prefix, uri) for prefix, uri in namespaces.items()]
            if namespaces:
                for ns in namespaces:
                    self.add_namespace(ns)

        def valid_qualified_name(
            self, qname: QualifiedNameCandidate
        ) -> QualifiedName | None:
            """
            Resolves an identifier to a valid qualified name.

            :param qname: Qualified name as :py:class:`~prov.identifier.QualifiedName`
                or a tuple (namespace, identifier).
            :return: :py:class:`~prov.identifier.QualifiedName` or None in case of
                failure.
            """
            if not qname:
                return None

            if isinstance(qname, QualifiedName):
                #  Register the namespace if it has not been registered before
                namespace = qname.namespace
                prefix = namespace.prefix
                local_part = qname.localpart
                if not prefix:
                    # the namespace is a default namespace
                    if self._default == namespace:
                        # the same default namespace is defined
                        new_qname = self._default[local_part]
                    elif self._default is None:
                        # no default namespace is defined, reused the one given
                        self._default = namespace
                        return qname  # no change, return the original
                    else:
                        # different default namespace,
                        # use the 'dn' prefix for the new namespace
                        dn_namespace = Namespace("dn", namespace.uri)
                        dn_namespace = self.add_namespace(dn_namespace)
                        new_qname = dn_namespace[local_part]
                elif prefix in self and self[prefix] == namespace:
                    # No need to add the namespace
                    existing_ns = self[prefix]
                    if existing_ns is namespace:
                        return qname
                    else:
                        # reuse the existing namespace
                        new_qname = existing_ns[local_part]
                else:
                    # Do not reuse the namespace object, making an identical copy
                    ns = self.add_namespace(Namespace(namespace.prefix, namespace.uri))
                    # minting the same Qualified Name from the namespace's copy
                    new_qname = ns[qname.localpart]
                # returning the new qname
                return new_qname

            # Trying to generate a valid qualified name from here
            if not isinstance(qname, (str, Identifier)):
                # Only proceed with a string or URI value
                return None
            # Extract the URI string value if it is an identifier
            str_value = qname.uri if isinstance(qname, Identifier) else qname
            if str_value.startswith("_:"):
                # this is a blank node ID
                return None
            elif ":" in str_value:
                #  check if the identifier contains a registered prefix
                prefix, local_part = str_value.split(":", 1)
                if prefix in self:
                    #  return a new QualifiedName
                    return self[prefix][local_part]
                if prefix in self._prefix_renamed_map:
                    #  return a new QualifiedName
                    return self._prefix_renamed_map[prefix][local_part]
                else:
                    #  assuming it is a URI (with the first part as its scheme)
                    #  check if the URI can be compacted by any of the registered namespaces
                    for namespace in self.values():
                        if str_value.startswith(namespace.uri):
                            #  create a QName with the namespace
                            return namespace[str_value.replace(namespace.uri, "")]
            elif self._default and isinstance(qname, str):
                # no colon in the identifier and a default namespace is defined,
                # create and return a qualified name in the default namespace
                return self._default[qname]

            if self.parent:
                # all attempts have failed so far
                # now delegate this to the parent NamespaceManager
                return self.parent.valid_qualified_name(qname)

            # Default to FAIL
            return None

        def get_anonymous_identifier(self, local_prefix: str = "id") -> Identifier:
            """
            Returns an anonymous identifier (without a namespace prefix).

            :param local_prefix: Optional local namespace prefix as a string
                (default: 'id').
            :return: :py:class:`~prov.identifier.Identifier`
            """
            self._anon_id_count += 1
            return Identifier("_:%s%d" % (local_prefix, self._anon_id_count))

        def _get_unused_prefix(self, original_prefix: str) -> str:
            if original_prefix not in self:
                return original_prefix
            count = 1
            while True:
                new_prefix = "_".join((original_prefix, str(count)))
                if new_prefix in self:
                    count += 1
                else:
                    return new_prefix

    class ProvBundle(object):
        """PROV Bundle"""

        def __init__(
            self,
            records: Optional[Iterable[ProvRecord]] = None,
            identifier: Optional[QualifiedName] = None,
            namespaces: Optional[NSCollection] = None,
            document: Optional["ProvDocument"] = None,
        ):
            """
            Constructor.

            :param records: Optional iterable of records to add to the bundle
                (default: None).
            :param identifier: Optional identifier of the bundle (default: None).
            :param namespaces: Optional iterable of :py:class:`~prov.identifier.Namespace`s
                to set the document up with (default: None).
            :param document: Optional document to add to the bundle (default: None).
            """
            #  Initializing bundle-specific attributes
            self._identifier = identifier
            self._records = list()  # type: list[ProvRecord]
            self._id_map = defaultdict(list)  # type: dict[QualifiedName, list[ProvRecord]]
            self._document = document
            self._namespaces = NamespaceManager(
                namespaces, parent=(document._namespaces if document is not None else None)
            )  # type: NamespaceManager
            if records:
                for record in records:
                    self.add_record(record)

        def __repr__(self) -> str:
            return "<%s: %s>" % (self.__class__.__name__, self._identifier)

        @property
        def namespaces(self) -> set[Namespace]:
            """
            Returns the set of registered namespaces.

            :return: Set of :py:class:`~prov.identifier.Namespace`.
            """
            return set(self._namespaces.get_registered_namespaces())

        @property
        def default_ns_uri(self) -> str | None:
            """
            Returns the default namespace's URI, if any.

            :return: URI as string.
            """
            default_ns = self._namespaces.get_default_namespace()
            return default_ns.uri if default_ns else None

        @property
        def document(self) -> ProvDocument | None:
            """
            Returns the parent document, if any.

            :return: :py:class:`ProvDocument`.
            """
            return self._document

        @property
        def identifier(self) -> QualifiedName | None:
            """
            Returns the bundle's identifier
            """
            return self._identifier

        @property
        def records(self) -> list[ProvRecord]:
            """
            Returns the list of all records in the current bundle
            """
            return list(self._records)

        #  Bundle configurations

        def set_default_namespace(self, uri: str) -> None:
            """
            Sets the default namespace through a given URI.

            :param uri: Namespace URI.
            """
            self._namespaces.set_default_namespace(uri)

        def get_default_namespace(self) -> Namespace | None:
            """
            Returns the default namespace.

            :return: :py:class:`~prov.identifier.Namespace`
            """
            return self._namespaces.get_default_namespace()

        def add_namespace(
            self, namespace_or_prefix: Namespace | str, uri: Optional[str] = None
        ) -> Namespace:
            """
            Adds a namespace (if not available, yet).

            :param namespace_or_prefix: :py:class:`~prov.identifier.Namespace` or its
                prefix as a string to add.
            :param uri: Namespace URI (default: None). Must be present if only a
                prefix is given in the previous parameter.
            """
            if isinstance(namespace_or_prefix, Namespace):
                return self._namespaces.add_namespace(namespace_or_prefix)
            else:
                if uri is not None:
                    return self._namespaces.add_namespace(
                        Namespace(namespace_or_prefix, uri)
                    )
                else:
                    raise ProvException("Cannot add a namespace without a URI")

        def get_registered_namespaces(self) -> Iterable[Namespace]:
            """
            Returns all registered namespaces.

            :return: Iterable of :py:class:`~prov.identifier.Namespace`.
            """
            return self._namespaces.get_registered_namespaces()

        def valid_qualified_name(
            self, identifier: QualifiedNameCandidate
        ) -> Optional[QualifiedName]:
            return self._namespaces.valid_qualified_name(identifier)

        def mandatory_valid_qname(
            self, identifier: QualifiedNameCandidate
        ) -> QualifiedName:
            """
            Determines if the given identifier is a valid qualified name and returns it.
            If the provided identifier is not valid, an exception is raised.
            """
            valid_qname = self.valid_qualified_name(identifier)
            if valid_qname is not None:
                return valid_qname
            else:
                raise ProvExceptionInvalidQualifiedName(identifier)

        def get_records(
            self, class_or_type_or_tuple: Optional[type | tuple[type]] = None
        ) -> Iterable[ProvRecord]:
            """
            Returns all records. Returned records may be filtered by the optional
            argument.

            :param class_or_type_or_tuple: A filter on the type for which records are
                to be returned (default: None). The filter checks by the type of the
                record using the `isinstance` check on the record.
            :return: List of :py:class:`ProvRecord` objects.
            """
            results = list(self._records)  # make a (shallow) copy of the record list
            if class_or_type_or_tuple:
                return filter(lambda rec: isinstance(rec, class_or_type_or_tuple), results)
            else:
                return results

        def get_record(self, identifier: QualifiedNameCandidate) -> list[ProvRecord]:
            """
            Returns one or more records matching a given identifier.

            :param identifier: Record identifier.
            :return: List of :py:class:`ProvRecord`
            """
            valid_id = self.valid_qualified_name(identifier)
            return list(self._id_map[valid_id]) if valid_id is not None else []

        # Miscellaneous functions

        def is_document(self) -> bool:
            """
            `True` if the object is a document, `False` otherwise.

            :return: bool
            """
            return False

        def is_bundle(self) -> bool:
            """
            `True` if the object is a bundle, `False` otherwise.

            :return: bool
            """
            return True

        def has_bundles(self) -> bool:
            """
            `True` if the object has at least one bundle, `False` otherwise.

            :return: bool
            """
            return False

        @property
        def bundles(self) -> Iterable[ProvBundle]:
            """
            Returns bundles contained in the document

            :return: Iterable of :py:class:`ProvBundle`.
            """
            raise ProvException("A PROV bundle does not contain sub-bundles")

        def get_provn(self, _indent_level: int = 0) -> str:
            """
            Returns the PROV-N representation of the bundle.

            :return: String
            """
            indentation = "" + ("  " * _indent_level)
            newline = "\n" + ("  " * (_indent_level + 1))

            #  if this is the document, start the document;
            # otherwise, start the bundle
            lines = ["document"] if self.is_document() else ["bundle %s" % self._identifier]

            default_namespace = self._namespaces.get_default_namespace()
            if default_namespace:
                lines.append("default <%s>" % default_namespace.uri)

            registered_namespaces = self._namespaces.get_registered_namespaces()
            if registered_namespaces:
                lines.extend(
                    [
                        "prefix %s <%s>" % (namespace.prefix, namespace.uri)
                        for namespace in registered_namespaces
                    ]
                )

            if default_namespace or registered_namespaces:
                #  a blank line between the prefixes and the assertions
                lines.append("")

            #  adding all the records
            lines.extend([record.get_provn() for record in self._records])
            if self.is_document():
                # Print out bundles
                lines.extend(bundle.get_provn(_indent_level + 1) for bundle in self.bundles)
            provn_str = newline.join(lines) + "\n"

            #  closing the structure
            provn_str += indentation + (
                "endDocument" if self.is_document() else "endBundle"
            )
            return provn_str

        def __eq__(self, other: Any) -> bool:
            if not isinstance(other, ProvBundle):
                return False
            other_records = set(other.get_records())
            this_records = set(self.get_records())
            if len(this_records) != len(other_records):
                return False
            #  check if all records for equality
            for record_a in this_records:
                #  Manually look for the record
                found = False
                for record_b in other_records:
                    if record_a == record_b:
                        other_records.remove(record_b)
                        found = True
                        break
                if not found:
                    logger.debug(
                        "Equality (ProvBundle): Could not find this record: %s",
                        str(record_a),
                    )
                    return False
            return True

        def __ne__(self, other: Any) -> bool:
            return not (self == other)

        __hash__ = None  # type: ignore

        # type: ignore # type: ignore # Transformations
        def _unified_records(self) -> list[ProvRecord]:
            """Returns a list of unified records."""
            # TODO: Check unification rules in the PROV-CONSTRAINTS document
            # This method simply merges the records having the same name
            merged_records = dict()
            for identifier, records in self._id_map.items():
                if len(records) > 1:
                    # more than one record having the same identifier
                    # merge the records
                    merged = records[0].copy()
                    for record in records[1:]:
                        merged.add_attributes(record.attributes)
                    # map all of them to the merged record
                    for record in records:
                        merged_records[record] = merged
            if not merged_records:
                # No merging done, just return the list of original records
                return list(self._records)

            added_merged_records = set()
            unified_records = list()
            for record in self._records:
                if record in merged_records:
                    merged = merged_records[record]
                    if merged not in added_merged_records:
                        unified_records.append(merged)
                        added_merged_records.add(merged)
                else:
                    # add the original record
                    unified_records.append(record)
            return unified_records

        def unified(self) -> ProvBundle:
            """
            Unifies all records in the bundle that haves same identifiers

            :returns: :py:class:`ProvBundle` -- the new unified bundle.
            """
            unified_records = self._unified_records()
            bundle = ProvBundle(records=unified_records, identifier=self.identifier)
            return bundle

        def update(self, other: ProvBundle) -> None:
            """
            Append all the records of the *other* ProvBundle into this bundle.

            :param other: the other bundle whose records to be appended.
            :type other: :py:class:`ProvBundle`
            :returns: None.
            """
            if isinstance(other, ProvBundle):
                if other.is_document() and other.has_bundles():
                    # Cannot add bundles to a bundle
                    raise ProvException(
                        "ProvBundle.update(): The other bundle is a document with "
                        "sub-bundle(s)."
                    )
                for record in other.get_records():
                    self.add_record(record)
            else:
                raise ProvException(
                    "ProvBundle.update(): The other bundle is not a ProvBundle "
                    "instance (%s)" % type(other)
                )

        # Provenance statements
        def _add_record(self, record: ProvRecord) -> None:
            # IMPORTANT: All records need to be added to a bundle/document via this
            # method. Otherwise, the _id_map dict will not be correctly updated
            identifier = record.identifier
            if identifier is not None:
                self._id_map[identifier].append(record)
            self._records.append(record)

        def new_record(
            self,
            record_type: QualifiedName,
            identifier: OptionalID,
            attributes: Optional[RecordAttributesArg] = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvRecord:
            """
            Creates a new record.

            :param record_type: Type of record (one of :py:const:`PROV_REC_CLS`).
            :param identifier: Identifier for new record.
            :param attributes: Attributes as a dictionary or list of tuples to be added
                to the record optionally (default: None).
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attr_list = []  # type: list[tuple[QualifiedNameCandidate, Any]]
            if attributes:
                if isinstance(attributes, dict):
                    attr_list.extend((attr, value) for attr, value in attributes.items())
                else:
                    # expecting a list of attributes here
                    attr_list.extend(attributes)
            if other_attributes:
                attr_list.extend(
                    other_attributes.items()
                    if isinstance(other_attributes, dict)
                    else other_attributes
                )
            record_identifier = (
                self.valid_qualified_name(identifier) if identifier else None
            )
            new_record = PROV_REC_CLS[record_type](self, record_identifier, attr_list)
            self._add_record(new_record)
            return new_record

        def add_record(self, record: ProvRecord) -> ProvRecord:
            """
            Adds a new record that to the bundle.

            :param record: :py:class:`ProvRecord` to be added.
            """
            return self.new_record(
                record.get_type(),
                record.identifier,
                record.formal_attributes,
                record.extra_attributes,
            )

        def entity(
            self,
            identifier: QualifiedNameCandidate,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvEntity:
            """
            Creates a new entity.

            :param identifier: Identifier for new entity.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            return self.new_record(PROV_ENTITY, identifier, None, other_attributes)  # type: ignore

        def activity(
            self,
            identifier: QualifiedNameCandidate,
            startTime: Optional[DatetimeOrStr] = None,
            endTime: Optional[DatetimeOrStr] = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvActivity:
            """
            Creates a new activity.

            :param identifier: Identifier for new activity.
            :param startTime: Optional start time for the activity (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param endTime: Optional start time for the activity (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_STARTTIME: _ensure_datetime(startTime),
                PROV_ATTR_ENDTIME: _ensure_datetime(endTime),
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_ACTIVITY,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def generation(
            self,
            entity: EntityRef,
            activity: Optional[ActivityRef] = None,
            time: Optional[DatetimeOrStr] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvRecord:
            """
            Creates a new generation record for an entity.

            :param entity: Entity or a string identifier for the entity.
            :param activity: Activity or string identifier of the activity involved in
                the generation (default: None).
            :param time: Optional time for the generation (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param identifier: Identifier for new generation record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_ENTITY: entity,
                PROV_ATTR_ACTIVITY: activity,
                PROV_ATTR_TIME: _ensure_datetime(time),
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_GENERATION,
                identifier,
                attributes,
                other_attributes,
            )

        def usage(
            self,
            activity: ActivityRef,
            entity: Optional[EntityRef] = None,
            time: Optional[DatetimeOrStr] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvUsage:
            """
            Creates a new usage record for an activity.

            :param activity: Activity or a string identifier for the entity.
            :param entity: Entity or string identifier of the entity involved in
                the usage relationship (default: None).
            :param time: Optional time for the usage (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param identifier: Identifier for new usage record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_ACTIVITY: activity,
                PROV_ATTR_ENTITY: entity,
                PROV_ATTR_TIME: _ensure_datetime(time),
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_USAGE,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def start(
            self,
            activity: ActivityRef,
            trigger: Optional[EntityRef] = None,
            starter: Optional[ActivityRef] = None,
            time: Optional[DatetimeOrStr] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvStart:
            """
            Creates a new start record for an activity.

            :param activity: Activity or a string identifier for the entity.
            :param trigger: Entity triggering the start of this activity.
            :param starter: Optionally extra activity to state a qualified start
                through which the trigger entity for the start is generated
                (default: None).
            :param time: Optional time for the start (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param identifier: Identifier for new start record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_ACTIVITY: activity,
                PROV_ATTR_TRIGGER: trigger,
                PROV_ATTR_STARTER: starter,
                PROV_ATTR_TIME: _ensure_datetime(time),
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_START,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def end(
            self,
            activity: ActivityRef,
            trigger: Optional[EntityRef] = None,
            ender: Optional[ActivityRef] = None,
            time: Optional[DatetimeOrStr] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvEnd:
            """
            Creates a new end record for an activity.

            :param activity: Activity or a string identifier for the entity.
            :param trigger: trigger: Entity triggering the end of this activity.
            :param ender: Optionally extra activity to state a qualified end
                through which the trigger entity for the end is generated
                (default: None).
            :param time: Optional time for the end (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param identifier: Identifier for new end record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_ACTIVITY: activity,
                PROV_ATTR_TRIGGER: trigger,
                PROV_ATTR_ENDER: ender,
                PROV_ATTR_TIME: _ensure_datetime(time),
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_END,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def invalidation(
            self,
            entity: EntityRef,
            activity: Optional[ActivityRef] = None,
            time: Optional[DatetimeOrStr] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvInvalidation:
            """
            Creates a new invalidation record for an entity.

            :param entity: Entity or a string identifier for the entity.
            :param activity: Activity or string identifier of the activity involved in
                the invalidation (default: None).
            :param time: Optional time for the invalidation (default: None).
                Either a :py:class:`datetime.datetime` object or a string that can be
                parsed by :py:func:`dateutil.parser`.
            :param identifier: Identifier for the new invalidation record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_ENTITY: entity,
                PROV_ATTR_ACTIVITY: activity,
                PROV_ATTR_TIME: _ensure_datetime(time),
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_INVALIDATION,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def communication(
            self,
            informed: ActivityRef,
            informant: ActivityRef,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvCommunication:
            """
            Creates a new communication record for an entity.

            :param informed: The informed activity (relationship destination).
            :param informant: The informing activity (relationship source).
            :param identifier: Identifier for new communication record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_INFORMED: informed,
                PROV_ATTR_INFORMANT: informant,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_COMMUNICATION,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def agent(
            self,
            identifier: QualifiedNameCandidate,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvAgent:
            """
            Creates a new agent.

            :param identifier: Identifier for new agent.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            return self.new_record(PROV_AGENT, identifier, None, other_attributes)  # type: ignore

        def attribution(
            self,
            entity: EntityRef,
            agent: AgentRef,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvAttribution:
            """
            Creates a new attribution record between an entity and an agent.

            :param entity: Entity or a string identifier for the entity (relationship
                source).
            :param agent: Agent or string identifier of the agent involved in the
                attribution (relationship destination).
            :param identifier: Identifier for new attribution record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_ENTITY: entity,
                PROV_ATTR_AGENT: agent,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_ATTRIBUTION,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def association(
            self,
            activity: ActivityRef,
            agent: Optional[AgentRef] = None,
            plan: Optional[EntityRef] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvAssociation:
            """
            Creates a new association record for an activity.

            :param activity: Activity or a string identifier for the activity.
            :param agent: Agent or string identifier of the agent involved in the
                association (default: None).
            :param plan: Optionally extra entity to state qualified association through
                an internal plan (default: None).
            :param identifier: Identifier for new association record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_ACTIVITY: activity,
                PROV_ATTR_AGENT: agent,
                PROV_ATTR_PLAN: plan,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_ASSOCIATION,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def delegation(
            self,
            delegate: AgentRef,
            responsible: AgentRef,
            activity: Optional[ActivityRef] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvDelegation:
            """
            Creates a new delegation record on behalf of an agent.

            :param delegate: Agent delegating the responsibility (relationship source).
            :param responsible: Agent the responsibility is delegated to (relationship
                destination).
            :param activity: Optionally extra activity to state qualified delegation
                internally (default: None).
            :param identifier: Identifier for new association record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_DELEGATE: delegate,
                PROV_ATTR_RESPONSIBLE: responsible,
                PROV_ATTR_ACTIVITY: activity,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_DELEGATION,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def influence(
            self,
            influencee: EntityRef | ActivityRef | AgentRef,
            influencer: EntityRef | ActivityRef | AgentRef,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvInfluence:
            """
            Creates a new influence record between two entities, activities or agents.

            :param influencee: Influenced entity, activity or agent (relationship
                source).
            :param influencer: Influencing entity, activity or agent (relationship
                destination).
            :param identifier: Identifier for new influence record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_INFLUENCEE: influencee,
                PROV_ATTR_INFLUENCER: influencer,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_INFLUENCE,
                identifier,
                attributes,
                other_attributes,
            )  # type: ignore

        def derivation(
            self,
            generatedEntity: EntityRef,
            usedEntity: EntityRef,
            activity: Optional[ActivityRef] = None,
            generation: Optional[GenrationRef] = None,
            usage: Optional[UsageRef] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvDerivation:
            """
            Creates a new derivation record for a generated entity from a used entity.

            :param generatedEntity: Entity or a string identifier for the generated
                entity (relationship source).
            :param usedEntity: Entity or a string identifier for the used entity
                (relationship destination).
            :param activity: Activity or string identifier of the activity involved in
                the derivation (default: None).
            :param generation: Optionally extra activity to state qualified generation
                through a generation (default: None).
            :param usage: XXX (default: None).
            :param identifier: Identifier for new derivation record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            attributes = {
                PROV_ATTR_GENERATED_ENTITY: generatedEntity,
                PROV_ATTR_USED_ENTITY: usedEntity,
                PROV_ATTR_ACTIVITY: activity,
                PROV_ATTR_GENERATION: generation,
                PROV_ATTR_USAGE: usage,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_DERIVATION, identifier, attributes, other_attributes
            )  # type: ignore

        def revision(
            self,
            generatedEntity: EntityRef,
            usedEntity: EntityRef,
            activity: Optional[ActivityRef] = None,
            generation: Optional[GenrationRef] = None,
            usage: Optional[UsageRef] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvDerivation:
            """
            Creates a new revision record for a generated entity from a used entity.

            :param generatedEntity: Entity or a string identifier for the generated
                entity (relationship source).
            :param usedEntity: Entity or a string identifier for the used entity
                (relationship destination).
            :param activity: Activity or string identifier of the activity involved in
                the revision (default: None).
            :param generation: Optionally to state qualified revision through a
                generation activity (default: None).
            :param usage: XXX (default: None).
            :param identifier: Identifier for new revision record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            record = self.derivation(
                generatedEntity,
                usedEntity,
                activity,
                generation,
                usage,
                identifier,
                other_attributes,
            )
            record.add_asserted_type(PROV["Revision"])
            return record

        def quotation(
            self,
            generatedEntity: EntityRef,
            usedEntity: EntityRef,
            activity: Optional[ActivityRef] = None,
            generation: Optional[GenrationRef] = None,
            usage: Optional[UsageRef] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvDerivation:
            """
            Creates a new quotation record for a generated entity from a used entity.

            :param generatedEntity: Entity or a string identifier for the generated
                entity (relationship source).
            :param usedEntity: Entity or a string identifier for the used entity
                (relationship destination).
            :param activity: Activity or string identifier of the activity involved in
                the quotation (default: None).
            :param generation: Optionally to state qualified quotation through a
                generation activity (default: None).
            :param usage: XXX (default: None).
            :param identifier: Identifier for new quotation record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            record = self.derivation(
                generatedEntity,
                usedEntity,
                activity,
                generation,
                usage,
                identifier,
                other_attributes,
            )
            record.add_asserted_type(PROV["Quotation"])
            return record

        def primary_source(
            self,
            generatedEntity: EntityRef,
            usedEntity: EntityRef,
            activity: Optional[ActivityRef] = None,
            generation: Optional[GenrationRef] = None,
            usage: Optional[UsageRef] = None,
            identifier: OptionalID = None,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvDerivation:
            """
            Creates a new primary source record for a generated entity from a used
            entity.

            :param generatedEntity: Entity or a string identifier for the generated
                entity (relationship source).
            :param usedEntity: Entity or a string identifier for the used entity
                (relationship destination).
            :param activity: Activity or string identifier of the activity involved in
                the primary source (default: None).
            :param generation: Optionally to state qualified primary source through a
                generation activity (default: None).
            :param usage: XXX (default: None).
            :param identifier: Identifier for new primary source record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            record = self.derivation(
                generatedEntity,
                usedEntity,
                activity,
                generation,
                usage,
                identifier,
                other_attributes,
            )
            record.add_asserted_type(PROV["PrimarySource"])
            return record  # type: ignore

        def specialization(
            self, specificEntity: EntityRef, generalEntity: EntityRef
        ) -> ProvSpecialization:
            """
            Creates a new specialisation record for a specific from a general entity.

            :param specificEntity: Entity or a string identifier for the specific
                entity (relationship source).
            :param generalEntity: Entity or a string identifier for the general entity
                (relationship destination).
            """
            attributes = {
                PROV_ATTR_SPECIFIC_ENTITY: specificEntity,
                PROV_ATTR_GENERAL_ENTITY: generalEntity,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_SPECIALIZATION,
                None,
                attributes,
            )  # type: ignore

        def alternate(self, alternate1: EntityRef, alternate2: EntityRef) -> ProvAlternate:
            """
            Creates a new alternate record between two entities.

            :param alternate1: Entity or a string identifier for the first entity
                (relationship source).
            :param alternate2: Entity or a string identifier for the second entity
                (relationship destination).
            """
            attributes = {
                PROV_ATTR_ALTERNATE1: alternate1,
                PROV_ATTR_ALTERNATE2: alternate2,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_ALTERNATE,
                None,
                attributes,
            )  # type: ignore

        def mention(
            self, specificEntity: EntityRef, generalEntity: EntityRef, bundle: EntityRef
        ) -> ProvMention:
            """
            Creates a new mention record for a specific from a general entity.

            :param specificEntity: Entity or a string identifier for the specific
                entity (relationship source).
            :param generalEntity: Entity or a string identifier for the general entity
                (relationship destination).
            :param bundle: XXX
            """
            attributes = {
                PROV_ATTR_SPECIFIC_ENTITY: specificEntity,
                PROV_ATTR_GENERAL_ENTITY: generalEntity,
                PROV_ATTR_BUNDLE: bundle,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_MENTION,
                None,
                attributes,
            )  # type: ignore

        def collection(
            self,
            identifier: QualifiedNameCandidate,
            other_attributes: Optional[RecordAttributesArg] = None,
        ) -> ProvEntity:
            """
            Creates a new collection record for a particular record.

            :param identifier: Identifier for new collection record.
            :param other_attributes: Optional other attributes as a dictionary or list
                of tuples to be added to the record optionally (default: None).
            """
            record = self.new_record(PROV_ENTITY, identifier, None, other_attributes)
            record.add_asserted_type(PROV["Collection"])
            return record  # type: ignore

        def membership(self, collection: EntityRef, entity: EntityRef) -> ProvMembership:
            """
            Creates a new membership record for an entity to a collection.

            :param collection: Collection the entity is to be added to.
            :param entity: Entity to be added to the collection.
            """
            attributes = {
                PROV_ATTR_COLLECTION: collection,
                PROV_ATTR_ENTITY: entity,
            }  # type: dict[QualifiedNameCandidate, Any]
            return self.new_record(
                PROV_MEMBERSHIP,
                None,
                attributes,
            )  # type: ignore

        def plot(
            self,
            filename: Optional[PathLike] = None,
            show_nary: bool = True,
            use_labels: bool = False,
            show_element_attributes: bool = True,
            show_relation_attributes: bool = True,
        ) -> None:
            """
            Convenience function to plot a PROV document.

            :param filename: The filename to save to. If not given, it will open
                an interactive matplotlib plot. The filetype is determined from
                the filename ending.
            :type filename: String
            :param show_nary: Shows all elements in n-ary relations.
            :type show_nary: bool
            :param use_labels: Uses the `prov:label` property of an element as its
                name (instead of its identifier).
            :type use_labels: bool
            :param show_element_attributes: Shows attributes of elements.
            :type show_element_attributes: bool
            :param show_relation_attributes: Shows attributes of relations.
            :type show_relation_attributes: bool
            """
            # Lazy imports to have soft dependencies on pydot and matplotlib
            # (imported even later).
            from prov import dot

            if filename:
                format = str(os.path.splitext(filename))[-1].lower().strip(os.path.extsep)
            else:
                format = "png"
            format = format.lower()
            d = dot.prov_to_dot(
                self,
                show_nary=show_nary,
                use_labels=use_labels,
                show_element_attributes=show_element_attributes,
                show_relation_attributes=show_relation_attributes,
            )
            method = "create_%s" % format
            if not hasattr(d, method):
                raise ValueError("Format '%s' cannot be saved." % format)
            with io.BytesIO() as buf:
                buf.write(getattr(d, method)())

                buf.seek(0, 0)
                if filename:
                    with open(filename, "wb") as fh:
                        fh.write(buf.read())
                else:
                    # Use matplotlib to show the image as it likely is more
                    # widespread than PIL and works nicely in the ipython notebook.
                    import matplotlib.pylab as plt  # type: ignore
                    import matplotlib.image as mpimg  # type: ignore

                    max_size = 30

                    img = mpimg.imread(buf)
                    # pydot makes a border around the image. remove it.
                    img = img[1:-1, 1:-1]
                    size = (img.shape[1] / 100.0, img.shape[0] / 100.0)
                    if max(size) > max_size:
                        scale = max_size / max(size)
                    else:
                        scale = 1.0
                    size = (scale * size[0], scale * size[1])

                    plt.figure(figsize=size)
                    plt.subplots_adjust(bottom=0, top=1, left=0, right=1)
                    plt.xticks([])
                    plt.yticks([])
                    plt.imshow(img)
                    plt.axis("off")
                    plt.show()

        #  Aliases
        wasGeneratedBy = generation
        used = usage
        wasStartedBy = start
        wasEndedBy = end
        wasInvalidatedBy = invalidation
        wasInformedBy = communication
        wasAttributedTo = attribution
        wasAssociatedWith = association
        actedOnBehalfOf = delegation
        wasInfluencedBy = influence
        wasDerivedFrom = derivation
        wasRevisionOf = revision
        wasQuotedFrom = quotation
        hadPrimarySource = primary_source
        alternateOf = alternate
        specializationOf = specialization
        mentionOf = mention
        hadMember = membership

    class ProvDocument(ProvBundle):
        """Provenance Document."""

        def __init__(
            self,
            records: Optional[Iterable[ProvRecord]] = None,
            namespaces: Optional[NSCollection] = None,
        ):
            """
            Constructor.

            :param records: Optional records to add to the document (default: None).
            :param namespaces: Optional iterable of :py:class:`~prov.identifier.Namespace`s
                to set the document up with (default: None).
            """
            ProvBundle.__init__(
                self, records=records, identifier=None, namespaces=namespaces
            )
            self._bundles = dict()  # type: dict[QualifiedName, ProvBundle]

        def __repr__(self) -> str:
            return "<ProvDocument>"

        def __eq__(self, other: Any) -> bool:
            if not isinstance(other, ProvDocument):
                return False
            # Comparing the documents' content
            if not super(ProvDocument, self).__eq__(other):
                return False

            # Comparing the documents' bundles
            for b_id, bundle in self._bundles.items():
                if b_id not in other._bundles:
                    return False
                other_bundle = other._bundles[b_id]
                if bundle != other_bundle:
                    return False

            # Everything is the same
            return True

        def is_document(self) -> bool:
            """
            `True` if the object is a document, `False` otherwise.

            :return: bool
            """
            return True

        def is_bundle(self) -> bool:
            """
            `True` if the object is a bundle, `False` otherwise.

            :return: bool
            """
            return False

        def has_bundles(self) -> bool:
            """
            `True` if the object has at least one bundle, `False` otherwise.

            :return: bool
            """
            return len(self._bundles) > 0

        @property
        def bundles(self) -> Iterable[ProvBundle]:
            """
            Returns bundles contained in the document

            :return: Iterable of :py:class:`ProvBundle`.
            """
            return self._bundles.values()

        # Transformations

        def flattened(self) -> ProvDocument:
            """
            Flattens the document by moving all the records in its bundles up
            to the document level.

            :returns: :py:class:`ProvDocument` -- the (new) flattened document.
            """
            if self._bundles:
                # Creating a new document for all the records
                new_doc = ProvDocument()
                bundled_records = itertools.chain(
                    *[b.get_records() for b in self._bundles.values()]
                )
                for record in itertools.chain(self._records, bundled_records):
                    new_doc.add_record(record)
                return new_doc
            else:
                # returning the same document
                return self

        def unified(self) -> ProvDocument:
            """
            Returns a new document containing all records having the same identifiers
            unified (including those inside bundles).

            :return: :py:class:`ProvDocument`
            """
            document = ProvDocument(self._unified_records())
            document._namespaces = self._namespaces
            for bundle in self.bundles:
                unified_bundle = bundle.unified()
                document.add_bundle(unified_bundle)
            return document

        def update(self, other: ProvBundle) -> None:
            """
            Append all the records of the *other* document/bundle into this document.
            Bundles having the same identifiers will be merged.

            :param other: The other document/bundle whose records to be appended.
            :type other: :py:class:`ProvDocument` or :py:class:`ProvBundle`
            :returns: None.
            """
            if isinstance(other, ProvBundle):
                for record in other.get_records():
                    self.add_record(record)
                if other.has_bundles():
                    for bundle in other.bundles:
                        bundle_id = bundle.identifier
                        assert bundle_id is not None
                        if bundle.identifier in self._bundles:
                            self._bundles[bundle.identifier].update(bundle)
                        else:
                            new_bundle = self.bundle(bundle_id)
                            new_bundle.update(bundle)
            else:
                raise ProvException(
                    "ProvDocument.update(): The other is not a ProvDocument or "
                    "ProvBundle instance (%s)" % type(other)
                )

        # Bundle operations

        def add_bundle(
            self, bundle: ProvBundle, identifier: Optional[QualifiedName] = None
        ) -> None:
            """
            Add a bundle to the current document.

            :param bundle: The bundle to add to the document.
            :type bundle: :py:class:`ProvBundle`
            :param identifier: The (optional) identifier to use for the bundle
                (default: None). If none given, use the identifier from the bundle
                itself.
            """
            if not isinstance(bundle, ProvBundle):
                raise ProvException(
                    "Only a ProvBundle instance can be added as a bundle in a "
                    "ProvDocument."
                )

            if bundle.is_document():
                if bundle.has_bundles():
                    raise ProvException(
                        "Cannot add a document with nested bundles as a bundle."
                    )
                # Make it a new ProvBundle
                new_bundle = ProvBundle(namespaces=bundle.namespaces)
                new_bundle.update(bundle)
                bundle = new_bundle

            if identifier is None:
                identifier = bundle.identifier

            if not identifier:
                raise ProvException("The provided bundle has no identifier")

            # Link the bundle namespace manager to the document's
            bundle._namespaces.parent = self._namespaces

            valid_id = bundle.mandatory_valid_qname(identifier)
            # IMPORTANT: Rewriting the bundle identifier for consistency
            bundle._identifier = valid_id

            if valid_id in self._bundles:
                raise ProvException("A bundle with that identifier already exists")

            self._bundles[valid_id] = bundle
            bundle._document = self

        def bundle(self, identifier: QualifiedNameCandidate) -> ProvBundle:
            """
            Returns a new bundle from the current document.

            :param identifier: The identifier to use for the bundle.
            :return: :py:class:`ProvBundle`
            """
            if identifier is None:
                raise ProvException(
                    "An identifier is required. Cannot create an unnamed bundle."
                )
            valid_id = self.valid_qualified_name(identifier)
            if valid_id is None:
                raise ProvException(
                    'The provided identifier "%s" is not valid' % identifier
                )
            if valid_id in self._bundles:
                raise ProvException("A bundle with that identifier already exists")
            b = ProvBundle(identifier=valid_id, document=self)
            self._bundles[valid_id] = b
            return b

        # Serializing and deserializing

        def serialize(
            self,
            destination: Optional[io.IOBase | PathLike] = None,
            format: str = "json",
            **args: Any,
        ) -> str | None:
            """
            Serialize the :py:class:`ProvDocument` to the destination.

            Available serializers can be queried by the value of
            `:py:attr:~prov.serializers.Registry.serializers` after loading them via
            `:py:func:~prov.serializers.Registry.load_serializers()`.

            :param destination: Stream object to serialize the output to. Default is
                `None`, which serializes as a string.
            :param format: Serialization format (default: 'json'), defaulting to
                PROV-JSON.
            :return: Serialization in a string if no destination was given,
                None otherwise.
            """
            serializer = serializers.get(format)(self)
            if destination is None:
                buffer = io.StringIO()
                serializer.serialize(buffer, **args)
                return buffer.getvalue()

            if isinstance(destination, IOBase):
                stream = destination
                serializer.serialize(stream, **args)
            else:
                location = str(destination)
                scheme, netloc, path, params, _query, fragment = urlparse(location)
                if netloc != "":
                    print(
                        "WARNING: not saving as location " + "is not a local file reference"
                    )
                    return None
                fd, name = tempfile.mkstemp()
                stream = os.fdopen(fd, "wb")
                serializer.serialize(stream, **args)
                stream.close()
                if hasattr(shutil, "move"):
                    shutil.move(name, path)
                else:
                    shutil.copy(name, path)
                    os.remove(name)
            return None

        @staticmethod
        def deserialize(
            source: Optional[io.IOBase | PathLike] = None,
            content: Optional[str | bytes] = None,
            format: str = "json",
            **args: Any,
        ) -> ProvDocument:
            """
            Deserialize the :py:class:`ProvDocument` from source (a stream or a
            file path) or directly from a string content.

            Available serializers can be queried by the value of
            `:py:attr:~prov.serializers.Registry.serializers` after loading them via
            `:py:func:~prov.serializers.Registry.load_serializers()`.

            Note: Not all serializers support deserialization.

            :param source: Stream object to deserialize the PROV document from
                (default: None).
            :param content: String to deserialize the PROV document from
                (default: None).
            :param format: Serialization format (default: 'json'), defaulting to
                PROV-JSON.
            :return: :py:class:`ProvDocument`
            """
            serializer = serializers.get(format)()

            if content is not None:
                # io.StringIO only accepts unicode strings
                stream = io.StringIO(
                    content if isinstance(content, str) else content.decode()
                )
                return serializer.deserialize(stream, **args)

            if source is not None:
                if isinstance(source, io.IOBase):
                    return serializer.deserialize(source, **args)
                else:
                    with open(source) as f:
                        return serializer.deserialize(f, **args)

            raise TypeError("Either source or content must be provided")

    def sorted_attributes(
        element: QualifiedName, attributes: Iterable[NameValuePair]
    ) -> list[NameValuePair]:
        """
        Helper function sorting attributes into the order required by PROV-XML.

        :param element: The prov element used to derive the type and the
            attribute order for the type.
        :param attributes: The attributes to sort.
        """
        attributes = list(attributes)
        order = list(PROV_REC_CLS[element].FORMAL_ATTRIBUTES)

        # Append label, location, role, type, and value attributes. This is
        # universal amongst all elements.
        order.extend([PROV_LABEL, PROV_LOCATION, PROV_ROLE, PROV_TYPE, PROV_VALUE])

        # Sort function. The PROV XML specification talks about alphabetical
        # sorting. We now interpret it as sorting by tag including the prefix
        # first and then sorting by the text, also including the namespace
        # prefix if given.
        def sort_fct(x: NameValuePair) -> tuple[str, str]:
            return str(x[0]), str(x[1].value if hasattr(x[1], "value") else x[1])

        sorted_elements = []
        for item in order:
            this_type_list = []
            for e in list(attributes):
                if e[0] != item:
                    continue
                this_type_list.append(e)
                attributes.remove(e)
            this_type_list.sort(key=sort_fct)
            sorted_elements.extend(this_type_list)
        # Add remaining attributes. According to the spec, the other attributes
        # have a fixed alphabetical order.
        attributes.sort(key=sort_fct)
        sorted_elements.extend(attributes)

        return sorted_elements