This commit is contained in:
Andrej730
2024-05-31 18:02:20 +05:00
parent c05abc4d5b
commit 162e66b775
5 changed files with 47 additions and 27 deletions
+17 -17
View File
@@ -24,7 +24,7 @@ import zipfile
import functools
import ifcopenshell
from pathlib import Path
from typing import Optional, Any, Union, Callable
from typing import Optional, Any, Union, Callable, Generator
from . import ifcopenshell_wrapper
from .entity_instance import entity_instance
@@ -59,13 +59,13 @@ class Transaction:
value,
)
def batch(self):
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):
def unbatch(self) -> None:
for inverses in self.batch_inverses:
if inverses:
self.operations.insert(self.batch_delete_index, {"action": "batch_delete", "inverses": inverses})
@@ -74,11 +74,11 @@ class Transaction:
self.batch_delete_ids = set()
self.batch_inverses = []
def store_create(self, element):
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, index, value):
def store_edit(self, element: ifcopenshell.entity_instance, index: int, value: Any) -> None:
if element.id():
self.operations.append(
{
@@ -120,7 +120,7 @@ class Transaction:
return False
return value == element
def rollback(self):
def rollback(self) -> None:
for operation in self.operations[::-1]:
if operation["action"] == "create":
element = self.file.by_id(operation["value"]["id"])
@@ -153,7 +153,7 @@ class Transaction:
for index, value in data:
inverse[index] = self.unserialise_value(inverse, value)
def commit(self):
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"])
@@ -260,19 +260,19 @@ class file:
file_dict[self.file_pointer()] = weakref.ref(self)
def __del__(self):
def __del__(self) -> None:
del file_dict[self.file_pointer()]
def set_history_size(self, size):
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):
def begin_transaction(self) -> None:
if self.history_size:
self.transaction = Transaction(self)
def end_transaction(self):
def end_transaction(self) -> None:
if self.transaction:
self.history.append(self.transaction)
if len(self.history) > self.history_size:
@@ -280,19 +280,19 @@ class file:
self.future = []
self.transaction = None
def discard_transaction(self):
def discard_transaction(self) -> None:
if self.transaction:
self.transaction.rollback()
self.transaction = None
def undo(self):
def undo(self) -> None:
if not self.history:
return
transaction = self.history.pop()
transaction.rollback()
self.future.append(transaction)
def redo(self):
def redo(self) -> None:
if not self.future:
return
transaction = self.future.pop()
@@ -405,7 +405,7 @@ class file:
else:
return getattr(self.wrapped_data, attr)
def __getitem__(self, key):
def __getitem__(self, key: Union[numbers.Integral, str, bytes]) -> entity_instance:
if isinstance(key, numbers.Integral):
return entity_instance(self.wrapped_data.by_id(key), self)
elif isinstance(key, (str, bytes)):
@@ -501,7 +501,7 @@ class file:
return [entity_instance(e, self) for e in fn(inst.wrapped_data, max_levels)]
def get_inverse(
self, inst: ifcopenshell.entity_instance, allow_duplicate=False, with_attribute_indices=False
self, inst: ifcopenshell.entity_instance, allow_duplicate: bool = False, with_attribute_indices: bool = False
) -> list[ifcopenshell.entity_instance]:
"""Return a list of entities that reference this entity
@@ -572,7 +572,7 @@ class file:
self.transaction.unbatch()
return self.wrapped_data.unbatch()
def __iter__(self):
def __iter__(self) -> Generator[ifcopenshell.entity_instance, None, None]:
return iter(self[id] for id in self.wrapped_data.entity_names())
def write(self, path: "os.PathLike | str", format: Optional[str] = None, zipped: bool = False) -> None:
@@ -22,9 +22,9 @@ import time
import ifcopenshell
import ifcopenshell.util.attribute
import ifcopenshell.ifcopenshell_wrapper as ifcopenshell_wrapper
from typing import Union, Any
# This is highly experimental and incomplete, however, it may work for simple datasets.
# In this simple implementation, we only support 2X3<->4 right now
cwd = os.path.dirname(os.path.realpath(__file__))
@@ -223,7 +223,9 @@ class Migrator:
"User": None,
}
def migrate(self, element: ifcopenshell.entity_instance, new_file: ifcopenshell.file) -> ifcopenshell.entity_instance:
def migrate(
self, element: ifcopenshell.entity_instance, new_file: ifcopenshell.file
) -> ifcopenshell.entity_instance:
if element.id() == 0:
return new_file.create_entity(element.is_a(), element.wrappedValue)
try:
@@ -241,7 +243,9 @@ class Migrator:
self.migrated_ids[element.id()] = new_element.id()
return new_element
def migrate_class(self, element, new_file):
def migrate_class(
self, element: ifcopenshell.entity_instance, new_file: ifcopenshell.file
) -> ifcopenshell.entity_instance:
try:
new_element = new_file.create_entity(element.is_a())
except:
@@ -260,7 +264,14 @@ class Migrator:
self.migrate_attribute(attribute, element, new_file, new_element, new_element_schema)
return new_element
def find_equivalent_attribute(self, new_element, attribute, element, attributes_mapping, reverse_mapping=False):
def find_equivalent_attribute(
self,
new_element: ifcopenshell.entity_instance,
attribute: ifcopenshell_wrapper.attribute,
element: ifcopenshell.entity_instance,
attributes_mapping: dict[str, dict[str, str]],
reverse_mapping: bool = False,
) -> Union[Any, None]:
# print("Searching for an equivalent", element, new_element, attribute.name())
try:
if reverse_mapping:
@@ -281,7 +292,14 @@ class Migrator:
)
raise e
def migrate_attribute(self, attribute, element, new_file: ifcopenshell.file, new_element, new_element_schema):
def migrate_attribute(
self,
attribute: ifcopenshell_wrapper.attribute,
element: ifcopenshell.entity_instance,
new_file: ifcopenshell.file,
new_element: ifcopenshell.entity_instance,
new_element_schema: ifcopenshell_wrapper.declaration,
) -> None:
# NOTE: `attribute` is an attribute in new file schema
# print("Migrating attribute", element, new_element, attribute.name())
old_file = element.wrapped_data.file
@@ -349,7 +367,7 @@ class Migrator:
if value is not None:
setattr(new_element, attribute.name(), value)
def generate_default_value(self, attribute, new_file):
def generate_default_value(self, attribute: ifcopenshell_wrapper.attribute, new_file: ifcopenshell.file) -> Any:
if attribute.name() in self.default_values:
return self.default_values[attribute.name()]
elif attribute.name() == "OwnerHistory":