diff --git a/README.md b/README.md index 3058690..c515be9 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,10 @@ Then, create your `main.py` to test your specs locally ```py # main.py -import maple as mp +from maple.base import Maple + +# Use the project and model id of one of your projects +mp=Maple(project_id="24fa0ed1c3", model_id="2696b4a381") def spec_a(): mp.it("checks window height is greater than 2600 mm") @@ -39,8 +42,6 @@ def spec_a(): .its('Height')\ .should('be.greater', 2600) -# Use the project and model id of one of your projects -mp.init_model(project_id="24fa0ed1c3", model_id="2696b4a381") mp.run(spec_a) ``` For this to work out of the box, you should have the [Speckle Manager](https://speckle.systems/download/) diff --git a/examples/federated_model.py b/examples/federated_model.py new file mode 100644 index 0000000..e2e5a18 --- /dev/null +++ b/examples/federated_model.py @@ -0,0 +1,32 @@ +from maple.base import Maple +from maple.utils import print_results +import logging + +logging.basicConfig(level=logging.INFO) + +arch_model = "045cdb6d44" +str_model = "df3fa4e97d" +mp = Maple( + project_id="566869f9e6", + model_ids=[arch_model, str_model], +) + + +def clash(): + mp.it("Clash detection between multiple models") + windows = mp.from_model(arch_model).get("properties.ifcType", "IfcWindow") + walls = mp.from_model(str_model).get("ifcType", "IfcColumn") + + mp.detect_collision(windows, walls) + + +def other_test(): + mp.it("other test") + windows = mp.from_model(arch_model).get("properties.ifcType", "IfcWindow") + windows.its("properties.ifcType").should("have.value", "ifcWindow") + + columns = mp.from_model(str_model).get("ifcType", "IfcColumn") + columns.its("ifcType").should("have.value", "ifcColumn") + + +mp.run(other_test, clash) diff --git a/examples/single_spec.py b/examples/single_spec.py index adaa09f..f96b537 100644 --- a/examples/single_spec.py +++ b/examples/single_spec.py @@ -1,18 +1,22 @@ import setup # noqa -import maple as mp +import logging + +from maple.base import Maple + +mp = Maple(model_id="53db0711db", project_id="1471fed2c0") + +logging.basicConfig(level=logging.INFO) def main(): - mp.init_model(project_id="1471fed2c0", model_id="53db0711db") - mp.set_logging(True) mp.run(test_check_door_height) - mp.generate_report(output_path="/tmp/") + # mp.generate_report(output_path="/tmp/") def test_check_door_height(): mp.it("Checks that the door height its at least 2.0 m") - mp.get("ifcType", "IFCSPATIALZONE").its("ownerId").should("be.equal", 1) + mp.get("ifcType", "IFCSPATIALZONE").its("ownerId").should("be.equal", 2) if __name__ == "__main__": diff --git a/pyproject.toml b/pyproject.toml index 480ac6c..5a2369d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,19 +4,18 @@ build-backend = "setuptools.build_meta" [project] name = "maple-spec" -version = "0.1.6" +version = "0.2" authors = [ { name="Gizem Demirhan", email="gizemdemirhaan@gmail.com" }, { name="Andres Buitrago", email="andrsbtrg@gmail.com" } ] description = "A testing library for Speckle models" readme = "README.md" -requires-python = ">=3.8" +requires-python = ">=3.11" dependencies = ["specklepy>=3", "importlib-metadata", "jinja2", "python-dotenv", "acers"] classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python :: 3", - "License :: OSI Approved :: Apache Software License", "Operating System :: OS Independent" ] diff --git a/src/maple/__init__.py b/src/maple/__init__.py index 873b3bc..2cfebde 100644 --- a/src/maple/__init__.py +++ b/src/maple/__init__.py @@ -1,7 +1,7 @@ # maple import from os import getenv -from typing import Any, Dict, Literal, List +from typing import Any, Dict, List from acers import clash_detection, Collision @@ -12,13 +12,15 @@ from specklepy.core.api.models.current import ModelWithVersions, Version from specklepy.objects import Base from specklepy.transports.server.server import ServerTransport -from typing_extensions import Callable, Self +from typing_extensions import Callable + +from maple.base.chainable import Chainable from .base_extensions import flatten_base -from .models import Assertion, Result -from .ops import CompOp, ComparisonOps, deep_get, property_equal +from .models import Result, Status +from .ops import property_equal from .report import HtmlReport -from .utils import print_results, log_collision +from .utils import print_results, log_collision, serialize_set import logging @@ -28,9 +30,6 @@ logging.getLogger("gql.transport.requests").setLevel(logging.WARNING) -Status = Literal["pass", "fail"] - - # GLOBALS # TODO: Refactor to remove globals _test_cases: list[Result] = [] # Contains the results of the runs @@ -39,6 +38,13 @@ _model_id: str = "" +DEPR_REASON = """ +Global mp functions are being deprecated and will be removed in the next version of maple. +Please use maple.base.Maple class and see migration. +""" + + +@deprecated(DEPR_REASON) def init(obj: Base) -> None: """ Caches the speckle object obj in the global _current_object @@ -84,6 +90,7 @@ def stream(id: str) -> None: return +@deprecated(DEPR_REASON) def init_model(project_id: str, model_id: str) -> None: """ Sets the global variables project id and model id for the @@ -119,6 +126,7 @@ def get_token() -> str | None: return token +@deprecated(DEPR_REASON) def get_model_id() -> str: """ Gets the Model id provided with mp.init_model @@ -129,6 +137,7 @@ def get_model_id() -> str: return _model_id +@deprecated(DEPR_REASON) def get_project_id() -> str: """ Gets the Model id provided with mp.init_model @@ -139,6 +148,7 @@ def get_project_id() -> str: return _project_id +@deprecated(DEPR_REASON) def get_current_obj() -> Base | None: """ Get the current object specified with mp.init() @@ -147,6 +157,7 @@ def get_current_obj() -> Base | None: return _current_object +@deprecated(DEPR_REASON) def get_current_test_case() -> Result | None: """ Get the current test case @@ -158,6 +169,7 @@ def get_current_test_case() -> Result | None: return current +@deprecated(DEPR_REASON) def get_test_cases() -> list[Result]: """ Gets the list of Test Cases @@ -166,6 +178,7 @@ def get_test_cases() -> list[Result]: return _test_cases +@deprecated(DEPR_REASON) def get_results() -> list[Any]: """ Gets the list of Results @@ -216,189 +229,7 @@ def get_results() -> list[Any]: # endof GLOBALS -class Chainable: - def __init__(self, data): - self.content = data - self.selector = "" - self.assertion: Assertion = Assertion() - - def _select_parameters_values(self, parameter_name: str) -> list[Any]: - """ - Gets a list of the values of each object in self.content - where the parameter_name matches - - Args: - parameter_name: - - Returns: a list of the value of the parameter matching - - Raises: - AttributeError: - - """ - parameter_values = [] - objs = self.content - # check on base object - for obj in objs: - value = deep_get(obj, parameter_name) - if not value: - break - parameter_values.append(value) - - if len(parameter_values) > 0: - return parameter_values - - return parameter_values - - def _should_have_length(self, length: int) -> Self: - """ - Use to check wether the content has length equal to length - - Args: - length (int): length to compare - - """ - objs = self.content - self.assertion.selector = "Collection" - if len(objs) == length: - self.assertion.set_passed("have.length") - else: - self.assertion.set_failed("have.length") - current = get_current_test_case() - if current is None: - raise Exception("Expected current test case not to be None") - current.assertions.append(self.assertion) - return self - - def _should_have_param_value(self, comparer: CompOp, assertion_value: Any) -> Self: - """ - Using the comparer will get the parameter given by the self.selector - for each object and compare each one against assertion_value - - Args: - comparer: CompOp - assertion_value: any value to compare - - Returns: Chainable - """ - selected_values = self._select_parameters_values(self.selector) - - objs = self.content - - # store results in the last Results in test_cases - for i, param_value in enumerate(selected_values): - if comparer.evaluate(param_value, assertion_value): - self.assertion.set_passed(objs[i].id) - else: - logger.warning(f"object id '{objs[i].id}' - value: {param_value}") - self.assertion.set_failed(objs[i].id) - - current = get_current_test_case() - if current is None: - raise Exception("Expected current test case not to be None") - current.assertions.append(self.assertion) - - return self - - def should(self, comparer: ComparisonOps, assertion_value) -> Self: - """ - Assert something inside the Chainable - Args: - comparer: one of CompOp possible enum values - assertion_value: value to assert - Raises: ValueError if comparer is not a defined CompOp - Returns: Chainable - """ - logger.info("Asserting - should: %s %s", comparer, assertion_value) - comparer_op = CompOp(comparer) - self.assertion.value = assertion_value - self.assertion.comparer = comparer_op - - if comparer_op == CompOp.HAVE_LENGTH: - return self._should_have_length(assertion_value) - else: - return self._should_have_param_value(comparer_op, assertion_value) - - def should_satisfy(self, func: Callable[[Any], bool]) -> Self: - """ - Asserts using a custom condition. - Args: - func: a function that takes one argument and returns true or false - Returns: Chainable - """ - logger.info("Asserting - should satisfy") - self.assertion.comparer = func - - selected_values = self._select_parameters_values(self.selector) - - objs = self.content - - # store results in the last Results in test_cases - for i, param_value in enumerate(selected_values): - if func(param_value): - self.assertion.set_passed(objs[i].id) - else: - logger.warning(f"object id '{objs[i].id}' - value: {param_value}") - self.assertion.set_failed(objs[i].id) - - current = get_current_test_case() - if current is None: - raise Exception("Expected current test case not to be None") - current.assertions.append(self.assertion) - - return self - - def its(self, property: str) -> Self: - """ - Selector of a parameter inside the Chainable object - - Args: - property: name of parameter to select from content - - Returns: Chainable - - Raises: - AttributeError: if the parameter name does not match in the - inner object selected with get - """ - logger.info("Selecting %s", property) - self.selector = property - self.assertion.selector = property - - objs = self.content - # check on base object - for obj in objs: - value = deep_get(obj, property) - if not value: - self.assertion.set_failed(obj.id) - logger.warning(f"object id: '{obj.id}' has no property '{property}'") - return self - - def where(self, selector: str, value: str) -> Self: - """ - Filters the current Speckle objects aquired by mp.get() - where the object's own property 'selector' is equal to 'value' - Args: - selector: The name of a property of a Speckle Object to select - e.g: type - value: The name of the value of the property to be filtered - - Returns: Chainable - """ - logger.info("Filtering by: %s - %s", selector, value) - current = get_current_test_case() - if current is None: - raise Exception("Expected current test case not to be None") - current.selected[selector] = value - - selected = list( - filter(lambda obj: property_equal(selector, value, obj), self.content) - ) - logger.info("Elements after filter: %i", len(selected)) - self.content = selected - return self - - +@deprecated(DEPR_REASON) def it(spec_name: str): """ Declares a new Spec and stores it globally in the test cases. @@ -413,6 +244,7 @@ def it(spec_name: str): get_test_cases().append(Result(spec_name)) +@deprecated(DEPR_REASON) def get(selector: str, value: str) -> Chainable: """ Does a speckle queries and then filters by 'selector'. @@ -438,7 +270,8 @@ def get(selector: str, value: str) -> Chainable: speckle_obj = get_current_obj() if not speckle_obj: - speckle_obj = get_last_obj() + speckle_obj = get_last_obj(get_project_id(), get_model_id()) + init(speckle_obj) if speckle_obj is None: raise Exception("Could not get a Base object to query.") @@ -447,10 +280,11 @@ def get(selector: str, value: str) -> Chainable: selected = list(filter(lambda obj: property_equal(selector, value, obj), objs)) logger.info("Got %i %s", len(selected), value) - return Chainable(selected) + return Chainable(selected, current_test) -def get_last_obj() -> Base: +@deprecated(DEPR_REASON) +def get_last_obj(project_id: str, model_id: str) -> Base: """ Gets the last object for the specified stream_id """ @@ -474,8 +308,8 @@ def get_last_obj() -> Base: else: logger.warning("No auth present") - project_id = get_project_id() - model_id = get_model_id() + # project_id = get_project_id() + # model_id = get_model_id() transport = ServerTransport(client=client, stream_id=project_id) models = client.model.get_models(project_id=project_id) @@ -499,10 +333,11 @@ def get_last_obj() -> Base: last_obj = operations.receive(obj_id=last_obj_id, remote_transport=transport) # cache the current obj - init(last_obj) + # init(last_obj) return last_obj +@deprecated(DEPR_REASON) def run(*specs: Callable): """ Runs any number of spec functions passed by args @@ -523,6 +358,7 @@ def run(*specs: Callable): print_results(get_test_cases()) +@deprecated(DEPR_REASON) def detect_collision( set_a: Chainable, set_b: Chainable, min_dist=0.0 ) -> List[Collision]: @@ -558,18 +394,6 @@ def detect_collision( return collisions -def serialize_set(test_set: Chainable) -> str: - """ - Serializes the seleced objects in a set to a string - """ - response = "" - for obj in test_set.content: - if isinstance(obj, Base): - if "displayValue" in obj.get_member_names(): - response += f"{obj.id}\t{operations.serialize(obj)}\n" - return response - - def print_info(specs): from importlib_metadata import version diff --git a/src/maple/base/__init__.py b/src/maple/base/__init__.py new file mode 100644 index 0000000..e6c7bd6 --- /dev/null +++ b/src/maple/base/__init__.py @@ -0,0 +1 @@ +from .maple import Maple diff --git a/src/maple/base/chainable.py b/src/maple/base/chainable.py new file mode 100644 index 0000000..ba6b655 --- /dev/null +++ b/src/maple/base/chainable.py @@ -0,0 +1,215 @@ +from typing import Any, Callable, Self +from maple.models import Assertion, Result +from maple.ops import CompOp, ComparisonOps, deep_get, property_equal + +import logging + +logger = logging.getLogger(__name__) + + +class Chainable: + def __init__(self, data, test_case: Result): + self.content = data + self.selector = "" + self.assertion: Assertion = Assertion() + self.current_test_case = test_case + + def _select_parameters_values(self, parameter_name: str) -> list[Any]: + """ + Gets a list of the values of each object in self.content + where the parameter_name matches + + Args: + parameter_name: + + Returns: a list of the value of the parameter matching + + Raises: + AttributeError: + + """ + parameter_values = [] + objs = self.content + # check on base object + for obj in objs: + value = deep_get(obj, parameter_name) + if not value: + break + parameter_values.append(value) + + if len(parameter_values) > 0: + return parameter_values + + return parameter_values + + def _should_have_length(self, length: int) -> Self: + """ + Use to check wether the content has length equal to length + + Args: + length (int): length to compare + + """ + objs = self.content + self.assertion.selector = "Collection" + if len(objs) == length: + self.assertion.set_passed("have.length") + else: + self.assertion.set_failed("have.length") + current = self.current_test_case + if current is None: + raise Exception("Expected current test case not to be None") + current.assertions.append(self.assertion) + return self + + def _should_have_param_value(self, comparer: CompOp, assertion_value: Any) -> Self: + """ + Using the comparer will get the parameter given by the self.selector + for each object and compare each one against assertion_value + + Args: + comparer: CompOp + assertion_value: any value to compare + + Returns: Chainable + """ + selected_values = self._select_parameters_values(self.selector) + + objs = self.content + + # store results in the last Results in test_cases + for i, param_value in enumerate(selected_values): + if comparer.evaluate(param_value, assertion_value): + self.assertion.set_passed(objs[i].id) + else: + logger.info( + f"Object id '{objs[i].id}'\n Expected {assertion_value}, got: {param_value}" + ) + self.assertion.set_failed(objs[i].id) + + current = self.current_test_case + current.assertions.append(self.assertion) + + return self + + def should(self, comparer: ComparisonOps, assertion_value) -> Self: + """ + Assert something inside the Chainable + Args: + comparer: one of CompOp possible enum values + assertion_value: value to assert + Raises: ValueError if comparer is not a defined CompOp + Returns: Chainable + """ + logger.info("Asserting - should: %s %s", comparer, assertion_value) + comparer_op = CompOp(comparer) + self.assertion.value = assertion_value + self.assertion.comparer = comparer_op + + if comparer_op == CompOp.HAVE_LENGTH: + return self._should_have_length(assertion_value) + else: + return self._should_have_param_value(comparer_op, assertion_value) + + def should_satisfy(self, func: Callable[[Any], bool]) -> Self: + """ + Asserts using a custom condition. + Args: + func: a function that takes one argument and returns true or false + Returns: Chainable + """ + logger.info("Asserting - should satisfy") + self.assertion.comparer = func + + selected_values = self._select_parameters_values(self.selector) + + objs = self.content + + # store results in the last Results in test_cases + for i, param_value in enumerate(selected_values): + if func(param_value): + self.assertion.set_passed(objs[i].id) + else: + logger.warning(f"object id '{objs[i].id}' - value: {param_value}") + self.assertion.set_failed(objs[i].id) + + current = self.current_test_case + if current is None: + raise Exception("Expected current test case not to be None") + current.assertions.append(self.assertion) + + return self + + def its(self, property: str) -> Self: + """ + Selector of a parameter inside the Chainable object + + Args: + property: name of parameter to select from content + + Returns: Chainable + + Raises: + AttributeError: if the parameter name does not match in the + inner object selected with get + """ + logger.info("Selecting %s", property) + self.selector = property + self.assertion.selector = property + + objs = self.content + # check on base object + for obj in objs: + value = deep_get(obj, property) + if not value: + self.assertion.set_failed(obj.id) + logger.warning(f"object id: '{obj.id}' has no property '{property}'") + return self + + def where(self, selector: str, value: str) -> Self: + """ + Filters the current Speckle objects aquired by mp.get() + where the object's own property 'selector' is equal to 'value' + Args: + selector: The name of a property of a Speckle Object to select + e.g: type + value: The name of the value of the property to be filtered + + Returns: Chainable + """ + logger.info("Filtering by: %s - %s", selector, value) + self.current_test_case.selected[selector] = value + + selected = list( + filter(lambda obj: property_equal(selector, value, obj), self.content) + ) + logger.info("Elements after filter: %i", len(selected)) + self.content = selected + return self + + def get(self, prop: str, value: str) -> Self: + """ + Returns the selected items inside the Chainable object + to start a chain of assertions + + Args: + prop: The name of a property of a Speckle Object to select + e.g: category, family + value: The objects whose selector matches this value will be filtered + + Returns: Chainable + + Raises: + Exception: If it was not possible to query a speckle object + """ + if len(self.content) == 0: + raise ValueError("Must first initialize with a valid list of objects") + + selected = list( + filter(lambda obj: property_equal(prop, value, obj), self.content) + ) + + logger.info("Got %i %s", len(selected), value) + self.current_test_case.selected[prop] = value + self.content = selected + return self diff --git a/src/maple/base/maple.py b/src/maple/base/maple.py new file mode 100644 index 0000000..1744c14 --- /dev/null +++ b/src/maple/base/maple.py @@ -0,0 +1,157 @@ +from typing import Any, Callable, Dict, overload, List, Optional +from specklepy.objects import Base + +from maple.base.chainable import Chainable +from maple.base_extensions import flatten_base + +from acers import clash_detection, Collision + + +import logging + +from maple.models import Result +from maple.ops import property_equal +from maple.speckle_utils import get_last_obj +from maple.utils import log_collision, print_results, serialize_set, print_info + +logger = logging.getLogger(__name__) + + +class Maple: + @overload + def __init__(self, *, project_id: str, model_id: str): ... + @overload + def __init__(self, *, project_id: str, model_ids: List[str]): ... + + def __init__( + self, + *, + project_id: str, + model_id: Optional[str] = None, + model_ids: Optional[list[str]] = None, + ): + if (model_id is None) == (model_ids is None): + raise ValueError("Provide exactly one of model_id or model_ids") + + self._results: list[Result] = [] + + self.project_id = project_id + self.model_ids: list[str] = [] + + if model_id: + self.model_ids.append(model_id) + if model_ids: + self.model_ids.extend(model_ids) + + # model cache + self.__model_store: Dict[str, Base] = {} + + @property + def results(self): + return self._results + + @property + def current_test_case(self): + if len(self.results) == 0: + raise ValueError("Results array is empty") + return self.results[-1] + + def get(self, prop: str, value: str): + if len(self.model_ids) == 0: + raise ValueError("model_id or model_ids is empty") + if len(self.model_ids) > 1: + logger.warning("multiple model_ids initialized. Please use 'from_model'") + + model_id = self.model_ids[0] + + speckle_obj = get_last_obj( + project_id=self.project_id, model_id=model_id, logger=logger + ) + + objs = list(flatten_base(speckle_obj)) + logger.info("Received %i speckle objects", len(objs)) + logger.info("Filtering by %s = %s", prop, value) + selected = list(filter(lambda obj: property_equal(prop, value, obj), objs)) + logger.info("Filtered %i where %s = %s", len(selected), prop, value) + + self.current_test_case.selected[prop] = value + + return Chainable(selected, self.current_test_case) + + def it(self, descr: str): + logger.info("Running test: %s", descr) + self.results.append(Result(descr)) + + def from_model(self, model_id: str) -> Chainable: + if model_id not in self.model_ids: + raise ValueError("Model id has not been initialized") + + speckle_obj = get_last_obj( + project_id=self.project_id, model_id=model_id, logger=logger + ) + + objs = list(flatten_base(speckle_obj)) + + return Chainable(objs, self.current_test_case) + + def run(self, *specs: Callable): + """ + Runs any number of spec functions passed by args + Args: + *specs: Callable + """ + print_info(specs) + + for i, spec in enumerate(specs): + if not callable(spec): + print( + "Warning - parameter at position " + + f"{i}" + + " is not spec function." + ) + continue + spec() + + # print results + print_results(self.results) + + def detect_collision( + self, set_a: Chainable, set_b: Chainable, min_dist=0.0 + ) -> List[Collision]: + """ + Check collision between all elements of two sets + Args: + set_a: Chainable + set_b: Chainable + min_dist: A distance between elements smaller than this will show as a clash. + Default = 0. Elements whose face are touching don't register as a collision + """ + + results = self.current_test_case + + results.type = "collision" + + logger.info("Executing collision detection") + set_a_string = serialize_set(set_a) + set_b_string = serialize_set(set_b) + collisions = clash_detection(set_a_string, set_b_string, min_dist) + + logger.info(f"Found {len(collisions)} collision(s)") + + results.collision_results = collisions + + if len(collisions) > 0: + for c in collisions: + log_collision(logger, c, set_a.content, set_b.content) + + return collisions + + def get_results(self) -> list[Any]: + """ + Gets the flattened list of Results + """ + + total_results = [] + for result in self.results: + total_results.extend(result.to_records()) + return total_results diff --git a/src/maple/models.py b/src/maple/models.py index 6fd6102..02380ea 100644 --- a/src/maple/models.py +++ b/src/maple/models.py @@ -1,6 +1,9 @@ -import maple from acers import Collision -from typing import Any, Callable, Literal, Self +from typing import Any, Callable, Dict, Literal + +from maple.ops import CompOp + +Status = Literal["pass", "fail"] class Assertion: @@ -17,7 +20,7 @@ class Assertion: """ def __init__(self) -> None: - self.comparer: maple.CompOp | Callable | None = None + self.comparer: CompOp | Callable | None = None self.value: Any = None # what will be compared to self.passing: list[str] = [] self.failing: list[str] = [] @@ -62,7 +65,47 @@ class Result: def __init__(self, spec_name: str) -> None: self.spec_name = spec_name - self.selected = {} + self.selected: Dict[str, str] = {} self.assertions: list[Assertion] = [] self.type: Literal["spec", "collision"] = "spec" self.collision_results: list[Collision] = [] + + def to_records(self): + total_selfs = [] + if self.type == "spec": + self_per_elem: Dict[str, Status] = {} + select = [] + for selector in self.selected.keys(): + select.append(f"{selector} = {self.selected[selector]}") + + for a in self.assertions: + descr = a.get_description() + for id in a.passing: + self_per_elem[id] = "pass" + for id in a.failing: + self_per_elem[id] = "fail" + overall: Status = "pass" if a.passed() else "fail" + total_selfs.append( + { + "type": "spec", + "spec_name": self.spec_name, + "get": select, + "spec": descr, + "result": overall, + "elements": self_per_elem, + } + ) + elif self.type == "collision": + collisions = self.collision_results + total_selfs.append( + { + "type": "collision", + "spec_name": self.spec_name, + "result": "fail" if len(collisions) > 0 else "pass", + "collisions": [ + {"ids": x.ids, "dist": x.dist, "point": x.point} + for x in collisions + ], + } + ) + return total_selfs diff --git a/src/maple/speckle_utils.py b/src/maple/speckle_utils.py new file mode 100644 index 0000000..4f67eae --- /dev/null +++ b/src/maple/speckle_utils.py @@ -0,0 +1,79 @@ +from os import getenv + +from specklepy.api.client import Account, SpeckleClient +from specklepy.api.credentials import get_account_from_token, get_default_account +from specklepy.core.api.models import ModelWithVersions, Version +from specklepy.objects import Base +from specklepy.transports.server.server import ServerTransport + +from specklepy.api import operations + + +def get_token() -> str | None: + """ + Get the token to authenticate with Speckle. + The token should be under the env variable 'SPECKLE_TOKEN' + """ + + token = getenv("SPECKLE_TOKEN") + return token + + +def account_match_host(account: Account, host: str) -> bool: + url = account.serverInfo.url + host_url = host.replace("https://", "") + host_url = host_url.replace("/", "") + return url == host_url + + +def get_last_obj(project_id: str, model_id: str, *, logger) -> Base: + """ + Gets the last object for the specified stream_id + """ + logger.info("Getting object from speckle") + host = getenv("SPECKLE_HOST") + if not host: + host = "https://app.speckle.systems" + logger.info("Using Speckle host: %s", host) + client = SpeckleClient(host) + # authenticate the client with a token + token = get_token() + if token: + logger.info("Auth with token") + account = get_account_from_token(token, host) + client.authenticate_with_token(token) + else: + account = get_default_account() + if account and account_match_host(account, host): + logger.info("Auth with default account") + client.authenticate_with_account(account) + else: + logger.warning("No auth present") + + # project_id = get_project_id() + # model_id = get_model_id() + transport = ServerTransport(client=client, stream_id=project_id) + + models = client.model.get_models(project_id=project_id) + found_model = next(filter(lambda x: x.id == model_id, models.items), None) + if not found_model: + raise Exception("Model not found: ", model_id) + + model: ModelWithVersions = client.model.get_with_versions( + project_id=project_id, model_id=found_model.id + ) + + versions = model.versions.items + if len(versions) == 0: + raise Exception("Model contains no versions.") + if type(versions[0]) is not Version: + raise Exception("Type of element is not Model: ", type(versions[0])) + last_obj_id = versions[0].referenced_object + + if not last_obj_id: + raise Exception("No object_id") + last_obj = operations.receive(obj_id=last_obj_id, remote_transport=transport) + + # cache the current obj + # init(last_obj) + return last_obj diff --git a/src/maple/utils.py b/src/maple/utils.py index 9d450b5..fecc27d 100644 --- a/src/maple/utils.py +++ b/src/maple/utils.py @@ -2,8 +2,11 @@ from acers import Collision from specklepy.objects import Base from .models import Result +from .base.chainable import Chainable from typing import Tuple, List +from specklepy.api import operations + import os @@ -73,3 +76,26 @@ def log_collision( # element_2 = next(base for base in set_b if base.id == c[1]) logger.info(f"Clash between {c.ids[0]} and {c.ids[1]}") + + +def serialize_set(test_set: Chainable) -> str: + """ + Serializes the seleced objects in a set to a string + """ + response = "" + for obj in test_set.content: + if isinstance(obj, Base): + if "displayValue" in obj.get_member_names(): + response += f"{obj.id}\t{operations.serialize(obj)}\n" + return response + + +def print_info(specs): + from importlib_metadata import version + + print_title("Test session") + + v = version("maple-spec") + print("Maple -", v) + print("collected", len(specs), "specs") + print() diff --git a/tests/test_clash_detections.py b/tests/test_clash_detections.py index 694c23e..e4b71cc 100644 --- a/tests/test_clash_detections.py +++ b/tests/test_clash_detections.py @@ -1,20 +1,39 @@ -import maple as mp +import logging from dotenv import load_dotenv -import logging +from maple.base.maple import Maple +import pytest logging.basicConfig(level=logging.DEBUG) - load_dotenv() -def test_should_clash(): +# Arrange +@pytest.fixture +def mp(): + model_id = "f4a1103c37" project_id = "21f8910cc7" - mp.init_model(project_id=project_id, model_id="f4a1103c37") + + mp = Maple( + project_id=project_id, + model_id=model_id, + ) + return mp + + +def test_should_clash(mp: Maple): + def clash_detection(): + mp.it("Clash detection between walls and windows") + windows = mp.get("category", "Windows") + walls = mp.get("category", "Walls") + + mp.detect_collision(windows, walls) + mp.run(clash_detection) results = mp.get_results() + print(results) assert len(results) == 1 result = results[0] assert result["type"] == "collision" @@ -24,12 +43,18 @@ def test_should_clash(): return -def test_should_not_clash(): - project_id = "21f8910cc7" - mp.init_model(project_id=project_id, model_id="f4a1103c37") +def test_should_not_clash(mp: Maple): + def clash_detection_pass(): + mp.it("Clash detection between windows and topography") + windows = mp.get("category", "Windows") + topo = mp.get("category", "Topography") + + mp.detect_collision(windows, topo) + mp.run(clash_detection_pass) results = mp.get_results() + print(results) assert len(results) == 1 result = results[0] assert result["type"] == "collision" @@ -37,19 +62,3 @@ def test_should_not_clash(): # assert len(result["collisions"]) == 14 # TODO: This is flaky assert len(result["collisions"]) == 0 return - - -def clash_detection(): - mp.it("Clash detection between walls and windows") - windows = mp.get("category", "Windows") - walls = mp.get("category", "Walls") - - mp.detect_collision(windows, walls) - - -def clash_detection_pass(): - mp.it("Clash detection between windows and topography") - windows = mp.get("category", "Windows") - topo = mp.get("category", "Topography") - - mp.detect_collision(windows, topo) diff --git a/tests/test_results.py b/tests/test_results.py index b487bf7..9bd7e28 100644 --- a/tests/test_results.py +++ b/tests/test_results.py @@ -1,6 +1,6 @@ -import maple as mp from dotenv import load_dotenv +from maple.base.maple import Maple from maple.ops import CompOp load_dotenv() @@ -11,8 +11,17 @@ def test_results(): stream_id = "21f8910cc7" - mp.init_model(project_id=stream_id, model_id="f4a1103c37") + mp = Maple(project_id=stream_id, model_id="f4a1103c37") + + def spec(): + mp.it(spec_name) + + mp.get("category", "Windows").where( + "speckle_type", "Objects.Other.Instance:Objects.Other.Revit.RevitInstance" + ).its("Height").should("be.greater", min_height) + mp.run(spec) + results = mp.get_results() assert len(results) == 1 result = results[0] @@ -25,11 +34,3 @@ def test_results(): assert result["spec"]["comparer"] == CompOp.BE_GREATER assert result["spec"]["value"] == min_height assert result["result"] == "pass" - - -def spec(): - mp.it(spec_name) - - mp.get("category", "Windows").where( - "speckle_type", "Objects.Other.Instance:Objects.Other.Revit.RevitInstance" - ).its("Height").should("be.greater", min_height) diff --git a/tests/test_run.py b/tests/test_run.py index cef72dc..dfcf4f2 100644 --- a/tests/test_run.py +++ b/tests/test_run.py @@ -1,19 +1,38 @@ -import maple as mp from dotenv import load_dotenv - +import pytest import logging +from maple.base.maple import Maple + logging.basicConfig(level=logging.DEBUG) load_dotenv() -def test_success_run(): - project_id = "21f8910cc7" - mp.init_model(project_id=project_id, model_id="f4a1103c37") +project_id = "21f8910cc7" + + +@pytest.fixture +def spec_fixture(): + mp = Maple(project_id=project_id, model_id="f4a1103c37") + + def spec(): + min_height = 900 + mp.it(f"checks window height is greater than {min_height} mm") + + mp.get("category", "Windows").where( + "speckle_type", "Objects.Other.Instance:Objects.Other.Revit.RevitInstance" + ).its("Height").should("be.greater", min_height) + + return mp, spec + + +def test_success_run(spec_fixture): + mp: Maple = spec_fixture[0] + spec = spec_fixture[1] mp.run(spec) - assert mp._project_id == project_id - test_case = mp.get_current_test_case() + assert mp.project_id == project_id + test_case = mp.current_test_case assert test_case is not None assert len(test_case.assertions) == 1 @@ -23,29 +42,10 @@ def test_success_run(): return -def test_error_run(): +def test_error_run(spec_fixture): + mp: Maple = spec_fixture[0] + spec = spec_fixture[1] + some = "hello" other = {"name": "i'm a function"} mp.run(spec, some, other) # type: ignore - - -def test_multiple_streams(): - stream_id = "21f8910cc7" - mp.init_model(project_id=stream_id, model_id="f4a1103c37") - mp.run(spec) - - # We set the stream id to another, it doesn't - # matter that is not valid since we will not use it to query - mp.init_model("other", "rehto") - - # Setting the stream should reset the current object - assert mp.get_current_obj() is None - - -def spec(): - min_height = 900 - mp.it(f"checks window height is greater than {min_height} mm") - - mp.get("category", "Windows").where( - "speckle_type", "Objects.Other.Instance:Objects.Other.Revit.RevitInstance" - ).its("Height").should("be.greater", min_height) diff --git a/tests/test_should_satisfy.py b/tests/test_should_satisfy.py index 4e5eec0..5f4e643 100644 --- a/tests/test_should_satisfy.py +++ b/tests/test_should_satisfy.py @@ -1,21 +1,22 @@ -import maple as mp from dotenv import load_dotenv +from maple.base.maple import Maple + load_dotenv() def test_success_run(): project_id = "21f8910cc7" - mp.init_model(project_id=project_id, model_id="f4a1103c37") - mp.run(spec) - results = mp.get_results() - assert len(results) == 1 + mp = Maple(project_id=project_id, model_id="f4a1103c37") + def spec(): + min_height = 900 + mp.it(f"checks window height is greater than {min_height} mm") -def spec(): - min_height = 900 - mp.it(f"checks window height is greater than {min_height} mm") + mp.get("category", "Windows").where( + "speckle_type", "Objects.Other.Instance:Objects.Other.Revit.RevitInstance" + ).its("Height").should_satisfy(lambda x: x > min_height) - mp.get("category", "Windows").where( - "speckle_type", "Objects.Other.Instance:Objects.Other.Revit.RevitInstance" - ).its("Height").should_satisfy(lambda x: x > min_height) + mp.run(spec) + results = mp.get_results() + assert len(results) == 1