+import inspect
+import logging
+import os
+import pickle
+import platform
+from pprint import pprint
+from typing import Union
+
+import matplotlib.pyplot as plt
+import networkx as nx
+import numpy as np
+
+# Import several signals from Configuration.
+from amworkflow.config.settings import (
+ CLEAN_UP,
+ ENABLE_CONCURRENT_MODE,
+ ENABLE_OCC,
+ ENABLE_SHELF,
+ ENABLE_SQL_DATABASE,
+ LOG_LEVEL,
+ STORAGE_PATH,
+)
+
+STATUS = (
+ "memory"
+ if (not ENABLE_SQL_DATABASE and not ENABLE_SHELF)
+ else ("database" if ENABLE_SQL_DATABASE else "shelf")
+)
+
+# Import multiprocessing and initialize the start method to 'fork' if concurrent mode is enabled. this also means the code will only be suitable for Unix-like systems.
+if ENABLE_CONCURRENT_MODE:
+ import multiprocessing
+
+ if multiprocessing.get_start_method(True) != "fork":
+ multiprocessing.set_start_method("fork")
+ logging.debug("%s starting.", multiprocessing.current_process().name)
+
+# Setting up the logging configuration.
+logging.basicConfig(
+ level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("amworkflow.geometry.builtinCAD")
+logger.setLevel(LOG_LEVEL)
+
+# Import OCC if enabled.
+if ENABLE_OCC:
+ try:
+ from OCC.Core.gp import gp_Pnt
+
+ import amworkflow.occ_helpers as occh
+ except ImportError as e:
+ print("OCC module couldn't be imported:", e)
+ ENABLE_OCC = False # Disable OCC if import fails
+
+else:
+ logging.warning("OCC is disabled. Try running without it now.")
+
+# Initialize some containers to store the objects and their properties.
+count_id = 0
+count_gid = [0 for i in range(7)]
+component_lib = {}
+component_container = {}
+TYPE_INDEX = {
+ 0: "point",
+ 1: "segment",
+ 2: "wire",
+ 3: "surface",
+ 4: "shell",
+ 5: "solid",
+ 6: "compound",
+}
+
+# Create a database to store the objects. Though python data structures can be used to store the objects, a database is used to store the objects for the support of concurrent access.
+if ENABLE_SQL_DATABASE:
+ logger.debug("SQL Database is enabled.")
+ import sqlite3
+
+ db_name = "bcad_sql"
+ db_path = os.path.join(STORAGE_PATH, f"{db_name}.db")
+ if os.path.exists(db_path) and CLEAN_UP:
+ # remove the existing database file
+ os.remove(db_path)
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ # Create the modified 'objects' table to store only Oid references
+ c.execute(
+ """CREATE TABLE IF NOT EXISTS objects
+ (id INTEGER PRIMARY KEY AUTOINCREMENT,
+ object_type TEXT,
+ gid INTEGER,
+ property BLOB)"""
+ )
+
+ # Create tables for each type of object
+ for key, value in TYPE_INDEX.items():
+ c.execute(
+ f"""CREATE TABLE IF NOT EXISTS {value}
+ (gid INTEGER PRIMARY KEY AUTOINCREMENT,
+ id INTEGER,
+ value BLOB,
+ obj BLOB,
+ FOREIGN KEY (id) REFERENCES objects(id) ON DELETE CASCADE)"""
+ )
+ logger.debug("Database created successfully.")
+ conn.commit()
+ conn.close()
+
+ def store_obj_to_db(obj_type: str, obj: any, obj_property: dict) -> tuple:
+ """
+ Insert an object to the database and return its id.
+ :param obj_type: The type of the object to be inserted.
+ :type obj_type: str
+ :param obj: The object to be inserted.
+ :type obj: any
+ :return: The id, gid of the object.
+ :rtype: tuple
+ """
+ # oid, gid = get_last_id_from_db(table=obj_type, gid=True)
+ # oid += 1
+ # gid += 1
+ # obj.id = oid
+ # obj.gid = gid
+ # obj.update_basic_property()
+ if obj_type not in TYPE_INDEX.values():
+ raise ValueError(f"Unrecognized object type: {obj_type}.")
+ local_conn = sqlite3.connect(db_path)
+ local_c = local_conn.cursor()
+
+ local_value = obj.value
+ local_c.execute(
+ f"INSERT INTO {obj_type} (obj, value) VALUES (?,?)",
+ (pickle.dumps(obj), pickle.dumps(local_value)),
+ )
+ local_conn.commit()
+ gid = local_c.lastrowid
+ local_c.execute(
+ """
+ INSERT INTO objects (object_type, gid, property)
+ VALUES (?, ?, ?)
+ """,
+ (
+ obj_type,
+ gid,
+ pickle.dumps(obj_property),
+ ),
+ )
+ local_conn.commit()
+ oid = local_c.lastrowid
+ local_c.execute(f"UPDATE {obj_type} SET id = ? WHERE gid = ?", (oid, gid))
+ local_conn.commit()
+ local_conn.close()
+ return oid, gid
+
+ def get_last_id_from_db(table: str = "", gid: bool = False) -> int:
+ """
+ Get the last id from the database.
+ :return: The last id from the database.
+ :rtype: int
+ """
+ local_conn = sqlite3.connect(db_path)
+ local_c = local_conn.cursor()
+ local_c.execute("SELECT MAX(id) FROM objects")
+ last_id = local_c.fetchone()[0]
+ if last_id is None:
+ last_id = 0
+ if gid:
+ local_c.execute(f"SELECT MAX(gid) FROM {table}")
+ last_gid = local_c.fetchone()[0]
+ local_conn.close()
+ if last_gid is None:
+ last_gid = 0
+ return last_id, last_gid
+ local_conn.close()
+ return last_id
+
+ def get_obj_from_db(
+ oid: int = None, it_value: list = None, obj_type: str = None
+ ) -> any:
+ """
+ Get an object from the database by its id.
+ :param oid: The id of the object to be retrieved.
+ :type oid: int
+ :return: The object retrieved.
+ :rtype: any
+ """
+ local_conn = sqlite3.connect(db_path)
+ local_c = local_conn.cursor()
+ if oid is not None:
+ local_c.execute("SELECT object_type FROM objects WHERE id = ?", (oid,))
+ obj_type = local_c.fetchone()[0]
+ local_c.execute(
+ f"""SELECT {obj_type}.obj, {obj_type}.gid
+ FROM objects
+ INNER JOIN {obj_type} ON objects.gid = {obj_type}.gid
+ WHERE objects.id = ?""",
+ (oid,),
+ )
+ row = local_c.fetchone()
+ if row:
+ obj = pickle.loads(row[0])
+ obj.id = oid
+ obj.gid = row[1]
+ obj.update_basic_property()
+ local_conn.close()
+ elif it_value is not None:
+ local_c.execute(
+ f"""SELECT {obj_type}.obj, {obj_type}.gid, {obj_type}.id
+ FROM {obj_type}
+ WHERE {obj_type}.value = ?)""",
+ (pickle.dumps(it_value),),
+ )
+ row = local_c.fetchone()
+ if row:
+ obj = pickle.loads(row[0])
+ obj.id = row[2]
+ obj.gid = row[1]
+ obj.update_basic_property()
+ local_conn.close()
+ else:
+ # local_c.execute(
+ # f"""SELECT {obj_type}.obj
+ # FROM {obj_type}
+ # INNER JOIN objects ON {obj_type}.gid = objects.gid
+ # """
+ # )
+ local_c.execute(
+ f"""SELECT obj, gid, id
+ FROM {obj_type}
+ """
+ )
+ # rows = local_c.fetchall()
+ for row in local_c:
+ obj = pickle.loads(row[0])
+ obj.id = row[2]
+ obj.gid = row[1]
+ obj.update_basic_property()
+ yield obj
+ local_conn.close()
+ return None
+ yield obj
+
+ def get_property_from_db(oid: int = None) -> dict:
+ """
+ Get the property of an object from the database by its id.
+ :param oid: The id of the object to be retrieved.
+ :type oid: int
+ :return: The property of the object retrieved.
+ :rtype: dict
+ """
+ local_conn = sqlite3.connect(db_path)
+ local_c = local_conn.cursor()
+ if int is not None:
+ local_c.execute("SELECT property FROM objects WHERE id = ?", (oid,))
+ it_property = pickle.loads(local_c.fetchone()[0])
+ yield it_property
+ else:
+ local_c.execute("SELECT property FROM objects")
+ for row in local_c:
+ it_property = pickle.loads(row[0])
+ yield it_property
+
+ def update_obj_in_db(
+ obj: any, obj_property: dict = None, oid: int = None, obj_type: str = None
+ ):
+ """
+ Update an object in the database.
+ :param oid: The id of the object to be updated.
+ :type oid: int
+ :param obj: The object to be updated.
+ :type obj: any
+ :param obj_property: The property of the object to be updated.
+ :type obj_property: dict
+ """
+ local_conn = sqlite3.connect(db_path)
+ local_c = local_conn.cursor()
+ if oid is None:
+ oid = obj.id
+ if obj_type is None:
+ obj_type = TYPE_INDEX[obj.type]
+ if obj_property is None:
+ obj_property = obj.property
+ local_c.execute(
+ "UPDATE objects SET property = ? WHERE id = ?",
+ (pickle.dumps(obj_property), oid),
+ )
+ local_c.execute(
+ f"UPDATE {obj_type} SET obj = ? WHERE gid = ?", (pickle.dumps(obj), obj.gid)
+ )
+ local_conn.commit()
+ local_conn.close()
+
+ def delete_obj_from_db(oid: int):
+ """
+ Delete an object from the database.
+ :param oid: The id of the object to be deleted.
+ :type oid: int
+ """
+ local_conn = sqlite3.connect(db_path)
+ local_c = local_conn.cursor()
+ local_c.execute("DELETE FROM objects WHERE id = ?", (oid,))
+ local_conn.commit()
+ local_conn.close()
+
+ def delete_sql_db_file(db_name):
+ # Construct the full filename including the extension
+ db_file = f"{db_name}.db"
+ db_path = os.path.join(STORAGE_PATH, db_file)
+ # Check if the file exists and delete it
+ try:
+ if os.path.exists(db_path):
+ os.remove(db_path)
+ print(f"Deleted database file: {db_file}")
+ else:
+ print(f"Database file {db_file} does not exist and cannot be deleted.")
+ except Exception as ex:
+ print(f"Error deleting database file {db_file}: {ex}")
+
+
+if ENABLE_SHELF:
+ import shelve
+
+ logger.debug("Shelf is enabled.")
+
+ def check_shelve_db_exists(shelve_name):
+ # Determine the current operating system
+ os_name = platform.system()
+ # Define file extensions based on the operating system
+ if os_name == "Darwin": # macOS
+ files_to_check = [f"{shelve_name}.db"]
+ elif os_name == "Linux":
+ files_to_check = [
+ f"{shelve_name}.dat",
+ f"{shelve_name}.dir",
+ ]
+ else:
+ raise NotImplementedError(f"OS {os_name} not supported.")
+
+ # Check if all required files exist
+ return all(
+ os.path.exists(os.path.join(STORAGE_PATH, file)) for file in files_to_check
+ )
+
+ def delete_shelve_db_files(shelve_name):
+ # Determine the current operating system
+ os_name = platform.system()
+ # Start with the common file extensions
+ if os_name == "Darwin": # macOS
+ files_to_delete = [
+ f"{shelve_name}.db"
+ ] # macOS typically uses a single .db file
+ elif os_name == "Linux" or os_name == "Windows":
+ # For Linux and Windows, start with .dat and .dir which are commonly used
+ files_to_delete = [f"{shelve_name}.dat", f"{shelve_name}.dir"]
+ else:
+ raise NotImplementedError(f"OS {os_name} not supported.")
+
+ # Always check for a .bak file regardless of the OS
+ bak_file = f"{shelve_name}.bak"
+ if os.path.exists(os.path.join(STORAGE_PATH, bak_file)):
+ files_to_delete.append(bak_file)
+
+ # Delete the files
+ for file in files_to_delete:
+ try:
+ if os.path.exists(os.path.join(STORAGE_PATH, file)):
+ os.remove(file)
+ print(f"Deleted {file}")
+ else:
+ print(f"{file} does not exist and cannot be deleted.")
+ except Exception as e:
+ print(f"Error deleting {file}: {e}")
+
+ shelve_name = "bcad_shelf" # Without the extension
+ if check_shelve_db_exists(shelve_name):
+ logger.debug("Shelve database exists.")
+ else:
+ logger.debug("Shelve database does not exist. Creating...")
+ shelf_path = os.path.join(STORAGE_PATH, shelve_name)
+ bcad_shelf = shelve.open(shelf_path, "c")
+ logger.debug("Shelf database created successfully.")
+
+ def get_next_id(db, update=False):
+ """Retrieve the next available ID, incrementing the stored value."""
+ if "next_id" not in db:
+ db["next_id"] = 0
+ elif update:
+ db["next_id"] += 1
+ else:
+ pass
+ return db["next_id"]
+
+ def add_entry(db, data):
+ """Add a new entry with an auto-incremented ID."""
+ next_id = get_next_id(db)
+ db[str(next_id)] = data
+ get_next_id(db, update=True)
+ return next_id
+
+ def get_entry(db: shelve.Shelf, oid: int = None):
+ """Get an entry from the shelf. If oid is not specified, return all entries.
+
+ :param db: shelf specified
+ :type db: Shelve
+ :param oid: id of the entry
+ :type oid: int
+ """
+ if oid is not None:
+ return db[str(oid)]
+ else:
+ for local_key, local_value in bcad_shelf.items():
+ try:
+ local_key = int(local_key)
+ except:
+ continue
+ yield local_value
+
+ def update_entry(db: shelve.Shelf, oid: int, obj: any):
+ """Update an entry in the shelf.
+
+ :param db: shelf specified
+ :type db: Shelve
+ :param oid: id of the entry
+ :type oid: int
+ :param data: data to be updated
+ :type data: any
+ """
+ db[str(oid)] = {"info": obj.property, "obj": obj}
+
+
+def clean_up():
+ if ENABLE_SQL_DATABASE:
+ conn.close()
+ delete_sql_db_file(db_name)
+ if ENABLE_SHELF:
+ bcad_shelf.close()
+ delete_shelve_db_files(shelve_name)
+
+
+
+
[docs]
+
def pnt(pt_coord) -> np.ndarray:
+
"""
+
Create a point.
+
:param pt_coord: The coordinate of the point. If the dimension is less than 3, the rest will be padded with 0. If the dimension is more than 3, an exception will be raised.
+
:type pt_coord: list
+
:return: The coordinate of the point.
+
:rtype: np.ndarray
+
"""
+
opt = np.array(pt_coord)
+
dim = np.shape(pt_coord)[0]
+
if dim > 3:
+
raise ValueError(
+
f"Got wrong point {pt_coord}: Dimension more than 3rd provided."
+
)
+
if dim < 3:
+
opt = np.lib.pad(opt, ((0, 3 - dim)), "constant", constant_values=0)
+
return opt
+
[docs]
+
class DuplicationCheck:
+
"""
+
Check if an item already exists in the index.
+
"""
+
+
def __init__(
+
self,
+
gtype: int,
+
gvalue: any,
+
) -> None:
+
self.gtype = gtype
+
self.gvalue = gvalue
+
self.check_type_validity(item_type=self.gtype)
+
self.new, self.exist_object = self.new_item(gvalue, gtype)
+
if not self.new:
+
self.exist_object.re_init = True
+
exist_obj_id = self.exist_object.id
+
logging.debug(
+
"%s %s %s already exists, return the old one.",
+
TYPE_INDEX[gtype],
+
exist_obj_id,
+
gvalue,
+
)
+
+
def __setstate__(self, state):
+
# This method is called during deserialization.
+
# Update the state to indicate this instance was deserialized from pickle.
+
state["deserialized_from_pickle"] = True
+
self.__dict__.update(state)
+
+
+
[docs]
+
def check_type_coincide(self, base_type: int, item_type: int) -> bool:
+
"""Check if items has the same type with the base item.
+
+
:param base_type: The type referred
+
:type base_type: int
+
:param item_type: The type to be examined
+
:type item_type: int
+
:return: True if coincident.
+
:rtype: bool
+
"""
+
different = base_type != item_type
+
self.check_type_validity(item_type=item_type)
+
if different:
+
return False
+
else:
+
return True
+
+
+
+
[docs]
+
def check_type_validity(self, item_type: int) -> bool:
+
"""Check if given type if valid
+
+
:param item_type: The type to be examined
+
:type item_type: int
+
:raises ValueError: Wrong geometry object type, perhaps a mistake made in development.
+
:return: True if valid
+
:rtype: bool
+
"""
+
valid = item_type in TYPE_INDEX
+
if not valid:
+
raise ValueError(
+
"Wrong geometry object type, perhaps a mistake made in development."
+
)
+
return valid
+
+
+
+
[docs]
+
def check_value_repetition(self, base_value: any, item_value: any) -> bool:
+
"""
+
Check if a value is close enough to the base value. True if repeated.
+
:param base_value: item to be compared with.
+
:param item_value: value to be examined.
+
"""
+
if type(base_value) is list:
+
base_value = np.array(base_value)
+
item_value = np.array(item_value)
+
+
return np.isclose(np.linalg.norm(base_value - item_value), 0, atol=1e-3)
+
+
+
+
[docs]
+
def new_item(self, item_value: any, item_type) -> tuple:
+
"""
+
Check if a value already exits in the index
+
:param item_value: value to be examined.
+
"""
+
match STATUS:
+
case "memory":
+
for _, item in component_lib.items():
+
if self.check_type_coincide(item["type"], item_type):
+
if self.check_value_repetition(item["value"], item_value):
+
return False, component_container[item["id"]]
+
case "database":
+
obj_type = TYPE_INDEX[item_type]
+
components = {}
+
for obj_item in get_from_source(obj_type=obj_type):
+
obj_gid = obj_item.gid
+
components.update({obj_gid: obj_item})
+
for gid, item in components.items():
+
if self.check_value_repetition(item.value, item_value):
+
exist_obj = item
+
return False, exist_obj # BUG: Recursion Problem.
+
case "shelf":
+
for it_value in get_entry(bcad_shelf):
+
if self.check_type_coincide(it_value["info"]["type"], item_type):
+
if self.check_value_repetition(
+
it_value["info"]["value"], item_value
+
):
+
exist_obj = it_value["obj"]
+
return False, exist_obj
+
return True, None
+
+
[docs]
+
def get_from_source(
+
oid: int = None, it_value: list = None, obj_type: str = None, target: str = "object"
+
):
+
"""A wrapper function to get the object from the source. if oid is not specified, it_value or obj_type will be used. If none of them is specified, all objects will be returned.
+
+
Notice: In case of stack overflow, when return all objects or all in one type, the function will return a generator.
+
+
:param oid: The id of the object to be retrieved.
+
:type oid: int
+
:param it_value: The value of the object to be retrieved.
+
:type it_value: list
+
:param obj_type: The type of the object to be retrieved.
+
:type obj_type: str
+
:param target: The target of the retrieval, either "object" or "property".
+
:type target: str
+
:return: The object retrieved.
+
:rtype: any
+
"""
+
mode = (
+
"id"
+
if oid is not None
+
else (
+
"value"
+
if it_value is not None
+
else ("type" if obj_type is not None else "all")
+
)
+
)
+
match mode:
+
case "id":
+
match STATUS:
+
case "memory":
+
if target == "object":
+
yield component_container[oid]
+
elif target == "property":
+
yield component_lib[oid]
+
case "database":
+
if target == "object":
+
yield from get_obj_from_db(oid=oid)
+
elif target == "property":
+
yield from get_property_from_db(oid=oid)
+
case "shelf":
+
if target == "object":
+
yield bcad_shelf[str(oid)]["obj"]
+
elif target == "property":
+
yield bcad_shelf[str(oid)]["info"]
+
case "value":
+
match STATUS:
+
case "memory":
+
if target == "object":
+
for item in component_container.values():
+
if TYPE_INDEX[item.type] == obj_type:
+
yield item
+
elif target == "property":
+
for item in component_lib.values():
+
if TYPE_INDEX[item["type"]] == obj_type:
+
yield item
+
case "database":
+
yield from get_obj_from_db(it_value=it_value)
+
case "shelf":
+
for item in get_entry(bcad_shelf):
+
if item["info"]["value"] == it_value:
+
if target == "object":
+
yield item["obj"]
+
elif target == "property":
+
yield item["info"]
+
case "type":
+
match STATUS:
+
case "memory":
+
if target == "object":
+
for item in component_container.values():
+
if TYPE_INDEX[item.type] == obj_type:
+
yield item
+
elif target == "property":
+
for item in component_lib.values():
+
if TYPE_INDEX[item["type"]] == obj_type:
+
yield item
+
case "database":
+
# This is a generator
+
yield from get_obj_from_db(obj_type=obj_type)
+
case "shelf":
+
for item in get_entry(bcad_shelf):
+
if item["info"]["type"] == obj_type:
+
if target == "object":
+
yield item["obj"]
+
elif target == "property":
+
yield item["info"]
+
case "all":
+
match STATUS:
+
case "memory":
+
# This is a dictionary
+
yield component_lib
+
case "database":
+
# This is a generator
+
yield from get_obj_from_db()
+
case "shelf":
+
# This is a generator
+
yield get_entry(bcad_shelf)
+
[docs]
+
def send_to_source(obj: TopoObj, obj_property: dict):
+
"""A wrapper function to send the object to the source.
+
+
:param obj: The object to be sent.
+
:type obj: TopoObj
+
:param obj_property: The property of the object to be sent.
+
:type obj_property: dict
+
"""
+
match STATUS:
+
case "memory":
+
obj.update_basic_property()
+
obj.update_component_container()
+
case "database":
+
store_obj_to_db(obj.type, obj, obj_property)
+
case "shelf":
+
add_entry(bcad_shelf, {"info": obj_property, "obj": obj})
+
[docs]
+
def update_source(obj: TopoObj, obj_property: dict = None):
+
"""A wrapper function to update the source of the object.
+
:param obj: The object to be updated.
+
:type obj: TopoObj
+
:param obj_property: The property of the object to be updated.
+
:type obj_property: dict
+
"""
+
match STATUS:
+
case "memory":
+
component_container.update({obj.id: obj})
+
component_lib.update({obj.id: obj.property})
+
case "database":
+
update_obj_in_db(obj, obj_property)
+
case "shelf":
+
update_entry(bcad_shelf, obj.id, obj)
+
[docs]
+
class Pnt(TopoObj):
+
"""
+
Create a point.
+
"""
+
+
def __new__(cls, coord: list = [], loading_state=None) -> None:
+
if cls._is_deserializing:
+
return super().__new__(cls)
+
point = pnt(coord)
+
checker = DuplicationCheck(0, point)
+
if checker.new:
+
instance = super().__new__(cls)
+
instance.re_init = False
+
return instance
+
else:
+
return checker.exist_object
+
+
def __init__(self, coord: list = [], loading_state=None) -> None:
+
if self._is_deserializing:
+
for key, value in loading_state.items():
+
setattr(self, key, value)
+
elif self.re_init:
+
return
+
else:
+
super().__init__()
+
self.re_init = False
+
self.type = 0
+
self.coord = pnt(coord)
+
self.value = self.coord
+
if ENABLE_OCC:
+
self.occ_pnt = gp_Pnt(*self.coord.tolist())
+
self.enrich_property({"occ_pnt": self.occ_pnt})
+
self.id = self.register_item()
+
logger.debug("Created point %s: %s", self.id, self.value)
+
+
@classmethod
+
def my_custom_reconstructor(cls, state_dict):
+
# Set the flag to indicate deserialization is in progress
+
cls._is_deserializing = True
+
# Create a new instance (which calls __new__ and then __init__)
+
instance = cls(loading_state=state_dict)
+
# Reset the flag to avoid affecting other instantiations
+
# Apply the saved state
+
instance.__dict__.update(state_dict)
+
cls._is_deserializing = False
+
return instance
+
+
def __reduce__(self):
+
return (self.my_custom_reconstructor, (self.__dict__,))
+
[docs]
+
def angle_of_two_arrays(a1: np.ndarray, a2: np.ndarray, rad: bool = True) -> float:
+
"""
+
@brief Returns the angle between two vectors. This is useful for calculating the rotation angle between a vector and another vector
+
@param a1 1D array of shape ( n_features )
+
@param a2 2D array of shape ( n_features )
+
@param rad If True the angle is in radians otherwise in degrees
+
@return Angle between a1 and a2 in degrees or radians depending on rad = True or False
+
"""
+
dot = np.dot(a1, a2)
+
norm = np.linalg.norm(a1) * np.linalg.norm(a2)
+
cos_value = np.round(dot / norm, 15)
+
if rad:
+
return np.arccos(cos_value)
+
else:
+
return np.rad2deg(np.arccos(cos_value))
+
[docs]
+
def project_array(array: np.ndarray, direct: np.ndarray) -> np.ndarray:
+
"""
+
Project an array to the specified direction.
+
"""
+
direct = direct / np.linalg.norm(direct)
+
return np.dot(array, direct) * direct
+
[docs]
+
def shortest_distance_point_line(
+
line: Union[list, Segment, np.ndarray], p: Union[list, Segment, np.ndarray]
+
):
+
"""
+
Calculate the shortest distance between a point and a line (segment).
+
+
:param line: The line (segment) to be examined.
+
:type line: Union[list, Segment, np.ndarray]
+
:param p: The point to be examined.
+
:type p: Union[list, Segment, np.ndarray]
+
:return: The relative coordinate of the point on the line (segment) and The shortest distance between the point and the line (segment).
+
:rtype: Tuple[Union[int, float], float]
+
"""
+
if isinstance(line, Segment):
+
pt1, pt2 = line.raw_value
+
p = p.value
+
else:
+
pt1, pt2 = line
+
s = pt2 - pt1
+
lmbda = (p - pt1).dot(s) / s.dot(s)
+
if lmbda < 1 and lmbda > 0:
+
pt_compute = pt1 + lmbda * s
+
distance = np.linalg.norm(pt_compute - p)
+
return lmbda, distance
+
elif lmbda <= 0:
+
distance = np.linalg.norm(pt1 - p)
+
return 0, distance
+
else:
+
distance = np.linalg.norm(pt2 - p)
+
return 1, distance
+
[docs]
+
def get_literal_vector(a: np.ndarray, d: bool):
+
"""
+
@brief This is used to create a vector which is perpendicular to the based vector on its left side ( d = True ) or right side ( d = False )
+
@param a vector ( a )
+
@param d True if on left or False if on right
+
@return A vector.
+
"""
+
z = np.array([0, 0, 1])
+
# cross product of z and a
+
if d:
+
na = np.cross(z, a)
+
else:
+
na = np.cross(-z, a)
+
norm = np.linalg.norm(na, na.shape[0])
+
return na / norm
+
[docs]
+
def bisect_angle(
+
a1: Union[np.ndarray, Segment], a2: Union[np.ndarray, Segment]
+
) -> np.ndarray:
+
"""
+
@brief Angular bisector between two vectors. The result is a vector splitting the angle between two vectors uniformly.
+
@param a1 1xN numpy array
+
@param a2 1xN numpy array
+
@return the bisector vector
+
"""
+
if isinstance(a1, Segment):
+
a1 = a1.vector
+
if isinstance(a2, Segment):
+
a2 = a2.vector
+
norm1 = np.linalg.norm(a1)
+
norm2 = np.linalg.norm(a2)
+
bst = a1 / norm1 + a2 / norm2
+
norm3 = np.linalg.norm(bst)
+
# The laterality indicator a2 norm3 norm3
+
if norm3 == 0:
+
opt = get_literal_vector(a2, True)
+
else:
+
opt = bst / norm3
+
return opt
+
[docs]
+
def p_rotate(
+
pts: np.ndarray,
+
angle_x: float = 0,
+
angle_y: float = 0,
+
angle_z: float = 0,
+
cnt: np.ndarray = None,
+
) -> np.ndarray:
+
"""
+
Rotate points around a center point
+
+
:param pts: The points to be rotated
+
:type pts: np.ndarray
+
:param angle_x: The angle to rotate around x-axis, defaults to 0
+
:type angle_x: float, optional
+
:param angle_y: The angle to rotate around y-axis, defaults to 0
+
:type angle_y: float, optional
+
:param angle_z: The angle to rotate around z-axis, defaults to 0
+
:type angle_z: float, optional
+
:param cnt: The center point to rotate around, defaults to None
+
:type cnt: np.ndarray, optional
+
:return: The rotated points
+
:rtype: np.ndarray
+
+
"""
+
pts = np.array(
+
[
+
np.array(list(i.Coord())) if isinstance(i, gp_Pnt) else np.array(i)
+
for i in pts
+
]
+
)
+
com = get_center_of_mass(pts)
+
if cnt is None:
+
cnt = np.array([0, 0, 0])
+
t_vec = cnt - com
+
pts += t_vec
+
rot_x = np.array(
+
[
+
[1, 0, 0],
+
[0, np.cos(angle_x), -np.sin(angle_x)],
+
[0, np.sin(angle_x), np.cos(angle_x)],
+
]
+
)
+
rot_y = np.array(
+
[
+
[np.cos(angle_y), 0, np.sin(angle_y)],
+
[0, 1, 0],
+
[-np.sin(angle_y), np.cos(angle_y), 0],
+
]
+
)
+
rot_z = np.array(
+
[
+
[np.cos(angle_z), -np.sin(angle_z), 0],
+
[np.sin(angle_z), np.cos(angle_z), 0],
+
[0, 0, 1],
+
]
+
)
+
R = rot_x @ rot_y @ rot_z
+
rt_pts = pts @ R
+
r_pts = rt_pts - t_vec
+
return r_pts
+
[docs]
+
def get_random_pnt(
+
xmin, xmax, ymin, ymax, zmin=0, zmax=0, numpy_array=True
+
) -> Union[Pnt, np.ndarray]:
+
"""
+
Get a random point within the specified range
+
+
:param xmin: The minimum value of x
+
:type xmin: int
+
:param xmax: The maximum value of x
+
:type xmax: int
+
:param ymin: The minimum value of y
+
:type ymin: int
+
:param ymax: The maximum value of y
+
:type ymax: int
+
:param zmin: The minimum value of z, defaults to 0
+
:type zmin: int, optional
+
:param zmax: The maximum value of z, defaults to 0
+
:type zmax: int, optional
+
:param numpy_array: If True, return a numpy array, otherwise return a Pnt object, defaults to True
+
:type numpy_array: bool, optional
+
:return: The random point
+
:rtype: Union[Pnt, np.ndarray]
+
+
"""
+
random_x = np.random.randint(xmin, xmax)
+
random_y = np.random.randint(ymin, ymax)
+
if zmin == 0 and zmax == 0:
+
random_z = 0
+
else:
+
random_z = np.random.randint(zmin, zmax)
+
if numpy_array:
+
result = np.array([random_x, random_y, random_z])
+
else:
+
result = Pnt([random_x, random_y, random_z])
+
+
return result
+
[docs]
+
def get_random_line(xmin, xmax, ymin, ymax, zmin=0, zmax=0):
+
"""
+
Get a random line within the specified range
+
+
:param xmin: The minimum value of x
+
:type xmin: int
+
:param xmax: The maximum value of x
+
:type xmax: int
+
:param ymin: The minimum value of y
+
:type ymin: int
+
:param ymax: The maximum value of y
+
:type ymax: int
+
:param zmin: The minimum value of z, defaults to 0
+
:type zmin: int, optional
+
:param zmax: The maximum value of z, defaults to 0
+
:type zmax: int, optional
+
:return: The random line
+
:rtype: np.ndarray
+
"""
+
+
pt1 = get_random_pnt(xmin, xmax, ymin, ymax, zmin, zmax)
+
pt2 = get_random_pnt(xmin, xmax, ymin, ymax, zmin, zmax)
+
return np.array([pt1, pt2])
+
[docs]
+
def find_intersect_node_on_edge(
+
line1: Segment, line2: Segment, update_property: dict = None
+
) -> tuple:
+
"""Find possible intersect node on two lines
+
+
:param line1: The first line
+
:type line1: Union[np.ndarray, Segment]
+
:param line2: The second line
+
:type line2: Union[np.ndarray, Segment]
+
:param update_property: The property to update, defaults to None
+
:type update_property: dict, optional
+
:return: The possible intersect node on the two lines
+
:rtype: tuple
+
"""
+
result = []
+
parallel, colinear = check_parallel_line_line(line1, line2)
+
if parallel:
+
if colinear:
+
index, coords = check_overlap(line1, line2)
+
if len(index) < 4:
+
for ind in index:
+
if ind in [0, 1]:
+
pnt_candidate = [line1.start_pnt, line1.end_pnt]
+
result.append((line2.id, pnt_candidate[ind]))
+
else:
+
pnt_candidate = [line2.start_pnt, line2.end_pnt]
+
result.append((line1.id, pnt_candidate[ind - 2]))
+
+
else:
+
distance = shortest_distance_line_line(line1, line2)
+
intersect = np.isclose(distance[0], 0)
+
if intersect:
+
intersect_pnt = Pnt(distance[1][0])
+
if update_property is not None:
+
intersect_pnt.enrich_property(update_property)
+
update_source(intersect_pnt)
+
result.append((line1.id, intersect_pnt.id))
+
result.append((line2.id, intersect_pnt.id))
+
return result
+
[docs]
+
def read_from_csv(file_path: str, delimiter: str = ",") -> list:
+
"""Read data from a csv file
+
+
:param file_path: The path of the csv file
+
:type file_path: str
+
:param delimiter: The delimiter of the csv file, defaults to ","
+
:type delimiter: str, optional
+
:return: The data read from the csv file
+
:rtype: list
+
"""
+
return np.genfromtxt(file_path, delimiter=delimiter, skip_header=1, dtype=float)
+The function returns the shortest distance between a point and a line (segment). For example in figure above the result will be 