Add cq_cache decorator

plugins/cq_cache/README.md

@@ -0,0 +1,67 @@
+# cq_cache
+
+This plugin provide a decorator function that allow you to add a file based cache to your functions to allow you to speed up the execution of your computational heavy cadquery functions.
+
+## Installation
+
+To install this plugin, the following line should be used.
+
+```
+pip install -e "git+https:/CadQuery/cadquery-plugins.git#egg=cq_cache&subdirectory=plugins/cq_cache"
+```
+You can also clone the repository of the plugin and run in the repository the following command :
+```
+python setup.py install
+```
+
+## Dependencies
+
+This plugin has no dependencies other than the cadquery library.
+
+## Usage
+
+To use this plugin after it has been installed, just import it and use the decorator on your functions
+
+```python
+#decorate your functions that build computational heavy shapes
+from cq_cache import cq_cache, clear_cq_cache
+
+@cq_cache(cache_size=1)
+def make_cube(a,b,c):
+ cube = cq.Workplane().box(a,b,c)
+ return cube
+
+for i in range(200):
+ make_cube(1,1,1+i)
+
+clear_cq_cache()
+#>>> Cache cleared for 1.036 MB
+```
+
+## Speed gain example 
+```python
+import cadquery as cq 
+import time, functools
+from cq_cache import cq_cache
+from itertools import cycle
+
+@cq_cache()
+def lofting(nb_sec):
+ wires = []
+ radius = cycle([2,5,8,6,10])
+ for i in range(nb_sec): 
+ if i%2==0:
+ Y = 1
+ else:
+ Y = 0 
+ wires.append(cq.Wire.makeCircle(next(radius),cq.Vector(0,0,5*i),cq.Vector(0,Y,1)))
+ loft = cq.Solid.makeLoft(wires)
+ return loft
+
+lofting(500)
+
+# First script run :
+# >>> 4500 ms
+# Second script run :
+# >>> 20 ms
+```

plugins/cq_cache/cq_cache.py

@@ -0,0 +1,138 @@
+import cadquery as cq
+import cadquery 
+from cadquery import exporters, importers
+from functools import wraps
+import tempfile
+import os
+import inspect
+
+from OCP.BRepTools import BRepTools
+from OCP.BRep import BRep_Builder
+from OCP.TopoDS import TopoDS_Shape
+
+
+
+
+
+TEMPDIR_PATH = tempfile.gettempdir()
+CACHE_DIR_NAME = "cadquery_geom_cache"
+CACHE_DIR_PATH = os.path.join(TEMPDIR_PATH,CACHE_DIR_NAME)
+CQ_TYPES = [cq.Shape, cq.Solid, cq.Shell, cq.Compound, cq.Face, cq.Wire, cq.Edge, cq.Vertex, TopoDS_Shape, cq.Workplane]
+
+if CACHE_DIR_NAME not in os.listdir(TEMPDIR_PATH):
+ os.mkdir(CACHE_DIR_PATH)
+
+def importBrep(file_path):
+ builder = BRep_Builder()
+ shape = TopoDS_Shape()
+ return_code = BRepTools.Read_s(shape, file_path, builder)
+ if return_code is False:
+ raise ValueError("Import failed, check file name")
+ return shape
+
+
+def get_cache_dir_size(cache_dir_path):
+ total_size = 0
+ for dirpath, dirnames, filenames in os.walk(cache_dir_path):
+ for f in filenames:
+ fp = os.path.join(dirpath, f)
+ total_size += os.path.getsize(fp)
+ return total_size
+
+def delete_oldest_file(cache_dir_path):
+ cwd = os.getcwd()
+ os.chdir(cache_dir_path)
+ files = sorted(os.listdir(os.getcwd()), key=os.path.getmtime)
+ oldest = files[0]
+ os.remove(os.path.join(cache_dir_path,oldest))
+ os.chdir(cwd)
+
+
+def build_file_name(fct, *args, **kwargs):
+ SPACER = "_"
+ file_name = fct.__name__
+ for arg in args:
+ file_name += SPACER + str(arg)
+ for kwarg_value in kwargs.values():
+ file_name += SPACER + str(kwarg_value)
+
+ return file_name
+
+def clear_cq_cache():
+ cache_size = get_cache_dir_size(CACHE_DIR_PATH)
+ for cache_file in os.listdir(CACHE_DIR_PATH):
+ os.remove(os.path.join(CACHE_DIR_PATH, cache_file))
+ print(f"Cache cleared for {round(cache_size*1e-6,3)} MB ")
+
+def using_same_function(fct, file_name):
+ with open(file_name,"r") as f :
+ cached_function = "".join(f.readlines()[:-1])
+
+ caching_function = inspect.getsource(fct)
+ if cached_function == caching_function:
+ return True 
+ else :
+ return False
+
+def return_right_wrapper(source, target_file): 
+ CQ_TYPES_STR = [str(cq_type) for cq_type in CQ_TYPES]
+
+ with open(target_file, "r") as tf:
+ target = tf.readlines()[-1]
+ target = target.replace("class ","").lstrip("<'").rstrip("'>") #eval cannot evaluate this "<class 'cadquery.cq.Workplane'>"" but this "cadquery.cq.Workplane" is ok
+ target = eval(target) #by the checking above forbids malicious excecution
+
+ for cq_type in CQ_TYPES:
+ if target == cq_type:
+ if cq_type == cq.Workplane:
+ shape = cq.Shape(source)
+ shape = cq.Workplane(obj=shape)
+ else:
+ shape = cq_type(source) 
+ return shape 
+
+def cq_cache(cache_size = 500):
+ """
+ Maximum cache memory in MB
+ """
+ def _cq_cache(function): 
+
+ @wraps(function) 
+ def wrapper(*args, **kwargs):
+ file_name = build_file_name(function, *args, **kwargs)
+ txt_file_path = os.path.join(CACHE_DIR_PATH, file_name+".txt")
+
+ if file_name+".txt" in os.listdir(CACHE_DIR_PATH) and using_same_function(function, txt_file_path) : #check that a change in function passed doesn't load up an old BREP file.
+ shape = importBrep(os.path.join(CACHE_DIR_PATH,file_name+".brep")) #If implemented in cadquery, could switch to the cadquery version of importBrep 
+ return return_right_wrapper(shape,txt_file_path) 
+
+
+ else:
+ shape = function(*args, **kwargs)
+ shape_type = type(shape)
+ if shape_type not in CQ_TYPES:
+ raise TypeError(f"cq_cache cannot wrap {shape_type} objects")
+ try :
+ shape_export = shape.val() # if shape is a workplane retrive only the shape object
+ except AttributeError:
+ shape_export = shape
+
+ shape_export.exportBrep(os.path.join(CACHE_DIR_PATH,file_name)+".brep")
+
+ with open(os.path.join(CACHE_DIR_PATH,file_name)+ ".txt", "w") as fun_file:
+ fun_file.write(inspect.getsource(function))
+ fun_file.write(str(shape_type))
+
+ cache_dir_size = get_cache_dir_size(CACHE_DIR_PATH)
+ while (cache_dir_size*1e-6) > cache_size:
+ delete_oldest_file(CACHE_DIR_PATH)
+ cache_dir_size = get_cache_dir_size(CACHE_DIR_PATH)
+
+ return shape
+
+ return wrapper
+
+ return _cq_cache
+
+
+

plugins/cq_cache/setup.py

@@ -0,0 +1,45 @@
+from setuptools import setup, find_packages
+
+version = "1.0.0" # Please update this version number when updating the plugin
+plugin_name = "cq_cache"
+description = "File based cache decorator"
+long_description = "Allow to use file based cache to not have to rebuild every cadquery model from scratch"
+author = "Romain FERRU"
+author_email = "[email protected]"
+install_requires = (
+ []
+) # Any dependencies that pip also needs to install to make this plugin work
+
+
+setup(
+ name=plugin_name,
+ version=version,
+ url="https:/CadQuery/cadquery-plugins",
+ license="Apache Public License 2.0",
+ author=author,
+ author_email=author_email,
+ description=description,
+ long_description=long_description,
+ packages=find_packages(where="cq_cache"),
+ install_requires=install_requires,
+ include_package_data=True,
+ zip_safe=False,
+ platforms="any",
+ test_suite="tests",
+ classifiers=[
+ "Development Status :: 5 - Production/Stable",
+ "Intended Audience :: Developers",
+ "Intended Audience :: End Users/Desktop",
+ "Intended Audience :: Information Technology",
+ "Intended Audience :: Science/Research",
+ "Intended Audience :: System Administrators",
+ "License :: OSI Approved :: Apache Software License",
+ "Operating System :: POSIX",
+ "Operating System :: MacOS",
+ "Operating System :: Unix",
+ "Programming Language :: Python",
+ "Topic :: Software Development :: Libraries :: Python Modules",
+ "Topic :: Internet",
+ "Topic :: Scientific/Engineering",
+ ],
+)

tests/test_cq_cache.py

@@ -0,0 +1,61 @@
+import cadquery as cq
+import cadquery
+from plugins.cq_cache.cq_cache import cq_cache, clear_cq_cache, get_cache_dir_size
+import tempfile
+import os
+
+TEMPDIR_PATH = tempfile.gettempdir()
+CACHE_DIR_NAME = "cadquery_geom_cache"
+CACHE_DIR_PATH = os.path.join(TEMPDIR_PATH,CACHE_DIR_NAME)
+CACHE_SIZE = 0.1
+
+@cq_cache(CACHE_SIZE)
+def cube(a,b,c):
+ cube = cq.Workplane().box(a,b,c)
+ return cube.val()
+
+def test_get_cache_dir_size():
+ with open(os.path.join(CACHE_DIR_PATH,"fill.txt"), "w") as f:
+ f.write("test") 
+ assert get_cache_dir_size(CACHE_DIR_PATH) == 4
+
+def test_clear_cache():
+ with open(os.path.join(CACHE_DIR_PATH,"fill.txt"), "w") as f:
+ f.write("test")
+ assert len(os.listdir(CACHE_DIR_PATH)) == 1
+ clear_cq_cache()
+ assert len(os.listdir(CACHE_DIR_PATH)) == 0
+
+def test_cache_file_creation():
+ clear_cq_cache()
+ cube1 = cube(1,1,1)
+ cube2 = cube(1,1,1)
+ files = os.listdir(CACHE_DIR_PATH)
+ assert len(files) == 2
+ assert "cube_1_1_1.brep" in files
+ assert "cube_1_1_1.txt" in files
+
+def test_not_exceeding_size():
+ clear_cq_cache()
+ for i in range(20):
+ cube(1,1,1+i)
+ assert get_cache_dir_size(CACHE_DIR_PATH) < CACHE_SIZE*1e6
+
+def test_cache_type_return():
+ cube1 = cube(1,1,1) #at first call get the type directly from function call
+ cube2 = cube(1,1,1) #at second call the decorator retrives the right type with some logic that may fail
+ assert isinstance(cube1,cadquery.occ_impl.shapes.Solid)
+ assert isinstance(cube2,cadquery.occ_impl.shapes.Solid)
+
+def test_cache_type_return_with_modified_function():
+
+ @cq_cache(CACHE_SIZE)
+ def cube(a,b,c):
+ cube = cq.Workplane().box(a,b,c)
+ return cube
+
+ cube1 = cube(1,1,1) #at first call get the type directly from function call
+ cube2 = cube(1,1,1) #at second call the decorator retrives the right type with some logic that may fail
+ assert isinstance(cube1,cq.Workplane)
+ assert isinstance(cube2,cq.Workplane)
+