ifcopenshell.file.Transaction - add static typing

This commit is contained in:
Andrej730
2025-05-08 12:36:44 +05:00
parent ec13294430
commit 5a7a368169
+44 -6
View File
@@ -32,7 +32,8 @@ from typing import TYPE_CHECKING
from typing import Union
from typing import overload
from typing import Literal
from typing import Any
from typing import TypedDict
from typing_extensions import assert_never
from . import ifcopenshell_wrapper
from .entity_instance import entity_instance
@@ -42,6 +43,31 @@ from ifcopenshell.util.mvd_info import MvdInfo, LARK_AVAILABLE
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",
@@ -66,6 +92,10 @@ class UndoSystemError(Exception):
class Transaction:
operations: list[TransactionOperation]
batch_inverses: list[ElementInverses]
batch_delete_ids: set[int]
def __init__(self, ifc_file: file):
self.file: file = ifc_file
self.operations = []
@@ -80,14 +110,14 @@ class Transaction:
info[key] = self.serialise_value(element, value)
return info
def serialise_value(self, element, value):
def serialise_value(self, element, value) -> Any:
return element.walk(
lambda v: isinstance(v, 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):
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"]),
@@ -137,10 +167,10 @@ class Transaction:
{"action": "delete", "inverses": inverses, "value": self.serialise_entity_instance(element)}
)
def get_element_inverses(self, element):
inverses = {}
def get_element_inverses(self, element: ifcopenshell.entity_instance) -> ElementInverses:
inverses: ElementInverses = {}
for inverse in self.file.get_inverse(element):
inverse_references = []
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)))
@@ -187,6 +217,8 @@ class Transaction:
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:
@@ -206,6 +238,8 @@ class Transaction:
self.file.remove(element)
elif operation["action"] == "batch_delete":
pass
else:
assert_never(operation["action"])
file_dict = {}
@@ -235,6 +269,10 @@ class file:
wrapped_data: ifcopenshell_wrapper.file
units: dict[str, 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."""