IfcOpenShell/src/ifcopenshell-python/ifcopenshell/file.py

# IfcOpenShell - IFC toolkit and geometry engine
# Copyright (C) 2021 Thomas Krijnen <thomas@aecgeeks.com>
#
# This file is part of IfcOpenShell.
#
# IfcOpenShell is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# IfcOpenShell is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with IfcOpenShell.  If not, see <http://www.gnu.org/licenses/>.

from __future__ import annotations

import functools
import numbers
import os
import re
import zipfile
from collections.abc import Callable, Generator
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, Optional, TypedDict, Union, overload

from typing_extensions import assert_never

import ifcopenshell
from ifcopenshell.util.mvd_info import LARK_AVAILABLE, MvdInfo

from . import ifcopenshell_wrapper

if TYPE_CHECKING:
    import ifcopenshell.util.schema

    InverseReference = tuple[int, Any]
    ElementInverses = dict[int, list[InverseReference]]

    class CreateOperation(TypedDict):
        action: Literal["create"]
        value: Any

    class EditOperation(TypedDict):
        action: Literal["edit"]
        id: int
        index: int
        old: Any
        new: Any

    class DeleteOperation(TypedDict):
        action: Literal["delete"]
        inverses: ElementInverses
        value: Any

    class BatchDeleteOperation(TypedDict):
        action: Literal["batch_delete"]
        inverses: ElementInverses

    TransactionOperation = Union[CreateOperation, EditOperation, DeleteOperation, BatchDeleteOperation]

HEADER_FIELDS = {
    "file_description": [
        "description",
        "implementation_level",
    ],
    "file_name": [
        "name",
        "time_stamp",
        "author",
        "organization",
        "preprocessor_version",
        "originating_system",
        "authorization",
    ],
}


class UndoSystemError(Exception):
    def __init__(self, message: str, transaction: Transaction):
        super().__init__(message)
        self.transaction = transaction


class Transaction:
    operations: list[TransactionOperation]
    batch_inverses: list[ElementInverses]
    batch_delete_ids: set[int]

    def __init__(self, ifc_file: ifcopenshell.file):
        self.file: ifcopenshell.file = ifc_file
        self.operations = []
        self.is_batched = False
        self.batch_delete_index = 0
        self.batch_delete_ids = set()
        self.batch_inverses = []

    def serialise_entity_instance(self, element: ifcopenshell.entity_instance) -> dict[str, Any]:
        info = element.get_info()
        for key, value in info.items():
            info[key] = self.serialise_value(element, value)
        return info

    def serialise_value(self, element, value) -> Any:
        return element.walk(
            lambda v: isinstance(v, ifcopenshell.entity_instance),
            lambda v: {"id": v.id()} if v.id() else {"type": v.is_a(), "value": v.wrappedValue},
            value,
        )

    def unserialise_value(self, element, value) -> Any:
        return element.walk(
            lambda v: isinstance(v, dict),
            lambda v: self.file.by_id(v["id"]) if v.get("id") else self.file.create_entity(v["type"], v["value"]),
            value,
        )

    def batch(self) -> None:
        self.is_batched = True
        self.batch_delete_index = len(self.operations)
        self.batch_delete_ids = set()
        self.batch_inverses = []

    def unbatch(self) -> None:
        for inverses in self.batch_inverses:
            if inverses:
                self.operations.insert(self.batch_delete_index, {"action": "batch_delete", "inverses": inverses})
        self.is_batched = False
        self.batch_delete_index = 0
        self.batch_delete_ids = set()
        self.batch_inverses = []

    def store_create(self, element: ifcopenshell.entity_instance) -> None:
        if element.id():
            self.operations.append({"action": "create", "value": self.serialise_entity_instance(element)})

    def store_edit(self, element: ifcopenshell.entity_instance, index: int, value: Any) -> None:
        if element.id():
            self.operations.append(
                {
                    "action": "edit",
                    "id": element.id(),
                    "index": index,
                    "old": self.serialise_value(element, element[index]),
                    "new": self.serialise_value(element, value),
                }
            )

    def store_delete(self, element: ifcopenshell.entity_instance) -> None:
        inverses = {}
        if self.is_batched:
            if element.id() not in self.batch_delete_ids:
                self.batch_inverses.append(self.get_element_inverses(element))
            self.batch_delete_ids.add(element.id())
        else:
            inverses = self.get_element_inverses(element)
        self.operations.append(
            {"action": "delete", "inverses": inverses, "value": self.serialise_entity_instance(element)}
        )

    def get_element_inverses(self, element: ifcopenshell.entity_instance) -> ElementInverses:
        inverses: ElementInverses = {}
        for inverse in self.file.get_inverse(element):
            inverse_references: list[InverseReference] = []
            for i, attribute in enumerate(inverse):
                if self.has_element_reference(attribute, element):
                    inverse_references.append((i, self.serialise_value(inverse, attribute)))
            inverses[inverse.id()] = inverse_references
        return inverses

    def has_element_reference(self, value: Any, element: ifcopenshell.entity_instance) -> bool:
        if isinstance(value, (tuple, list)):
            for v in value:
                if self.has_element_reference(v, element):
                    return True
            return False
        return value == element

    def rollback(self) -> None:
        for operation in self.operations[::-1]:
            if operation["action"] == "create":
                element = self.file.by_id(operation["value"]["id"])
                if hasattr(element, "GlobalId") and element.GlobalId is None:
                    # hack, otherwise ifcopenshell gets upset
                    element.GlobalId = "x"
                self.file.remove(element)
            elif operation["action"] == "edit":
                element = self.file.by_id(operation["id"])
                try:
                    element[operation["index"]] = self.unserialise_value(element, operation["old"])
                except:
                    # Catch discrepancy where IfcOpenShell creates but doesn't allow editing of invalid values
                    pass
            elif operation["action"] == "delete":
                e = self.file.create_entity(operation["value"]["type"], id=operation["value"]["id"])
                for k, v in operation["value"].items():
                    try:
                        setattr(e, k, self.unserialise_value(e, v))
                    except:
                        # Catch discrepancy where IfcOpenShell creates but doesn't allow editing of invalid values
                        pass
                for inverse_id, data in operation["inverses"].items():
                    inverse = self.file.by_id(inverse_id)
                    for index, value in data:
                        inverse[index] = self.unserialise_value(inverse, value)
            elif operation["action"] == "batch_delete":
                for inverse_id, data in operation["inverses"].items():
                    inverse = self.file.by_id(inverse_id)
                    for index, value in data:
                        inverse[index] = self.unserialise_value(inverse, value)
            else:
                assert_never(operation["action"])

    def commit(self) -> None:
        for operation in self.operations:
            if operation["action"] == "create":
                e = self.file.create_entity(operation["value"]["type"], id=operation["value"]["id"])
                for k, v in operation["value"].items():
                    try:
                        setattr(e, k, self.unserialise_value(e, v))
                    except:
                        # Catch discrepancy where IfcOpenShell creates but doesn't allow editing of invalid values
                        pass
            elif operation["action"] == "edit":
                element = self.file.by_id(operation["id"])
                element[operation["index"]] = self.unserialise_value(element, operation["new"])
            elif operation["action"] == "delete":
                element = self.file.by_id(operation["value"]["id"])
                self.file.remove(element)
            elif operation["action"] == "batch_delete":
                pass
            else:
                assert_never(operation["action"])


import struct


def consume_buffer(val, inner):
    while val:
        s = struct.unpack("@q", val[:8])[0]
        val = val[8:]
        yield inner(val[0:s])
        val = val[s:]


binary_deserializers = (
    lambda __, _: None,
    lambda __, _: None,
    lambda __, val: struct.unpack("@i", val)[0],
    lambda __, val: val[0] == 1,
    # @todo 3 state
    lambda __, val: val[0] == 1,
    lambda __, val: struct.unpack("@d", val)[0],
    lambda __, val: val.decode("utf-8"),
    lambda __, val: val.decode("utf-8"),
    lambda storage, val: ifcopenshell_wrapper.schema_by_name(storage.schema_identifier)
    .declarations()[struct.unpack("@q", val[:8])[0]]
    .enumeration_items()[struct.unpack("@q", val[8:])[0]],
    lambda storage, val: storage.by_id((val[0] == 105, struct.unpack("@q", val[1:])[0])),
    lambda __, _: (),
    lambda __, val: struct.unpack("@" + "i" * (len(val) // 4), val),
    lambda __, val: struct.unpack("@" + "d" * (len(val) // 8), val),
    lambda __, val: tuple(consume_buffer(val, lambda inner: inner.decode("utf-8"))),
    lambda __, val: tuple(consume_buffer(val, lambda inner: inner.decode("utf-8"))),
    lambda storage, val: tuple(
        storage.by_id((val[i * 9] == 105, struct.unpack("@q", val[i * 9 + 1 : i * 9 + 9])[0]))
        for i in range(len(val) // 9)
    ),
    lambda __, _: ((),),
    lambda __, val: tuple(
        consume_buffer(val, lambda inner: struct.unpack("@" + 105 * (len(inner) // struct.calcsize("@i")), inner))
    ),
    lambda __, val: tuple(
        consume_buffer(val, lambda inner: struct.unpack("@" + "d" * (len(inner) // struct.calcsize("@d")), inner))
    ),
    lambda storage, val: tuple(
        consume_buffer(
            val,
            lambda inner: tuple(
                storage.by_id((inner[i * 9] == 105, struct.unpack("@q", inner[i * 9 + 1 : i * 9 + 9])[0]))
                for i in range(len(inner) // 9)
            ),
        )
    ),
)


@functools.cache
def attribute_lookup(schema_name, entity_name):
    decl = ifcopenshell_wrapper.schema_by_name(schema_name).declaration_by_name(entity_name)
    attributes = decl.as_entity().all_attributes()
    di = {v: k for k, v in enumerate(a.name() for a in attributes)}
    all_inverses = decl.as_entity().all_inverse_attributes()
    for inv in all_inverses:

        def visit(decl):
            yield decl.index_in_schema()
            for ty in decl.subtypes():
                yield from visit(ty)

        attribute_index = inv.entity_reference().attribute_index(inv.attribute_reference())
        entity_indices = list(visit(inv.entity_reference()))
        di[inv.name()] = (entity_indices, attribute_index)
    return di


@functools.cache
def entity_name_lookup(schema_name, index):
    return ifcopenshell_wrapper.schema_by_name(schema_name).declarations()[index].name()


class rocksdb_lazy_instance:
    __slots__ = ("storage", "name")

    def _transform_value(self, val: bytes) -> Any:
        if not val:
            return None
        # ord('A') is 65
        return binary_deserializers[val[0] - 65](self.storage, val[1:])

    def __init__(self, storage, name):
        self.storage = storage
        self.name = name

    @functools.cache
    def is_a(self):
        if self.name.startswith("h|"):
            return self.name[2:]
        idx = struct.unpack("@q", self.storage.read(f"{self.name}|_"))[0]
        return entity_name_lookup(self.storage.schema_identifier, idx)

    def __getattr__(self, name):
        attr = attribute_lookup(self.storage.schema_identifier, self.is_a()).get(name)
        if isinstance(attr, int):
            return self[attr]
        else:
            entity_indices, attribute_index = attr

            def _():
                for index_in_schema in entity_indices:
                    buffer = self.storage.read(f"v|{self.name[2:]}|{index_in_schema}|{attribute_index}") or b""
                    yield from map(self.storage.by_id, struct.unpack("<" + "I" * (len(buffer) // 4), buffer))

            return list(_())

    def __getitem__(self, index):
        return self._transform_value(self.storage.read(f"{self.name}|{index}"))

    @functools.cache
    def __len__(self):
        return (
            max(
                map(
                    int,
                    filter(
                        lambda s: s.isdigit(),
                        (k.split(b"|")[2] for k, v in self.storage.prefix(f"{self.name}|").items()),
                    ),
                ),
                default=-1,
            )
            + 1
        )

    def __iter__(self):
        for i in range(len(self)):
            yield self[i]

    def __repr__(self):
        pre = f"#{self.name[2:]}=" if self.name.startswith("i|") else ""

        def val_repr(val):
            if isinstance(val, rocksdb_lazy_instance):
                if val.name[0] == "i":
                    return f"#{val.name[2:]}"
                else:
                    return repr(val)
            elif isinstance(val, (tuple, list)):
                return f'({",".join(map(val_repr, val))})'
            elif val is None:
                return "$"
            else:
                return repr(val)

        return f'{pre}{self.is_a()}({",".join(map(val_repr, self))})'

    def id(self):
        if self.name.startswith("i|"):
            return int(self.name[2:])
        else:
            # compatibility with C++
            return 0

    def __bool__(self):
        return len(self) > 0

    @property
    def _comparison_tuple(self):
        return self.storage.file.file_pointer(), self.name

    def __eq__(self, other):
        if not isinstance(other, rocksdb_lazy_instance):
            return False
        return self._comparison_tuple == other._comparison_tuple

    def __hash__(self):
        return hash(self._comparison_tuple)


class rocksdb_file_storage:
    def __init__(self, file, prefix=""):
        self.file = file
        self._prefix = prefix

    def items(self):
        it = self.file.key_value_store_iter(self._prefix)
        while it and it.valid():
            yield it.key(), it.value()
            it.next()

    def read(self, key):
        return self.file.key_value_store_query(key)

    def by_id(self, name):
        if isinstance(name, tuple):
            inst = rocksdb_lazy_instance(self, f'{"i" if name[0] else "t"}|{name[1]}')
        else:
            inst = rocksdb_lazy_instance(self, f"i|{name}")
        if not inst:
            raise KeyError(f"Instance with name {name} not found in file")
        return inst

    def by_type(self, ty: str):
        def visit(decl):
            yield decl.index_in_schema()
            for ty in decl.subtypes():
                yield from visit(ty)

        decl = ifcopenshell_wrapper.schema_by_name(self.schema_identifier).declaration_by_name(ty)

        def _():
            for index in visit(decl):
                buff = self.read(f"t|{index}") or b""
                yield from map(self.by_id, struct.unpack("@" + "q" * (len(buff) // 8), buff))

        return list(_())

    __getitem__ = by_id

    def __iter__(self):
        previous = None
        for k, v in self.items():
            if k.startswith(b"i|"):
                name = int(k[2:].split(b"|")[0])
                if name != previous:
                    previous = name
                    yield rocksdb_lazy_instance(self, f"i|{name}")

    def prefix(self, prefix):
        return rocksdb_file_storage(self.file, self._prefix + prefix)

    @functools.cached_property
    def schema_identifier(self):
        """Full IFC schema version: IFC2X3_TC1, IFC4_ADD2, IFC4X3_ADD2, etc."""
        return rocksdb_lazy_instance(self, f"h|file_schema")[0][0]

    @functools.cached_property
    def schema(self) -> ifcopenshell.util.schema.IFC_SCHEMA:
        """General IFC schema version: IFC2X3, IFC4, IFC4X3."""
        prefixes = ("IFC", "X", "_ADD", "_TC")
        reg = "".join(f"(?P<{s}>{s}\\d+)?" for s in prefixes)
        match = re.match(reg, self.schema_identifier)
        version_tuple = tuple(
            map(
                lambda pp: int(pp[1][len(pp[0]) :]) if pp[1] else None,
                ((p, match.group(p)) for p in prefixes),
            )
        )
        return "".join("".join(map(str, t)) if t[1] else "" for t in zip(prefixes, version_tuple[0:2]))


class file_mixin:
    """Base class for containing IFC files.

    Class has instance methods for filtering by element Id, Type, etc.
    Instantiated objects can be subscripted by Id or Guid

    Example:

    .. code:: python

        model = ifcopenshell.open(file_path)
        products = model.by_type("IfcProduct")
        print(products[0].id(), products[0].GlobalId) # 122 2XQ$n5SLP5MBLyL442paFx
        print(products[0] == model[122] == model["2XQ$n5SLP5MBLyL442paFx"]) # True
    """

    units: dict[str, ifcopenshell.entity_instance] = {}
    history_size: int = 64
    history: list[Transaction]
    """Chronological order - from oldest to newest."""
    future: list[Transaction]
    """Reversed chronological order - from newest to oldest."""

    to_delete: Union[set[ifcopenshell.entity_instance], None] = None
    """Entities for batch removal."""

    registry = {}

    def post_init(self, iden=None):
        if iden is None:
            iden = int(self.this)
        if state := self.registry.get(iden):
            self.state = state
        else:
            self.state = self.registry[iden] = [[], [], None]

    @property
    def history(self):
        return self.state[0]

    @property
    def future(self):
        return self.state[1]

    @property
    def transaction(self):
        return self.state[2]

    @history.setter
    def history(self, v):
        self.state[0] = v

    @future.setter
    def future(self, v):
        self.state[1] = v

    @transaction.setter
    def transaction(self, v):
        self.state[2] = v

    def set_history_size(self, size: int) -> None:
        self.history_size = size
        while len(self.history) > self.history_size:
            self.history.pop(0)

    def begin_transaction(self) -> None:
        if self.history_size:
            self.transaction = Transaction(self)

    def end_transaction(self) -> None:
        if self.transaction:
            self.history.append(self.transaction)
            if len(self.history) > self.history_size:
                self.history.pop(0)
            self.future = []
            self.transaction = None

    def discard_transaction(self) -> None:
        if self.transaction:
            self.transaction.rollback()
        self.transaction = None

    def undo(self) -> None:
        if not self.history:
            return
        transaction = self.history.pop()
        try:
            transaction.rollback()
        except Exception as e:
            raise UndoSystemError("Error during transaction undo.", transaction) from e
        self.future.append(transaction)

    def redo(self) -> None:
        if not self.future:
            return
        transaction = self.future.pop()
        try:
            transaction.commit()
        except Exception as e:
            raise UndoSystemError("Error during transaction redo.", transaction) from e
        self.history.append(transaction)

    def create_entity(self, type: str, *args: Any, **kwargs: Any) -> ifcopenshell.entity_instance:
        """Create a new IFC entity in the file.

        You can also use dynamic methods similar to `ifc_file.createIfcWall(...)`
        to create IFC entities. They work exactly the same as if you would do
        `ifc_file.create_entity("IfcWall", ...)` but the resulting typing
        is not as accurate as for `create_entity` due to a dynamic nature
        of those methods.

        :param type: Case insensitive name of the IFC class
        :param args: The positional arguments of the IFC class
        :param kwargs: The keyword arguments of the IFC class
        :returns: An entity instance

        Example:

        .. code:: python

            f = ifcopenshell.file()
            f.create_entity("IfcPerson")
            # >>> #1=IfcPerson($,$,$,$,$,$,$,$)
            f.create_entity("IfcPerson", "Foobar")
            # >>> #2=IfcPerson('Foobar',$,$,$,$,$,$,$)
            f.create_entity("IfcPerson", Identification="Foobar")
            # >>> #3=IfcPerson('Foobar',$,$,$,$,$,$,$)
        """
        eid = kwargs.pop("id", -1)

        e = self.create(type, eid)

        # Create pairs of {attribute index, attribute value}.
        # Keyword arguments are mapped to their corresponding
        # numeric index with get_argument_index().

        # @todo we should probably check that values for
        # attributes are not passed as duplicates using
        # both regular arguments and keyword arguments.
        kwargs_attrs = [(e.get_argument_index(name), arg) for name, arg in kwargs.items()]
        attrs = list(enumerate(args)) + kwargs_attrs

        if len(attrs) > len(e):
            raise ValueError(
                "entity instance of type '%s' has only %s attributes but %s attributes were provided."
                % (e.is_a(True), len(e), len(attrs))
            )

        # Don't store these attributes as transactions
        # as the creation it self is already stored with
        # it's arguments
        transaction = None
        if attrs:
            transaction = self.transaction
            self.transaction = None

        try:
            for idx, arg in attrs:
                e[idx] = arg
        except IndexError:
            invalid_attrs = []
            for (attr_index, _), attr_name in zip(kwargs_attrs, kwargs):
                if attr_index == 0xFFFFFFFF:
                    invalid_attrs.append(attr_name)
            raise ValueError(
                "entity instance of type '%s' doesn't have the following attributes: %s."
                % (e.is_a(True), ", ".join(invalid_attrs))
            )

        # Restore transaction status
        if attrs:
            self.transaction = transaction

        if self.transaction:
            self.transaction.store_create(e)

        return e

    @property
    def schema(self) -> ifcopenshell.util.schema.IFC_SCHEMA:
        """General IFC schema version: IFC2X3, IFC4, IFC4X3."""
        prefixes = ("IFC", "X", "_ADD", "_TC")
        reg = "".join(f"(?P<{s}>{s}\\d+)?" for s in prefixes)
        match = re.match(reg, self.schema_identifier)
        version_tuple = tuple(
            map(
                lambda pp: int(pp[1][len(pp[0]) :]) if pp[1] else None,
                ((p, match.group(p)) for p in prefixes),
            )
        )
        return "".join("".join(map(str, t)) if t[1] else "" for t in zip(prefixes, version_tuple[0:2]))

    @property
    def schema_version(self) -> tuple[int, int, int, int]:
        """Numeric representation of the full IFC schema version.

        E.g. IFC4X3_ADD2 is represented as (4, 3, 2, 0).
        """
        schema = self.schema_identifier
        version = []
        for prefix in ("IFC", "X", "_ADD", "_TC"):
            number = re.search(prefix + r"(\d)", schema)
            version.append(int(number.group(1)) if number else 0)
        return tuple(version)

    @property
    def mvd(self):
        if not LARK_AVAILABLE:
            return None
        return MvdInfo(self.header)

    def __getattr__(self, attr) -> Union[Any, Callable[..., ifcopenshell.entity_instance]]:
        if attr[0:6] == "create":
            return functools.partial(self.create_entity, attr[6:])
        else:
            raise AttributeError(f"'file' object has no attribute '{attr}'.")

    def __getitem__(self, key: Union[numbers.Integral, str, bytes]) -> ifcopenshell.entity_instance:
        if isinstance(key, numbers.Integral):
            return self.by_id(key)
        elif isinstance(key, (str, bytes)):
            return self.by_guid(str(key))
        else:
            raise TypeError("Indexing into file requires either an integral number or compressed guid string")

    def add(self, inst: ifcopenshell.entity_instance, _id: int = None) -> ifcopenshell.entity_instance:
        """Adds an entity including any dependent entities to an IFC file.
        If the entity already exists, it is not re-added. Existence of entity is checked by it's `.identity()`.

        :param inst: The entity instance to add
        :returns: An ifcopenshell.entity_instance
        """

        max_id = None
        if self.transaction:
            max_id = self.get_max_id()

        result = self._add(inst, -1 if _id is None else _id)

        if self.transaction:
            assert max_id is not None
            added_elements = [e for e in self.traverse(result) if e.id() > max_id]
            [self.transaction.store_create(e) for e in reversed(added_elements)]
        return result

    def by_type(self, type: str, include_subtypes=True) -> list[ifcopenshell.entity_instance]:
        """Return IFC objects filtered by IFC Type and wrapped with the entity_instance class.

        If an IFC type class has subclasses, all entities of those subclasses are also returned.

        :param type: The case insensitive type of IFC class to return.
        :param include_subtypes: Whether or not to return subtypes of the IFC class

        :raises RuntimeError: If `type` is not found in IFC schema.

        :returns: A list of ifcopenshell.entity_instance objects
        """
        if include_subtypes:
            return self._by_type(type)
        return self._by_type_excl_subtypes(type)

    def traverse(
        self, inst: ifcopenshell.entity_instance, max_levels: Optional[int] = None, breadth_first: bool = False
    ) -> list[ifcopenshell.entity_instance]:
        """Get a list of all referenced instances for a particular instance including itself

        :param inst: The entity instance to get all sub instances
        :param max_levels: How far deep to recursively fetch sub instances. None or -1 means infinite.
        :param breadth_first: Whether to use breadth-first search, the default is depth-first.
        :returns: A list of ifcopenshell.entity_instance objects
        """
        if max_levels is None:
            max_levels = -1

        if breadth_first:
            fn = self._traverse_breadth_first
        else:
            fn = self._traverse

        return fn(inst, max_levels)

    @overload
    def get_inverse(
        self,
        inst: ifcopenshell.entity_instance,
        allow_duplicate: Literal[False] = False,
        with_attribute_indices: bool = False,
    ) -> set[ifcopenshell.entity_instance]: ...
    @overload
    def get_inverse(
        self,
        inst: ifcopenshell.entity_instance,
        allow_duplicate: Literal[True],
        with_attribute_indices: Literal[False] = False,
    ) -> list[ifcopenshell.entity_instance]: ...
    @overload
    def get_inverse(
        self,
        inst: ifcopenshell.entity_instance,
        allow_duplicate: Literal[True],
        with_attribute_indices: Literal[True],
    ) -> list[tuple[ifcopenshell.entity_instance, int]]: ...
    @overload
    def get_inverse(
        self,
        inst: ifcopenshell.entity_instance,
        allow_duplicate: bool,
        with_attribute_indices: bool = False,
    ) -> Union[list[ifcopenshell.entity_instance], set[ifcopenshell.entity_instance]]: ...
    def get_inverse(
        self,
        inst: ifcopenshell.entity_instance,
        allow_duplicate: bool = False,
        with_attribute_indices: bool = False,
    ) -> (
        list[ifcopenshell.entity_instance]
        | set[ifcopenshell.entity_instance]
        | list[tuple[ifcopenshell.entity_instance, int]]
    ):
        """Return a list of entities that reference this entity

        Warning: this is a slow function, especially when there is a large
        number of inverses (such as for a shared owner history). If you are
        only interested in the total number of inverses (typically 0, 1, or N),
        consider using :func:`get_total_inverses`.

        :param inst: The entity instance to get inverse relationships
        :param allow_duplicate: Returns a `list` when True, `set` when False
        :param with_attribute_indices: Returns pairs of <i, idx>
           where i[idx] is inst or contains inst. Requires allow_duplicate=True
        :returns: A list or set of ifcopenshell.entity_instance objects.
        """
        if with_attribute_indices and not allow_duplicate:
            raise ValueError("with_attribute_indices requires allow_duplicate to be True")

        inverses = self._get_inverse(inst)

        if allow_duplicate:
            if with_attribute_indices:
                idxs = self._get_inverse_indices(inst)
                # TODO: include in typing.
                return list(zip(inverses, idxs))
            else:
                return inverses

        return set(inverses)

    def remove(self, inst: ifcopenshell.entity_instance) -> None:
        """Deletes an IFC object in the file.

        Attribute values in other entity instances that reference the deleted
        object will be set to null. In the case of a list or set of references,
        the reference to the deleted will be removed from the aggregate.

        :param inst: The entity instance to delete
        """
        if self.transaction:
            self.transaction.store_delete(inst)
        return self._remove(inst)

    def batch(self):
        """Low-level mechanism to speed up deletion of large subgraphs"""
        if self.transaction:
            self.transaction.batch()
        return self.batch()

    def unbatch(self):
        """Low-level mechanism to speed up deletion of large subgraphs"""
        if self.transaction:
            self.transaction.unbatch()
        return self.unbatch()

    def __iter__(self) -> Generator[ifcopenshell.entity_instance, None, None]:
        return iter(self[id] for id in self.entity_names())

    def assign_header_from(self, other: ifcopenshell.file) -> None:
        for k, vs in HEADER_FIELDS.items():
            for v in vs:
                try:
                    setattr(getattr(self.header, k), v, getattr(getattr(other.header, k), v))
                except:
                    pass  # Header is invalid

    def write(self, path: os.PathLike | str, format: Optional[str] = None, zipped: bool = False) -> None:
        """Write ifc model to file.

        :param format: Force use of a specific format. Guessed from file name
            if None.  Supported formats : .ifc, .ifcZIP (equivalent to
            format=".ifc" with zipped=True)
        :param zipped: zip the file after it is written

        Example:

        .. code:: python

            model.write("path/to/model.ifc")
            model.write("path/to/model.ifcZIP")
            model.write("path/to/model.anyextension", format=".ifc")
        """
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)

        if format == None:
            format = ifcopenshell.guess_format(path)
        if format == ".ifcXML":
            raise NotImplementedError("Writing .ifcXML files is not supported")
        if format == ".ifcZIP":
            return self.write(path, ".ifc", zipped=True)
        self._write(str(path))

        if zipped:
            unzipped_path = path.with_suffix(format)
            path.rename(unzipped_path)
            with zipfile.ZipFile(path, "w") as zip_file:
                zip_file.write(
                    unzipped_path,
                    unzipped_path.name,
                    compress_type=zipfile.ZIP_DEFLATED,
                )
                unzipped_path.unlink()
        return

    def to_string(self) -> str:
        return self.to_string()

    @staticmethod
    def _determine_schema_identifier(
        schema: Optional[ifcopenshell.util.schema.IFC_SCHEMA] = None,
        schema_version: Optional[tuple[int, int, int, int]] = None,
    ) -> str:
        if schema_version:
            prefixes = ("IFC", "X", "_ADD", "_TC")
            schema = "".join("".join(map(str, t)) if t[1] else "" for t in zip(prefixes, schema_version))
        elif schema:
            schema = {"IFC4X3": "IFC4X3_ADD2"}.get(schema, schema)
        else:
            schema = "IFC4"
        return schema

    @property
    def storage(self) -> Optional[rocksdb_file_storage]:
        """
        Returns:
            Optional[rocksdb_file_storage]: underlying key-value store interface when opened as a RocksDB-backed file
        """
        if self.storage_mode() == 1:
            return rocksdb_file_storage(self)