From 4e3d6f177397d80e2f6a194681d367abbd0cbd9d Mon Sep 17 00:00:00 2001 From: Andrej730 Date: Tue, 24 Sep 2024 10:44:10 +0500 Subject: [PATCH] typing --- src/bonsai/bonsai/tool/loader.py | 4 ++- .../ifcopenshell/__init__.py | 19 +++++++++-- .../ifcopenshell/stream.py | 21 ++++++------ src/ifcpatch/ifcpatch/recipes/Ifc2Sql.py | 32 ++++++++++--------- 4 files changed, 49 insertions(+), 27 deletions(-) diff --git a/src/bonsai/bonsai/tool/loader.py b/src/bonsai/bonsai/tool/loader.py index ed7a4a1a38..6f7849fcc6 100644 --- a/src/bonsai/bonsai/tool/loader.py +++ b/src/bonsai/bonsai/tool/loader.py @@ -880,7 +880,9 @@ class Loader(bonsai.core.tool.Loader): return Matrix(matrix.tolist()) @classmethod - def convert_geometry_to_mesh(cls, geometry, mesh: bpy.types.Mesh, verts=None) -> bpy.types.Mesh: + def convert_geometry_to_mesh( + cls, geometry: ifcopenshell.geom.ShapeType, mesh: bpy.types.Mesh, verts=None + ) -> bpy.types.Mesh: if verts is None: verts = geometry.verts if geometry.faces: diff --git a/src/ifcopenshell-python/ifcopenshell/__init__.py b/src/ifcopenshell-python/ifcopenshell/__init__.py index d158e7c94d..99b5b737cd 100644 --- a/src/ifcopenshell-python/ifcopenshell/__init__.py +++ b/src/ifcopenshell-python/ifcopenshell/__init__.py @@ -59,7 +59,7 @@ import sys import zipfile import tempfile from pathlib import Path -from typing import Optional, Union, TYPE_CHECKING, Any +from typing import Optional, Union, TYPE_CHECKING, Any, overload, Literal if TYPE_CHECKING: import ifcopenshell.express.schema_class @@ -119,9 +119,24 @@ class SchemaError(Error): pass -def open(path: Union[os.PathLike, str], format: Optional[str] = None, should_stream: bool = False) -> file: +@overload +def open( + path: Union[os.PathLike, str], format: Optional[str] = None, *, should_stream: Literal[False] = False +) -> Union[file, sqlite]: ... +@overload +def open(path: Union[os.PathLike, str], format: Optional[str] = None, *, should_stream: Literal[True]) -> stream: ... +@overload +def open( + path: Union[os.PathLike, str], format: Optional[str] = None, *, should_stream: bool +) -> Union[file, sqlite, stream]: ... +def open( + path: Union[os.PathLike, str], format: Optional[str] = None, should_stream: bool = False +) -> Union[file, sqlite, stream]: """Loads an IFC dataset from a filepath + :param should_stream: Whether to open the file in streaming mode. Could be useful + for reading large files. + You can specify a file format. If no format is given, it is guessed from its extension. Currently supported specified format: .ifc | .ifcZIP | .ifcXML. diff --git a/src/ifcopenshell-python/ifcopenshell/stream.py b/src/ifcopenshell-python/ifcopenshell/stream.py index f6598624f8..46f593ad50 100644 --- a/src/ifcopenshell-python/ifcopenshell/stream.py +++ b/src/ifcopenshell-python/ifcopenshell/stream.py @@ -19,6 +19,7 @@ from __future__ import annotations try: + import os import re import ifcopenshell.util.attribute @@ -31,7 +32,7 @@ try: from typing import Any, NoReturn, Union, Optional class StreamTransformer(Transformer): - file: "file" + file: file def string(self, items): return str(items[0])[1:-1] @@ -98,12 +99,12 @@ try: self.filepath = filepath self.file = open(filepath, "r") - self.id_map = {} - self.class_map = {} - self.id_offset = {} + self.id_map: dict[int, str] = {} + self.class_map: dict[str, list[int]] = {} + self.id_offset: dict[int, int] = {} self.reference_pattern = re.compile(r"#(\d+)") self.entity_cache: dict[int, stream_entity] = {} - self.inverses = {} + self.inverses: dict[int, list[int]] = {} # common.INT doesn't support negative integers. grammar = r""" @@ -175,7 +176,7 @@ try: self.id_offset[step_id] = offset elif line.startswith("FILE_SCHEMA"): self.schema = line.split("'")[1] - self.ifc_schema = ifcopenshell.ifcopenshell_wrapper.schema_by_name(self.schema) + self.ifc_schema = ifcopenshell.schema_by_name(self.schema) for ifc_class in exclude_classes: declaration = self.ifc_schema.declaration_by_name(ifc_class) exclude.update([st.name().upper() for st in ifcopenshell.util.schema.get_subtypes(declaration)]) @@ -186,7 +187,7 @@ try: def preprocess_schema(self) -> None: self.ifc_class_names = {} self.ifc_class_subtypes = {} - self.ifc_class_attributes = {} + self.ifc_class_attributes: dict[str, dict[str, ifcopenshell_wrapper.attribute]] = {} self.ifc_class_inverse_attributes = {} self.ifc_class_references = {} self.ifc_class_inverses = {} @@ -289,6 +290,8 @@ try: pass class stream_entity(entity_instance): + stream_wrapper: stream_wrapper + def __init__(self, id: int, ifc_class: str, file: stream = None): if not ifc_class: print(id, ifc_class, file) @@ -401,11 +404,11 @@ try: self.file = file self.attributes = self.file.ifc_class_attributes[self.ifc_class] self.inverse_attributes = self.file.ifc_class_inverse_attributes[self.ifc_class] - self.attribute_cache = {} + self.attribute_cache: dict[str, Any] = {} self.inverse_attribute_cache = {} def __repr__(self) -> str: - return "todo" + return f"stream_wrapper '#{self.id}={self.ifc_class}(...)'" except ImportError as e: import sys diff --git a/src/ifcpatch/ifcpatch/recipes/Ifc2Sql.py b/src/ifcpatch/ifcpatch/recipes/Ifc2Sql.py index b3e18714d0..1b27ba3433 100644 --- a/src/ifcpatch/ifcpatch/recipes/Ifc2Sql.py +++ b/src/ifcpatch/ifcpatch/recipes/Ifc2Sql.py @@ -28,11 +28,13 @@ import numpy as np import multiprocessing import ifcopenshell import ifcopenshell.geom -import ifcopenshell.util.unit -import ifcopenshell.util.shape -import ifcopenshell.util.schema import ifcopenshell.util.attribute +import ifcopenshell.util.element import ifcopenshell.util.placement +import ifcopenshell.util.schema +import ifcopenshell.util.shape +import ifcopenshell.util.unit +from typing import Any SQLTypes = typing.Literal["SQLite", "MySQL"] @@ -103,7 +105,7 @@ class Patcher: self.password = password self.database = database - def patch(self): + def patch(self) -> None: self.full_schema = True # Set true for ifcopenshell.sqlite self.is_strict = False self.should_expand = False # Set false for ifcopenshell.sqlite @@ -173,7 +175,7 @@ class Patcher: self.db.commit() self.db.close() - def create_geometry(self): + def create_geometry(self) -> None: self.unit_scale = ifcopenshell.util.unit.calculate_unit_scale(self.file) self.shape_rows = {} @@ -239,7 +241,7 @@ class Patcher: break print("Done creating geometry") - def create_id_map(self): + def create_id_map(self) -> None: if self.sql_type == "sqlite": statement = ( "CREATE TABLE IF NOT EXISTS id_map (ifc_id integer PRIMARY KEY NOT NULL UNIQUE, ifc_class text);" @@ -254,7 +256,7 @@ class Patcher: """ self.c.execute(statement) - def create_metadata(self): + def create_metadata(self) -> None: # There is no "standard" SQL serialisation, so we propose a convention # of a "metadata" table to hold high level metadata. This includes the # preprocessor field to uniquely identify the "variant" of SQL schema @@ -278,7 +280,7 @@ class Patcher: self.c.execute(statement) self.c.execute("INSERT INTO metadata VALUES (%s, %s, %s);", metadata) - def create_pset_table(self): + def create_pset_table(self) -> None: statement = """ CREATE TABLE IF NOT EXISTS psets ( ifc_id integer NOT NULL, @@ -289,7 +291,7 @@ class Patcher: """ self.c.execute(statement) - def create_geometry_table(self): + def create_geometry_table(self) -> None: statement = """ CREATE TABLE IF NOT EXISTS shape ( ifc_id integer NOT NULL, @@ -319,7 +321,7 @@ class Patcher: self.c.execute(statement) - def create_sqlite_table(self, ifc_class, declaration): + def create_sqlite_table(self, ifc_class: str, declaration: ifcopenshell.ifcopenshell_wrapper.declaration) -> None: statement = f"CREATE TABLE IF NOT EXISTS {ifc_class} (" if self.should_expand: @@ -360,7 +362,7 @@ class Patcher: print(statement) self.c.execute(statement) - def create_mysql_table(self, ifc_class, declaration): + def create_mysql_table(self, ifc_class: str, declaration: ifcopenshell.ifcopenshell_wrapper.declaration) -> None: declaration = self.schema.declaration_by_name(ifc_class) statement = f"CREATE TABLE IF NOT EXISTS {ifc_class} (" statement += "`ifc_id` int(10) unsigned NOT NULL," @@ -404,7 +406,7 @@ class Patcher: print(statement) self.c.execute(statement) - def insert_data(self, ifc_class): + def insert_data(self, ifc_class: str) -> None: print("Extracting data for", ifc_class) elements = self.file.by_type(ifc_class, include_subtypes=False) @@ -478,14 +480,14 @@ class Patcher: if pset_rows: self.c.executemany("INSERT INTO psets VALUES (%s, %s, %s, %s);", pset_rows) - def serialise_value(self, element, value): + def serialise_value(self, element: ifcopenshell.entity_instance, value: Any) -> Any: return element.walk( lambda v: isinstance(v, ifcopenshell.entity_instance), lambda v: v.id() if v.id() else {"type": v.is_a(), "value": v.wrappedValue}, value, ) - def get_permutations(self, lst, indexes): + def get_permutations(self, lst: list[Any], indexes: list[int]) -> list[Any]: nested_lists = [lst[i] for i in indexes] # Generate the Cartesian product of the nested lists @@ -501,7 +503,7 @@ class Patcher: return final_lists - def is_entity_list(self, attribute): + def is_entity_list(self, attribute: ifcopenshell.ifcopenshell_wrapper.attribute) -> bool: attribute = str(attribute.type_of_attribute()) if (attribute.startswith("", attribute):