diff --git a/README.md b/README.md index 2a1cf83..bd10de3 100644 --- a/README.md +++ b/README.md @@ -1,100 +1,138 @@ ## python-mvdxml -A mvdXML checker and w3c SPARQL converter, as an IfcOpenShell submodule or stand-alone. +An mvdXML parser, checker, and W3C SPARQL converter provided as an +IfcOpenShell submodule. -WARNING: While this repository has many useful building blocks to build software around mvdXML and IFC, there are many mvdXML dialects and not all variants are likely to be fully supported. +> [!WARNING] +> While this package has useful building blocks for mvdXML and IFC, there are +> many mvdXML dialects and not all variants are fully supported. -### Quickstart - -#### Extraction +### Parsing + +Parsed documents are immutable dataclasses. Templates and template references +are resolved while parsing, so repeated access returns the same parsed object. ```python -import ifcopenshell -from ifcopenshell.mvd import mvd +from ifcopenshell.mvd import parse + +concept_roots = parse("mvd_examples/wall_extraction.mvdxml") +concept_root = concept_roots[0] + +print(concept_root.name) +print(concept_root.entity) +print([concept.name for concept in concept_root.concepts()]) +``` + +`parse()` returns a tuple of `concept_root` objects. A document containing only +concept templates returns a tuple of `template` objects instead. Invalid XML +and missing template references raise `ValueError` with the relevant document +or template identifier. Recursive template branches are expanded once. -mvd_concept = mvd.open_mvd("examples/wall_extraction.mvdxml") -file = ifcopenshell.open("Duplex_A_20110505.ifc") +### Graphviz concept templates -all_data = mvd.get_data(mvd_concept, file, spreadsheet_export=True) +The lightweight concept graphs used by the buildingSMART IFC4.x documentation +can be read directly into the same immutable `template` representation: -non_respecting_entities = mvd.get_non_respecting_entities(file, all_data[1]) -respecting_entities = mvd.get_respecting_entities(file, all_data[1]) +~~~python +from ifcopenshell.mvd import template +source = """ +This prose is ignored. ``` +concept { + IfcObject:ObjectType -> IfcLabel + IfcObject:ObjectType[binding="UserDefinedType"] +} +``` +""" + +parsed_template = template.from_graphviz( + source, + name="Object Predefined Type", +) +~~~ + +Only `concept {}` declarations inside triple-backtick fences are read; the +surrounding Markdown is not parsed. Edges, attribute bindings, constraint nodes, +and named template references follow the syntax used by buildingSMART's +`templates_to_mvdxml.py`. Referenced templates must already be parsed and +provided by name: + +~~~python +parent = template.from_graphviz( + parent_source, + references={"Surface Color Style": surface_color_style}, +) +~~~ + +Reference names are matched without spaces or underscores. Graph parsing uses +`networkx`, available through IfcOpenShell's `advanced` optional dependencies. + +### Extraction + +Extraction returns native Python containers and IFC values. ```python -# Create a new file -new_file = ifcopenshell.file(schema=file.schema) -proj = file.by_type("IfcProject")[0] -new_file.add(proj) +import ifcopenshell + +from ifcopenshell.mvd import concept_root, parse -for e in respecting_entities: - new_file.add(e) +parsed = parse("mvd_examples/wall_extraction.mvdxml") +root = parsed[0] +assert isinstance(root, concept_root) +ifc_file = ifcopenshell.open("Duplex_A_20110505.ifc") -new_file.write("new_file.ifc") +all_data, verification = root.get_data(ifc_file) +non_respecting = root.get_non_respecting_entities(ifc_file, verification) +respecting = root.get_respecting_entities(ifc_file, verification) ``` +The result is: + ```python -# Visualize results -mvd.visualize(file, non_respecting_entities) +tuple[ + list[dict[str, object]], # one GlobalId-to-value mapping per concept + dict[str, dict[str, int]], # GlobalId-to-concept verification matrix +] ``` -##### Validation +Individual structures also expose their own behavior: -~~~py -import ifcopenshell +```python +concept = next(root.concepts()) +parsed_template = concept.template() +entity = ifc_file.by_type(root.entity)[0] -from ifcopenshell.mvd import mvd -from colorama import Fore -from colorama import Style - -concept_roots = list(ifcopenshell.mvd.concept_root.parse(MVDXML_FILENAME)) -file = ifcopenshell.open(IFC_FILENAME) - -tt = 0 # total number of tests -ts = 0 # total number of successful tests - -for concept_root in concept_roots: - print("ConceptRoot: ", concept_root.entity) - for concept in concept_root.concepts(): - tt = tt + 1 - print("Concept: ", concept.name) - try: - - if len(concept.template().rules) > 1: - attribute_rules = [] - for rule in concept.template().rules: - attribute_rules.append(rule) - rules_root = ifcopenshell.mvd.rule("EntityRule", concept_root.entity, attribute_rules) - else: - rules_root = concept.template().rules[0] - ts = ts + 1 - finst = 0 #failed instances - - for inst in file.by_type(concept_root.entity): - try: - data = mvd.extract_data(rules_root, inst) - valid, output = mvd.validate_data(concept, data) - if not valid: - finst = finst + 1 - print("[VALID]" if valid else Fore.RED +"[failure]"+Style.RESET_ALL, inst) - print(output) - except Exception as e: - print(Fore.RED+"EXCEPTION: ", e, Style.RESET_ALL,inst) - print () - print (int(finst), "out of", int(len(file.by_type(concept_root.entity))), "instances failed the check") - print ("---------------------------------") - except Exception as e: - print("EXCEPTION: "+Fore.RED,e,Style.RESET_ALL) - print("---------------------------------") - print("---------------------------------") -print("---------------------------------") - -tf = tt-ts # total number of failed tests - -print ("\nRESULTS OVERVIEW") -print ("Total number of tests: ",tt) -print ("Total number of executed tests: ", ts) -print ("Total number of failed tests: ", tf) -~~~ +extracted = parsed_template.extract(entity) +valid, report = concept.validate(extracted) +``` + +`extracted` is a list of dictionaries mapping immutable `rule` objects to IFC +values. `valid` is a boolean and `report` is a string; validation itself does +not print. + +### Visualization and export + +Visualization and spreadsheet generation are deliberately outside this +package. Use the returned GlobalIds to select or colour entities in the caller's +viewer. For CSV, JSON, dataframe, or spreadsheet output, transform `all_data` +and `verification` with the corresponding Python library. Keeping those +operations at the application boundary means importing this package does not +initialize a geometry backend or require a spreadsheet dependency. + +### Command line + +Inspect a document: + +```console +python -m ifcopenshell.mvd mvd_examples/wall_extraction.mvdxml +``` + +Generate and execute SPARQL against an IFC-OWL Turtle file: + +```console +python -m ifcopenshell.mvd model.mvdxml model.ttl +``` + +Use `python -m ifcopenshell.mvd --help` for argument details. diff --git a/__init__.py b/__init__.py index 027f341..40237ae 100644 --- a/__init__.py +++ b/__init__.py @@ -1,200 +1,4 @@ -from . import mvdxml_expression +from .model import concept_or_applicability, concept_root, rule, template +from .parser import parse -from xml.dom.minidom import parse, Element - -class rule(object): - """ - A class for representing an mvdXML EntityRule or AttributeRule - """ - parent = None - - def __init__(self, tag, attribute, nodes, bind=None, optional=False): - self.tag, self.attribute, self.nodes, self.bind = tag, attribute, nodes, bind - self.optional = optional - - def to_string(self, indent=0): - # return "%s%s%s[%s](%s%s)%s" % ("\n" if indent else "", " "*indent, self.tag, self.attribute, "".join(n.to_string(indent+2) for n in self.nodes), ("\n" + " "*indent) if len(self.nodes) else "", (" -> %s" % self.bind) if self.bind else "") - return "<%s %s%s>" % (self.tag, f"{self.bind}=" if self.bind else "", self.attribute) - - def __repr__(self): - return self.to_string() - -class template(object): - """ - Representation of an mvdXML template - """ - - def __init__(self, concept, root, constraints=None, rules=None, parent=None): - self.concept, self.root, self.constraints, self.parent = concept, root, (constraints or []), parent - self.rules = rules or [] - self.entity = str(root.attributes['applicableEntity'].value) - try: - self.name = root.attributes['name'].value - except: - self.name = None - - def bind(self, constraints): - return template(self.concept, self.root, constraints, self.rules) - - def parse(self, visited=None): - for rules in self.root.getElementsByTagNameNS("*", "Rules"): - for r in rules.childNodes: - if not isinstance(r, Element): continue - self.rules.append(self.parse_rule(r, visited=visited)) - - def traverse(self, fn, root=None, with_parents=False): - def visit(n, p=root, ps=[root]): - if with_parents: - close = fn(rule=n, parents=ps) - else: - close = fn(rule=n, parent=p) - - for s in n.nodes: - visit(s, n, ps + [n]) - - if close: - close() - - for r in self.rules: - visit(r) - - def parse_rule(self, root, visited=None): - def visit(node, prefix="", visited=None, parent=None): - r = None - n = node - nm = None - p = prefix - optional = False - visited = set() if visited is None else visited - - if node.localName == "AttributeRule": - r = node.attributes["AttributeName"].value - try: - nm = node.attributes["RuleID"].value - except: - # without binding, it's wrapped in a SPARQL OPTIONAL {} clause - # Aim is to insert this clause once as high in the stack as possible - # All topmost attribute rules are optional anyway as in the binding requirements on existence is specified - - def child_has_ruleid_or_prefix(node): - if type(node).__name__ == "Element": - if "RuleID" in node.attributes or "IdPrefix" in node.attributes: - return True - for n in node.childNodes: - if child_has_ruleid_or_prefix(n): return True - - optional = node.parentNode.localName == "Rules" or not child_has_ruleid_or_prefix(node) - elif node.localName == "EntityRule": - r = node.attributes["EntityName"].value - elif node.localName == "Template": - ref = node.attributes['ref'].value - # we break infinite recursion using this set - if ref not in visited: - n = self.concept.template(ref, visited=visited | {ref}).root - try: - p = p + node.attributes["IdPrefix"].value - except: - pass - elif node.localName == "Constraint": - r = mvdxml_expression.parse(node.attributes["Expression"].value) - elif node.localName == "EntityRules": pass - elif node.localName == "AttributeRules": pass - elif node.localName == "Rules": pass - elif node.localName == "Constraints": pass - elif node.localName == "References": pass - elif node.localName == "Definitions": return - elif node.localName == "SubTemplates": return # @todo perhaps just traverse them? - else: - raise ValueError(node.localName) - - def _(n): - for subnode in n.childNodes: - if not isinstance(subnode, Element): continue - for x in visit(subnode, p, visited=visited): yield x - - if r: - R = rule(node.localName, r, list(_(n)), (p + nm) if nm else nm, optional=optional) - for rr in R.nodes: - rr.parent = R - yield R - else: - for subnode in n.childNodes: - if not isinstance(subnode, Element): continue - for x in visit(subnode, p, visited=visited): yield x - - return list(visit(root, visited=visited))[0] - -class concept_or_applicability(object): - """ - Representation of either a mvdXML Concept or the Applicability node. Basically a structure - for the hierarchical TemplateRule - """ - - def __init__(self, root, c): - self.root = root - self.concept_node = c - try: - self.name = c.attributes["name"].value - except: - # probably applicability and not concept - self.name = "Applicability" - - def template(self, id=None, visited=None): - if id is None: - id = self.concept_node.getElementsByTagNameNS("*","Template")[0].attributes['ref'].value - - for node in self.root.dom.getElementsByTagNameNS('*',"ConceptTemplate"): - if node.attributes["uuid"].value == id: - t = template(self, node) - t.parse(visited=visited) - t_with_rules = t.bind(self.rules()) - return t_with_rules - - def rules(self): - # Get the top most TemplateRule and traverse - try: - rules = self.concept_node.getElementsByTagNameNS("*","TemplateRules")[0] - except: - return [] - - def visit(rules): - def _(): - for i, r in enumerate([c for c in rules.childNodes if isinstance(c, Element)]): - if i: - yield rules.attributes["operator"].value - if r.localName == "TemplateRules": - yield visit(r) - elif r.localName == "TemplateRule": - yield mvdxml_expression.parse(r.attributes["Parameters"].value) - else: - raise Exception() - - return list(_()) - - return visit(rules) - -class concept_root(object): - def __init__(self, dom, root): - self.dom, self.root = dom, root - self.name = root.attributes['name'].value - self.entity = str(root.attributes['applicableRootEntity'].value) - - def applicability(self): - return concept_or_applicability(self, self.root.getElementsByTagNameNS("*","Applicability")[0]) - - def concepts(self): - for c in self.root.getElementsByTagNameNS("*","Concept"): - yield concept_or_applicability(self, c) - - @staticmethod - def parse(fn): - dom = parse(fn) - if len(dom.getElementsByTagNameNS("*","ConceptRoot")): - for root in dom.getElementsByTagNameNS("*","ConceptRoot"): - CR = concept_root(dom, root) - yield CR - else: - for templ in dom.getElementsByTagNameNS("*","ConceptTemplate"): - t = template(None, templ) - t.parse() - yield t +__all__ = ["concept_or_applicability", "concept_root", "parse", "rule", "template"] diff --git a/__main__.py b/__main__.py index e0f84b2..f4cb4b8 100644 --- a/__main__.py +++ b/__main__.py @@ -1,35 +1,70 @@ -from __future__ import print_function +from __future__ import annotations + +import argparse +from collections.abc import Sequence + +from .model import concept_root, template +from .parser import parse + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="python -m ifcopenshell.mvd", + description="Inspect mvdXML or execute its generated SPARQL against an IFC-OWL file.", + ) + parser.add_argument("mvdxml", help="Path to an mvdXML document") + parser.add_argument( + "ifcowl", + nargs="?", + help="Optional path to an IFC-OWL Turtle file; omit it to inspect the mvdXML", + ) + return parser + + +def inspect_mvd(filename: str) -> None: + for item in parse(filename): + if isinstance(item, template): + print(item.name or item.entity) + _print_template(item) + continue + + for concept in item.concepts(): + print(concept.name) + print() + _print_template(concept.template()) + print() + + +def _print_template(parsed_template: template) -> None: + def dump(rule, parents) -> None: + print(" " * len(parents), rule.tag, rule.attribute) + + print("RootEntity", parsed_template.entity) + parsed_template.traverse(dump, with_parents=True) + print(" ".join(map(str, parsed_template.constraints))) + + +def execute_sparql(parsed_root: concept_root, mvdxml: str, ifcowl: str) -> None: + from . import sparql + + sparql.derive_prefix(ifcowl) + inferred_ifcowl = sparql.infer_subtypes(ifcowl) + print(sparql.executor.run(parsed_root, mvdxml, inferred_ifcowl), end="") + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + if args.ifcowl is None: + inspect_mvd(args.mvdxml) + return 0 + + roots = [item for item in parse(args.mvdxml) if isinstance(item, concept_root)] + if not roots: + raise ValueError(f"mvdXML document {args.mvdxml!r} contains no ConceptRoot") + for root in roots: + execute_sparql(root, args.mvdxml, args.ifcowl) + return 0 + if __name__ == "__main__": - import sys - from . import concept_root - - if len(sys.argv) == 2: - mvdfn = sys.argv[1] - for mvd in concept_root.parse(mvdfn): - - def dump(rule, parents): - print(" " * len(parents), rule.tag, rule.attribute) - - for c in mvd.concepts(): - print(c.name) - print() - - t = c.template() - print("RootEntity", t.entity) - t.traverse(dump, with_parents=True) - print(" ".join(map(str, t.constraints))) - - print() - - elif len(sys.argv) == 3: - from . import sparql - mvdfn,ttlfn = sys.argv[1:] - sparql.derive_prefix(ttlfn) - ttlfn = sparql.infer_subtypes(ttlfn) - for mvd in concept_root.parse(mvdfn): - sparql.executor.run(mvd, mvdfn, ttlfn) - - else: - print(sys.executable, "ifcopenshell.mvd", "<.mvdxml>") - print(sys.executable, "ifcopenshell.mvd", "<.mvdxml>", "<.ifc>") + raise SystemExit(main()) diff --git a/model.py b/model.py new file mode 100644 index 0000000..8a5c64b --- /dev/null +++ b/model.py @@ -0,0 +1,376 @@ +# This file was generated with the assistance of an AI coding tool. + +from __future__ import annotations + +import ast +import io +import itertools +import operator +from dataclasses import dataclass +from functools import partial, reduce +from typing import Any, Callable, Iterable, Iterator, Mapping + +import ifcopenshell + +extracted_data = list[dict["rule", Any]] +concept_data = dict[str, Any] +verification_matrix = dict[str, dict[str, int]] + + +def _parse_mvdxml_token(value: str) -> Any: + if value.lower() == "true": + return True + if value.lower() == "false": + return False + return ast.literal_eval(value) + + +def _merge_dictionaries(dicts: Iterable[dict[rule, Any]]) -> dict[rule, Any]: + result: dict[rule, Any] = {} + for value in dicts: + result.update(value) + return result + + +def _format_export_value(value: Any) -> Any: + if isinstance(value, ifcopenshell.entity_instance): + return getattr(value, "GlobalId", None) or str(value) + return value + + +def _format_data_from_nodes(recurse_output: extracted_data) -> Any: + if len(recurse_output) > 1: + return [ + value + for resulting_dict in recurse_output + for value in resulting_dict.values() + ] + + if len(recurse_output) == 1: + values = list(recurse_output[0].values()) + if len(values) > 1: + for value in values: + if not isinstance(value, str): + return _format_export_value(value) + return list(map(_format_export_value, values)) + return _format_export_value(values[0]) + + return [] + + +@dataclass(frozen=True, eq=False) +class rule: + """An immutable mvdXML EntityRule, AttributeRule, or Constraint.""" + + tag: str + attribute: Any + nodes: tuple[rule, ...] = () + bind: str | None = None + optional: bool = False + + def __post_init__(self) -> None: + object.__setattr__(self, "nodes", tuple(self.nodes)) + + def to_string(self, indent: int = 0) -> str: + return "<%s %s%s>" % ( + self.tag, + f"{self.bind}=" if self.bind else "", + self.attribute, + ) + + def __repr__(self) -> str: + return self.to_string() + + def extract(self, ifc_data: ifcopenshell.entity_instance | Any) -> extracted_data: + """Extract values from one IFC entity or IFC attribute value.""" + + if not self.nodes: + if self.tag == "AttributeRule": + try: + value = getattr(ifc_data, self.attribute) + except (AttributeError, TypeError): + return [{self: "Invalid Attribute"}] + return [{self: value}] + if ( + self.tag == "EntityRule" + and isinstance(ifc_data, ifcopenshell.entity_instance) + and not ifc_data.is_a(self.attribute) + ): + return [] + return [{self: ifc_data}] + + if self.tag == "AttributeRule": + try: + values_from_attribute = getattr(ifc_data, self.attribute) + except (AttributeError, TypeError): + return [{self: "Invalid attribute rule"}] + + if values_from_attribute is None: + return [{self: "Nonexistent value"}] + + if isinstance(values_from_attribute, (list, tuple)): + if not values_from_attribute: + return [{self: "empty data structure"}] + values = values_from_attribute + else: + values = (values_from_attribute,) + + return [ + child_value + for child in self.nodes + for value in values + for child_value in child.extract(value) + ] + + if self.tag == "EntityRule": + if ( + self.nodes + and isinstance(ifc_data, ifcopenshell.entity_instance) + and not ifc_data.is_a(self.attribute) + ): + return [] + + to_combine: list[extracted_data] = [] + for child in self.nodes: + if child.tag == "Constraint": + on_node = child.attribute[0][0].c.replace("'", "") + if isinstance(ifc_data, ifcopenshell.entity_instance): + typed_node = type(ifc_data[0])(on_node) + if ifc_data[0] == typed_node: + return [{self: ifc_data}] + elif ifc_data == on_node: + return [{self: ifc_data}] + else: + to_combine.append(child.extract(ifc_data)) + + if to_combine: + return list(map(_merge_dictionaries, itertools.product(*to_combine))) + + return [] + + +@dataclass(frozen=True) +class template: + """An immutable, fully parsed mvdXML concept template.""" + + entity: str + name: str | None + rules: tuple[rule, ...] = () + constraints: tuple[Any, ...] = () + + def __post_init__(self) -> None: + object.__setattr__(self, "rules", tuple(self.rules)) + object.__setattr__(self, "constraints", tuple(self.constraints)) + + @classmethod + def from_graphviz( + cls, + source: str, + *, + name: str | None = None, + references: Mapping[str, template] | None = None, + ) -> template: + """Parse a fenced buildingSMART concept graph into a template.""" + + from .graphviz import parse + + return parse(source, name=name, references=references) + + def traverse( + self, + fn: Callable[..., Callable[[], None] | None], + root: rule | None = None, + with_parents: bool = False, + ) -> None: + def visit( + node: rule, parent: rule | None, parents: tuple[rule | None, ...] + ) -> None: + if with_parents: + close = fn(rule=node, parents=parents) + else: + close = fn(rule=node, parent=parent) + + for child in node.nodes: + visit(child, node, parents + (node,)) + + if close: + close() + + for top_level_rule in self.rules: + visit(top_level_rule, root, (root,)) + + def root_rule(self) -> rule: + if not self.rules: + raise ValueError(f"Template {self.name or self.entity!r} contains no rules") + if len(self.rules) == 1: + return self.rules[0] + return rule("EntityRule", self.entity, self.rules) + + def extract(self, ifc_data: ifcopenshell.entity_instance | Any) -> extracted_data: + return self.root_rule().extract(ifc_data) + + def binding_for(self, target: rule) -> str | None: + binding: str | None = None + + def find(rule: rule, parent: rule | None) -> None: + nonlocal binding + if rule is target: + binding = rule.bind or (parent.bind if parent else None) + + self.traverse(find) + return binding + + +@dataclass(frozen=True) +class concept_or_applicability: + """An immutable parsed Concept or Applicability definition.""" + + name: str + parsed_template: template + template_rules: tuple[Any, ...] = () + root_name: str | None = None + root_entity: str | None = None + is_root_applicability: bool = False + + def __post_init__(self) -> None: + object.__setattr__(self, "template_rules", tuple(self.template_rules)) + + def template(self) -> template: + return self.parsed_template + + def rules(self) -> tuple[Any, ...]: + return self.template_rules + + def extract( + self, entities: Iterable[ifcopenshell.entity_instance], filtering: bool = False + ) -> concept_data: + extracted_entities_data: concept_data = {} + for entity in entities: + output = _format_data_from_nodes(self.parsed_template.extract(entity)) + if not filtering or output: + extracted_entities_data[entity.GlobalId] = output + return extracted_entities_data + + def validate(self, data: extracted_data) -> tuple[bool, str]: + rules = [value[0] for value in self.rules() if not isinstance(value, str)] + + def transform_data(values: dict[rule, Any]) -> dict[str | None, Any]: + return { + self.parsed_template.binding_for(key): value + for key, value in values.items() + } + + transformed_data = list(map(transform_data, data)) + output = io.StringIO() + + def operation_reduce(x: Any, y: Any) -> Any: + if callable(x): + return x(y) + return partial(y, x) + + def apply_rules() -> Iterator[bool]: + for expression in rules: + + def apply_data() -> Iterator[bool]: + for values in transformed_data: + + def translate(value: Any) -> Any: + if isinstance(value, str): + return getattr(operator, value.lower() + "_") + if value.b in {"Value", None}: + return values.get(value.a) == _parse_mvdxml_token( + value.c + ) + if value.b == "Type": + item = values.get(value.a) + return bool( + item is not None + and item.is_a(_parse_mvdxml_token(value.c)) + ) + if value.b == "Exists": + return ( + values.get(value.a) is not None + ) == _parse_mvdxml_token(value.c) + raise ValueError( + f"Unsupported template rule operand: {value.b}" + ) + + yield reduce(operation_reduce, map(translate, expression)) + + valid = any(apply_data()) + print(("Met:" if valid else "Not met:"), expression, file=output) + yield valid + + valid = all(apply_rules()) + return valid, output.getvalue() + + +@dataclass(frozen=True) +class concept_root: + """An immutable mvdXML ConceptRoot and its fully parsed concepts.""" + + name: str + entity: str + parsed_concepts: tuple[concept_or_applicability, ...] = () + parsed_applicability: concept_or_applicability | None = None + + def __post_init__(self) -> None: + object.__setattr__(self, "parsed_concepts", tuple(self.parsed_concepts)) + + def applicability(self) -> concept_or_applicability: + if self.parsed_applicability is None: + raise ValueError(f"Concept root {self.name!r} has no Applicability") + return self.parsed_applicability + + def concepts(self) -> Iterator[concept_or_applicability]: + return iter(self.parsed_concepts) + + def get_data( + self, ifc_file: ifcopenshell.file + ) -> tuple[list[concept_data], verification_matrix]: + entities = list(ifc_file.by_type(self.entity)) + selected_entities = entities + verification: verification_matrix = {entity.GlobalId: {} for entity in entities} + concepts = sorted( + self.parsed_concepts, + key=lambda concept: concept.name.startswith("AP"), + reverse=True, + ) + all_data: list[concept_data] = [] + + for concept in concepts: + filtering = concept.name.startswith("AP") + extracted = concept.extract(selected_entities, filtering=filtering) + all_data.append(extracted) + + if filtering: + selected_ids = set(extracted) + selected_entities = [ + entity + for entity in selected_entities + if entity.GlobalId in selected_ids + ] + for entity in entities: + verification[entity.GlobalId][concept.name] = int( + entity.GlobalId not in selected_ids + ) + + return all_data, verification + + def get_non_respecting_entities( + self, ifc_file: ifcopenshell.file, verification: verification_matrix + ) -> list[ifcopenshell.entity_instance]: + return [ + ifc_file.by_guid(global_id) + for global_id, values in verification.items() + if sum(values.values()) != 0 + ] + + def get_respecting_entities( + self, ifc_file: ifcopenshell.file, verification: verification_matrix + ) -> list[ifcopenshell.entity_instance]: + return [ + ifc_file.by_guid(global_id) + for global_id, values in verification.items() + if sum(values.values()) == 0 + ] diff --git a/mvd.py b/mvd.py deleted file mode 100644 index 89dc5fd..0000000 --- a/mvd.py +++ /dev/null @@ -1,500 +0,0 @@ -import ifcopenshell -import ifcopenshell.geom - -import os -import itertools - -import csv -import xlsxwriter - - -def is_applicability(concept): - """ - Check whether the Concept created has a filtering purpose. - Actually, MvdXML has a specific Applicability node. - - :param concept: mvdXML Concept object - """ - return concept.name.startswith("AP") - - -def merge_dictionaries(dicts): - d = {} - for e in dicts: - d.update(e) - return d - - -def extract_data(mvd_node, ifc_data): - """ - Recursively traverses mvdXML Concept tree structure. - This tree is made of different mvdXML Rule nodes: AttributesRule - and EntityRule. - - :param mvd_node: an mvdXML Concept - :param ifc_data: an IFC instance or an IFC value - - - """ - to_combine = [] - return_value = [] - - if len(mvd_node.nodes) == 0: - if mvd_node.tag == "AttributeRule": - try: - values_from_attribute = getattr(ifc_data, mvd_node.attribute) - return [{mvd_node: values_from_attribute}] - except: - return [{mvd_node: "Invalid Attribute"}] - - else: - return [{mvd_node: ifc_data}] - - if mvd_node.tag == 'AttributeRule': - data_from_attribute = [] - try: - values_from_attribute = getattr(ifc_data, mvd_node.attribute) - if values_from_attribute is None: - return [{mvd_node:"Nonexistent value"}] - - except: - return [{mvd_node:"Invalid attribute rule"}] - - - if isinstance(values_from_attribute, (list, tuple)): - if len(values_from_attribute) == 0: - return [{mvd_node: 'empty data structure'}] - data_from_attribute.extend(values_from_attribute) - - else: - data_from_attribute.append(values_from_attribute) - - for child in mvd_node.nodes: - for data in data_from_attribute: - child_values = extract_data(child, data) - if isinstance(child_values, (list, tuple)): - return_value.extend(child_values) - else: - return_value.append(child_values) - return return_value - - elif mvd_node.tag == 'EntityRule': - # Avoid things like Quantities on Psets - if len(mvd_node.nodes): - if isinstance(ifc_data, ifcopenshell.entity_instance) and not ifc_data.is_a(mvd_node.attribute): - return [] - - for child in mvd_node.nodes: - if child.tag == "Constraint": - on_node = child.attribute[0].c - on_node = on_node.replace("'", "") - if isinstance(ifc_data, ifcopenshell.entity_instance): - ifc_type = type(ifc_data[0]) - typed_node = (ifc_type)(on_node) - - if ifc_data[0] == typed_node: - return [{mvd_node: ifc_data}] - - elif ifc_data == on_node: - return [{mvd_node: ifc_data}] - else: - to_combine.append(extract_data(child, ifc_data)) - - if len(to_combine): - return_value = list(map(merge_dictionaries, itertools.product(*to_combine))) - - return return_value - - -def open_mvd(filename): - """ - Open an mvdXML file. - - :param filename: Path of the mvdXML file. - :return: mvdXML Concept instance. - """ - my_concept_object = list(ifcopenshell.mvd.concept_root.parse(filename))[0] - return my_concept_object - - -def format_data_from_nodes(recurse_output): - """ - Enable to format data collected such that the value to be exported is extracted. - - :param recurse_output: Data extracted from the recursive function - - """ - if len(recurse_output) > 1: - output = [] - for resulting_dict in recurse_output: - intermediate_storing = [] - for value in resulting_dict.values(): - intermediate_storing.append(value) - output.extend(intermediate_storing) - return output - - elif len(recurse_output) == 1: - return_list = [] - intermediate_list = list(recurse_output[0].values()) - if len(intermediate_list) > 1: - returned_value = intermediate_list - for element in intermediate_list: - # In case of a property that comes with all its path - # (like ['PSet_WallCommon, 'IsExternal', IfcBoolean(.F.) - # return only the list element which is not of string type - # todo: check above condition with ifcopenshell type - if not isinstance(element, str): - returned_value = element - if returned_value != intermediate_list: - return returned_value - else: - return intermediate_list - else: - return intermediate_list[0] - - else: - return [] - - -def get_data_from_mvd(entities, tree, filtering=False): - """ - Apply the recursive function on the entities to return - the values extracted. - - :param entities: IFC instances to be processed. - :param tree: mvdXML Concept instance tree root. - :param filtering: Indicates whether the mvdXML tree is an applicability. - - """ - filtered_entities = [] - extracted_entities_data = {} - - for entity in entities: - entity_id = entity.GlobalId - combinations = extract_data(tree, entity) - desired_results = [] - - for dictionary in combinations: - desired_results.append(dictionary) - - output = format_data_from_nodes(desired_results) - - if filtering: - if len(output): - extracted_entities_data[entity_id] = output - else: - extracted_entities_data[entity_id] = output - - return extracted_entities_data - - -def correct_for_export(all_data): - """ - Process the data for spreadsheet export. - """ - for d in all_data: - for k, v in d.items(): - if isinstance(v, list) or isinstance(v, tuple): - if len(v): - new_list = [] - for data in v: - new_list.append(str(data)) - d[k] = ','.join(new_list) - if len(v) == 0: - d[k] = 0 - - elif isinstance(v, ifcopenshell.entity_instance): - if g := getattr(v, 'GlobalId', None): - d[k] = g - else: - d[k] = str(v) - return all_data - - -def export_to_xlsx(xlsx_name, concepts, all_data): - """ - Export data towards XLSX spreadsheet format. - - :param xlsx_name: Name of the outputted file. - :param concepts: List of mvdXML Concept instances. - :param all_data: Data extracted. - - """ - - if not os.path.isdir("spreadsheet_output/"): - os.mkdir("spreadsheet_output/") - - workbook = xlsxwriter.Workbook("spreadsheet_output/" + xlsx_name) - worksheet = workbook.add_worksheet() - # Formats - bold_format = workbook.add_format() - bold_format.set_bold() - bold_format.set_center_across() - # Write first row - column_index = 0 - for concept in concepts: - worksheet.write(0, column_index, concept.name, bold_format) - column_index += 1 - - col = 0 - for feature in all_data: - row = 1 - for d in feature.values(): - worksheet.write(row, col, d) - row += 1 - col += 1 - - workbook.close() - - -def export_to_csv(csv_name, concepts, all_data): - """ - Export data towards CSV spreadsheet format. - - :param csv_name: Name of the file outputted file. - :param concepts: List of mvdXML Concept instances. - :param all_data: Data extracted. - """ - - if not os.path.isdir("spreadsheet_output/"): - os.mkdir("spreadsheet_output/") - - with open('spreadsheet_output/' + csv_name, 'w', newline='') as f: - writer = csv.writer(f) - header = [concept.name for concept in concepts] - first_row = writer.writerow(header) - - values_by_row = [] - for val in all_data: - values_by_row.append(list(val.values())) - entities_number = len(all_data[0].keys()) - for i in range(0, entities_number): - row_to_write = [] - for r in values_by_row: - row_to_write.append(r[i]) - - f = writer.writerow(row_to_write) - - -def get_data(mvd_concept, ifc_file, spreadsheet_export=True): - """ - Use the majority of all the other functions to return the data - queried by the mvdXML file in python format. - - :param mvd_concept: mvdXML Concept instance. - :param ifc_file: IFC file from any schema. - :param spreadsheet_export: The spreadsheet export is carried out when set to True. - - - - """ - - # Check if IFC entities have been filtered at least once - filtered = 0 - - entities = ifc_file.by_type(mvd_concept.entity) - selected_entities = entities - verification_matrix = {} - for entity in selected_entities: - verification = dict() - verification_matrix[entity.GlobalId] = verification - - # For each Concept(ConceptTemplate) in the ConceptRoot - concepts = sorted(mvd_concept.concepts(), key=is_applicability, reverse=True) - all_data = [] - counter = 0 - for concept in concepts: - if is_applicability(concept): - filtering = True - else: - filtering = False - - # Access all the Rules of the ConceptTemplate - if len(concept.template().rules) > 1: - attribute_rules = [] - for rule in concept.template().rules: - attribute_rules.append(rule) - rules_root = ifcopenshell.mvd.rule("EntityRule", mvd_concept.entity, attribute_rules) - else: - rules_root = concept.template().rules[0] - - - extracted_data = get_data_from_mvd(selected_entities, rules_root, filtering=filtering) - all_data.append(extracted_data) - - if filtering: - filtered = 1 - new_entities = [] - for entity_id in all_data[counter].keys(): - if len(all_data[counter][entity_id]) != 0: - entity = ifc_file.by_id(entity_id) - new_entities.append(entity) - - selected_entities = new_entities - not_respecting_entities = [item for item in entities if item not in selected_entities] - for entity in entities: - val = 0 - if entity in not_respecting_entities: - val = 1 - verification_matrix[entity.GlobalId].update({concept.name: val}) - counter += 1 - - all_data = correct_for_export(all_data) - - if spreadsheet_export: - if filtered != 0: - export_name = "output_filtered" - else: - export_name = "output_non_filtered" - export_to_xlsx(export_name + '.xlsx', concepts, all_data) - export_to_csv(export_name + '.csv', concepts, all_data) - - - return all_data, verification_matrix - - -def get_non_respecting_entities(file, verification_matrix): - non_respecting = [] - for k, v in verification_matrix.items(): - entity = file.by_id(k) - print(list(v.values())) - if sum(v.values()) != 0: - non_respecting.append(entity) - - return non_respecting - - - - -def get_respecting_entities(file, verification_matrix): - respecting = [] - for k, v in verification_matrix.items(): - entity = file.by_id(k) - print(list(v.values())) - if sum(v.values()) == 0: - respecting.append(entity) - - return respecting - - -def visualize(file, not_respecting_entities): - """ - Visualize the instances of the entity type targeted by the mvdXML ConceptRoot. - At display, a color differentiation is made between the entities which comply with - mvdXML requirements and the ones which don't. - - :param file: IFC file from any schema. - :param not_respecting_entities: Entities which don't comply with mvdXML requirements. - - """ - - s = ifcopenshell.geom.main.settings() - s.set(s.USE_PYTHON_OPENCASCADE, True) - s.set(s.DISABLE_OPENING_SUBTRACTIONS, False) - - viewer = ifcopenshell.geom.utils.initialize_display() - - entity_type = not_respecting_entities[0].is_a() - - other_entities = [x for x in file.by_type("IfcBuildingElement") if x.is_a() != str(entity_type)] - - set_of_entities = set(not_respecting_entities) | set(file.by_type(entity_type)) - set_to_display = set_of_entities.union(set(other_entities)) - - for el in set_to_display: - if el in not_respecting_entities: - c = (1, 0, 0, 1) - elif el in other_entities: - c = (1, 1, 1, 0) - else: - c = (0, 1, 0.5, 1) - - try: - shape = ifcopenshell.geom.create_shape(s, el) - # OCC.BRepTools.breptools_Write(shape.geometry, "test.brep") - ds = ifcopenshell.geom.utils.display_shape(shape, clr=c) - except: - pass - - viewer.FitAll() - - ifcopenshell.geom.utils.main_loop() - - -def validate_data(concept, data): - import io - import ast - import operator - from functools import reduce, partial - - rules = [x[0] for x in concept.rules() if not isinstance(x, str)] - - def transform_data(d): - """ - Transform dictionary keys from tree nodes to rule ids - """ - - return {(k.parent if k.bind is None and (k.parent is not None and k.parent.bind is not None) else k).bind: v for k, v in d.items()} - - - def parse_mvdxml_token(v): - if v.lower() == "true": - return True - if v.lower() == "false": - return False - # @todo make more permissive and tolerant - return ast.literal_eval(v) - - - data = list(map(transform_data, data)) - - output = io.StringIO() - - # https://stackoverflow.com/a/70227259 - def operation_reduce(x, y): - """ - Takes alternating value and function as input and - reduces while applying function - """ - - if callable(x): - return x(y) - else: - return partial(y, x) - - - def apply_rules(): - - for r in rules: - - def apply_data(): - - for d in data: - - def translate(v): - if isinstance(v, str): - return getattr(operator, v.lower() + "_") - else: - if v.b == "Value" or v.b is None: - return d.get(v.a) == parse_mvdxml_token(v.c) - elif v.b == "Type": - return d.get(v.a) is not None and d.get(v.a).is_a(parse_mvdxml_token(v.c)) - elif v.b == "Exists": - return (d.get(v.a) is not None) == parse_mvdxml_token(v.c) - else: - raise RuntimeError(f"Invalid rule predicate {v.b}") - - r2 = list(map(translate, r)) - yield reduce(operation_reduce, r2) - - v = any(list(apply_data())) - print(("Met:" if v else "Not met:"), r, file=output) - yield v - - - valid = all(list(apply_rules())) - return valid, output.getvalue() - - -if __name__ == '__main__': - print('functions to parse MVD rules and extract IFC data/filter IFC entities from them') diff --git a/mvdxml_expression.py b/mvdxml_expression.py index cc008aa..ebefa6a 100644 --- a/mvdxml_expression.py +++ b/mvdxml_expression.py @@ -1,29 +1,46 @@ +from __future__ import annotations + +from dataclasses import dataclass + import pyparsing as pp -class node(object): - def __init__(self, args): - if len(args) == 3 and args[1] == '=': - self.a, self.b, self.c = args[0], None, args[2] - elif (args[1], args[3], args[4]) == ('[', ']', '='): - self.a, self.b, self.c = args[0], args[2], args[5] - else: - self.a, self.b, self.c = None, args[1], args[4] - def __repr__(self): return "{%s[%s]=%s}" % (self.a, self.b, self.c) +@dataclass(frozen=True) +class node: + a: str | None + b: str | None + c: str + + @classmethod + def from_tokens(cls, args: pp.ParseResults) -> node: + if len(args) == 3 and args[1] == "=": + return cls(args[0], None, args[2]) + if (args[1], args[3], args[4]) == ("[", "]", "="): + return cls(args[0], args[2], args[5]) + return cls(None, args[1], args[4]) -word = pp.Word(pp.alphanums+"_"+" "+"/"+"#") + def __repr__(self): + return "{%s[%s]=%s}" % (self.a, self.b, self.c) + + +word = pp.Word(pp.alphanums + "_" + " " + "/" + "#") quoted = pp.Combine("'" + word + "'") bool_value = pp.CaselessLiteral("TRUE") | pp.CaselessLiteral("FALSE") ref_val = word + "[" + word + "]" rhs = quoted | bool_value | ref_val | word -stmt = (pp.Optional(word) + pp.Optional("[" + word + "]") + "=" + rhs).setParseAction(node) -bool_op = pp.CaselessLiteral("AND") | pp.CaselessLiteral("OR") +stmt = (pp.Optional(word) + pp.Optional("[" + word + "]") + "=" + rhs).setParseAction( + node.from_tokens +) +bool_op = pp.CaselessLiteral("AND") | pp.CaselessLiteral("OR") grammar = stmt + pp.Optional(pp.OneOrMore(bool_op + stmt)) -def parse(exprs): - def _(): - for expr in exprs.split(";"): - expr = "".join(c for c in expr if c not in "\r\n") - if not expr: continue - yield grammar.parseString(expr) - return list(_()) + +def parse(exprs: str) -> tuple[tuple[node | str, ...], ...]: + parsed: list[tuple[node | str, ...]] = [] + for expression in exprs.split(";"): + expression = "".join( + character for character in expression if character not in "\r\n" + ) + if expression: + parsed.append(tuple(grammar.parseString(expression, parseAll=True))) + return tuple(parsed) diff --git a/parser.py b/parser.py new file mode 100644 index 0000000..e0ed36c --- /dev/null +++ b/parser.py @@ -0,0 +1,237 @@ +# This file was generated with the assistance of an AI coding tool. + +from __future__ import annotations + +import os +from dataclasses import replace +from typing import Any +from xml.dom import minidom +from xml.dom.minidom import Document, Element + +from . import mvdxml_expression +from .model import concept_or_applicability, concept_root, rule, template + + +def _elements(node: Element) -> list[Element]: + return [child for child in node.childNodes if isinstance(child, Element)] + + +class _parser: + def __init__(self, dom: Document): + self.dom = dom + self.templates = { + node.attributes["uuid"].value: node + for node in dom.getElementsByTagNameNS("*", "ConceptTemplate") + } + + def parse_template( + self, + template_id: str, + constraints: tuple[Any, ...] = (), + visited: frozenset[str] = frozenset(), + ) -> template: + try: + root = self.templates[template_id] + except KeyError as error: + raise ValueError( + f"Unknown ConceptTemplate reference: {template_id}" + ) from error + + if template_id in visited: + raise ValueError(f"Recursive ConceptTemplate reference: {template_id}") + + parsed_rules: list[rule] = [] + next_visited = visited | {template_id} + for rules_node in root.getElementsByTagNameNS("*", "Rules"): + for node in _elements(rules_node): + parsed_rules.append(self.parse_rule(node, visited=next_visited)) + + return template( + entity=str(root.attributes["applicableEntity"].value), + name=root.attributes["name"].value if "name" in root.attributes else None, + rules=tuple(parsed_rules), + constraints=constraints, + ) + + def parse_rule( + self, + root: Element, + prefix: str = "", + visited: frozenset[str] = frozenset(), + ) -> rule: + parsed = self._visit_rule(root, prefix, visited) + if len(parsed) != 1: + raise ValueError( + f"Expected one rule below {root.localName}, found {len(parsed)}" + ) + return parsed[0] + + def _visit_rule( + self, + node: Element, + prefix: str, + visited: frozenset[str], + ) -> tuple[rule, ...]: + attribute: Any = None + bind: str | None = None + optional = False + target = node + child_prefix = prefix + + if node.localName == "AttributeRule": + attribute = node.attributes["AttributeName"].value + bind = ( + node.attributes["RuleID"].value if "RuleID" in node.attributes else None + ) + if bind is None: + optional = ( + node.parentNode.localName == "Rules" + or not self._child_has_rule_id_or_prefix(node) + ) + elif node.localName == "EntityRule": + attribute = node.attributes["EntityName"].value + elif node.localName == "Template": + reference = node.attributes["ref"].value + if reference in visited: + return () + try: + target = self.templates[reference] + except KeyError as error: + raise ValueError( + f"Unknown ConceptTemplate reference: {reference}" + ) from error + if "IdPrefix" in node.attributes: + child_prefix += node.attributes["IdPrefix"].value + visited = visited | {reference} + elif node.localName == "Constraint": + attribute = mvdxml_expression.parse(node.attributes["Expression"].value) + elif node.localName in { + "EntityRules", + "AttributeRules", + "Rules", + "Constraints", + "References", + }: + pass + elif node.localName in {"Definitions", "SubTemplates"}: + return () + else: + raise ValueError(f"Unsupported mvdXML rule element: {node.localName}") + + children = tuple( + parsed + for child in _elements(target) + for parsed in self._visit_rule(child, child_prefix, visited) + ) + + if attribute is not None: + return ( + rule( + tag=node.localName, + attribute=attribute, + nodes=children, + bind=child_prefix + bind if bind else None, + optional=optional, + ), + ) + return children + + def _child_has_rule_id_or_prefix(self, node: Element) -> bool: + if "RuleID" in node.attributes or "IdPrefix" in node.attributes: + return True + return any( + self._child_has_rule_id_or_prefix(child) for child in _elements(node) + ) + + def parse_template_rules(self, concept_node: Element) -> tuple[Any, ...]: + template_rules = concept_node.getElementsByTagNameNS("*", "TemplateRules") + if not template_rules: + return () + + def visit(rules_node: Element) -> tuple[Any, ...]: + output: list[Any] = [] + for index, child in enumerate(_elements(rules_node)): + if index: + output.append(rules_node.attributes["operator"].value) + if child.localName == "TemplateRules": + output.append(visit(child)) + elif child.localName == "TemplateRule": + output.append( + mvdxml_expression.parse(child.attributes["Parameters"].value) + ) + else: + raise ValueError( + f"Unsupported TemplateRules element: {child.localName}" + ) + return tuple(output) + + return visit(template_rules[0]) + + def parse_concept( + self, + concept_node: Element, + root_name: str, + root_entity: str, + is_applicability: bool = False, + ) -> concept_or_applicability: + template_nodes = concept_node.getElementsByTagNameNS("*", "Template") + if not template_nodes: + raise ValueError(f"{concept_node.localName} contains no Template reference") + + template_id = template_nodes[0].attributes["ref"].value + constraints = self.parse_template_rules(concept_node) + parsed_template = replace( + self.parse_template(template_id), constraints=constraints + ) + return concept_or_applicability( + name=( + concept_node.attributes["name"].value + if "name" in concept_node.attributes + else "Applicability" + ), + parsed_template=parsed_template, + template_rules=constraints, + root_name=root_name, + root_entity=root_entity, + is_root_applicability=is_applicability, + ) + + def parse_root(self, root: Element) -> concept_root: + name = root.attributes["name"].value + entity = str(root.attributes["applicableRootEntity"].value) + applicability_nodes = root.getElementsByTagNameNS("*", "Applicability") + applicability = ( + self.parse_concept( + applicability_nodes[0], name, entity, is_applicability=True + ) + if applicability_nodes + else None + ) + concepts = tuple( + self.parse_concept(node, name, entity) + for node in root.getElementsByTagNameNS("*", "Concept") + ) + return concept_root(name, entity, concepts, applicability) + + +def parse(source: str | os.PathLike[str]) -> tuple[concept_root | template, ...]: + """Parse an mvdXML document into immutable model objects.""" + + try: + dom = minidom.parse(os.fspath(source)) + except Exception as error: + raise ValueError( + f"Unable to parse mvdXML document {os.fspath(source)!r}: {error}" + ) from error + + parser = _parser(dom) + roots = dom.getElementsByTagNameNS("*", "ConceptRoot") + if roots: + return tuple(parser.parse_root(root) for root in roots) + + if parser.templates: + return tuple( + parser.parse_template(template_id) for template_id in parser.templates + ) + + raise ValueError("mvdXML document contains no ConceptRoot or ConceptTemplate") diff --git a/sparql.py b/sparql.py index 3dd35a7..af5bedd 100644 --- a/sparql.py +++ b/sparql.py @@ -10,7 +10,8 @@ from collections import defaultdict -import mvdxml_expression +from . import mvdxml_expression + def camel(s): """ @@ -20,28 +21,31 @@ def camel(s): """ s = s.title().replace(" ", "") - if s.endswith("s"): s = s[:-1] + if s.endswith("s"): + s = s[:-1] return s[0].lower() + s[1:] STANDARD_PREFIXES = { - 'rdf': '', - 'owl': '', - 'xsd': '', - 'list': '', - 'ifcowl': '', - 'express': '', + "rdf": "", + "owl": "", + "xsd": "", + "list": "", + "ifcowl": "", + "express": "", } + def derive_prefix(ttlfn): with open(ttlfn, "r") as f: for ln in f: ln.strip() if ln.startswith("@prefix ifcowl"): - uri = ln.split(':', 1)[1].strip()[:-1].strip() - print("Detected ifcowl prefix", uri) - STANDARD_PREFIXES['ifcowl'] = uri - break + uri = ln.split(":", 1)[1].strip()[:-1].strip() + STANDARD_PREFIXES["ifcowl"] = uri + return uri + raise ValueError(f"No ifcowl prefix found in {ttlfn!r}") + def withschema(fn): """ @@ -54,19 +58,23 @@ def withschema(fn): """ def _(*args, **kwargs): - schema_name = STANDARD_PREFIXES['ifcowl'].split('/')[-1][:-2] + schema_name = STANDARD_PREFIXES["ifcowl"].split("/")[-1][:-2] if "_" in schema_name: - schema_name = schema_name.split('_')[0] + schema_name = schema_name.split("_")[0] S = ifcopenshell.ifcopenshell_wrapper.schema_by_name(schema_name) return fn(S, *args, **kwargs) + return _ + noop = lambda *args: None + class rule_binding(object): """ Object for mapping rules to generated SPARQL variables """ + def __init__(self): pass @@ -85,8 +93,8 @@ def append(self, *stmt): for i, pos in enumerate(stmt[1:]): for po in pos.split("/"): if ":" in pos: - a, b = po.split(':') - self.prefixes[a] = '' + a, b = po.split(":") + self.prefixes[a] = "" self.statements.append(stmt) def bind(self, di): @@ -99,16 +107,19 @@ def x(self): def __repr__(self): def f(s): S = " ".join(s) - if len(s) == 3: S += ' .' + if len(s) == 3: + S += " ." return S def g(s): return "PREFIX %s: %s" % s - return "\n".join(itertools.chain( - (g(s) for s in self.prefixes.items()), - (f(s) for s in self.statements) - )) + return "\n".join( + itertools.chain( + (g(s) for s in self.prefixes.items()), (f(s) for s in self.statements) + ) + ) + class ifcOwl(object): """ @@ -126,14 +137,15 @@ def supertypes(S, entity): :return: """ - a, b = entity.split('#') + a, b = entity.split("#") try: en = S.declaration_by_name(b) if en.__class__.__name__ == "entity": while en.supertype(): yield "%s#%s" % (a, en.supertype().name()) en = en.supertype() - except: pass + except: + pass @staticmethod def get_names(e, c): @@ -188,7 +200,9 @@ def is_boxed(S, entity, attribute, predCount=0): while isinstance(ty, ifcopenshell.ifcopenshell_wrapper.type_declaration): is_boxed = True ty = ty.declared_type() - if is_boxed and isinstance(ty, ifcopenshell.ifcopenshell_wrapper.simple_type): + if is_boxed and isinstance( + ty, ifcopenshell.ifcopenshell_wrapper.simple_type + ): ty = ty.declared_type() return "express:has%s%s" % (ty[0].upper(), ty[1:]) @@ -210,15 +224,19 @@ def name(S, entity, attribute): while True: st = en.supertype() - attribute_names = ifcOwl.get_names(en, "all_attributes") | \ - ifcOwl.get_names(en, "all_inverse_attributes") + attribute_names = ifcOwl.get_names(en, "all_attributes") | ifcOwl.get_names( + en, "all_inverse_attributes" + ) if st: - attribute_names -= ifcOwl.get_names(st, "all_attributes") | \ - ifcOwl.get_names(st, "all_inverse_attributes") + attribute_names -= ifcOwl.get_names( + st, "all_attributes" + ) | ifcOwl.get_names(st, "all_inverse_attributes") if attribute in attribute_names: - return "ifcowl:" + attribute[0].lower() + attribute[1:] + "_" + en.name() + return ( + "ifcowl:" + attribute[0].lower() + attribute[1:] + "_" + en.name() + ) en = st @@ -253,7 +271,8 @@ def is_inverse(S, entity, attribute): en = S.declaration_by_name(entity) attrs = [a for a in en.all_inverse_attributes() if a.name() == attribute] - if not attrs: return False, False + if not attrs: + return False, False a = attrs[0] assert a.type_of_aggregation_string() == "set" @@ -261,6 +280,7 @@ def is_inverse(S, entity, attribute): attr = a.attribute_reference().name() return "ifcowl:" + entity, ifcOwl.name(entity, attr) + class convertor(object): @staticmethod @@ -278,8 +298,8 @@ def concept_or_applicability(concept): bld = builder() t = concept.template() - bld.append("# %s" % camel(concept.root.name)) - convertor.template(t, bld, concept.root.entity) + bld.append("# %s" % camel(concept.root_name)) + convertor.template(t, bld, concept.root_entity) bld.bind(STANDARD_PREFIXES) return bld @@ -299,7 +319,7 @@ def root(rootEntity): return b @staticmethod - def template(template, bld = None, rootEntity = None): + def template(template, bld=None, rootEntity=None): if bld is None: bld = builder() @@ -324,7 +344,7 @@ def enumerate(rule, **kwargs): bld.append("?URI", "ifcowl:globalId_IfcRoot/express:hasString", "?GlobalId") nm = "?URI" - ROOT = type('_', (), {'attribute': rootEntity})() + ROOT = type("_", (), {"attribute": rootEntity})() # rule_stack = [ROOT] # name_stack = [nm] # callback_stack = [noop] @@ -349,7 +369,11 @@ def build(rule, parents): if not ifcOwl.is_select(rule.attribute): # SELECT types should never be qualified as they cannot be inferred - bld.append(INDENT + rule_mapping[parents[-1]].name, "rdf:type", "ifcowl:" + rule.attribute) + bld.append( + INDENT + rule_mapping[parents[-1]].name, + "rdf:type", + "ifcowl:" + rule.attribute, + ) # propagate binding name rule_mapping[rule].name = rule_mapping[parents[-1]].name @@ -373,7 +397,9 @@ def build(rule, parents): if rule.bind: if len(rule.nodes) == 1: - indirect = ifcOwl.is_boxed(parents[-1].attribute, rule.attribute, predCount=bld.x()) + indirect = ifcOwl.is_boxed( + parents[-1].attribute, rule.attribute, predCount=bld.x() + ) if rule.bind and not indirect: next_nm = "?" + rule.bind @@ -382,7 +408,9 @@ def build(rule, parents): rule_mapping[rule].name = next_nm - inventy, invattr = ifcOwl.is_inverse(parents[-1].attribute, rule.attribute) + inventy, invattr = ifcOwl.is_inverse( + parents[-1].attribute, rule.attribute + ) if invattr: # This seems not to be necessary, because the entity name is also stated in mvdXML # q.append( @@ -391,15 +419,13 @@ def build(rule, parents): # inventy # ) bld.append( - INDENT + next_nm, - invattr, - rule_mapping[parents[-1]].name + INDENT + next_nm, invattr, rule_mapping[parents[-1]].name ) else: bld.append( INDENT + rule_mapping[parents[-1]].name, ifcOwl.name(parents[-1].attribute, rule.attribute), - next_nm + next_nm, ) if rule.bind and indirect: @@ -422,7 +448,7 @@ def build(rule, parents): # while callback_stack: # callback_stack.pop()() - if template.params: + if template.constraints: bld.append(convertor.build_filter(template)) bld.append("}") @@ -434,7 +460,7 @@ def build_filter(self): def v(p): if isinstance(p, mvdxml_expression.node): if p.b == "Value": - if p.c.lower() in {'true', 'false'}: + if p.c.lower() in {"true", "false"}: yield "(%s?%s)" % ("!" if p.c.lower() == "false" else "", p.a) else: yield "(?%s = %s)" % (p.a, p.c) @@ -443,18 +469,15 @@ def v(p): else: raise Exception("Unsupported " + p.b) elif isinstance(p, str): - yield { - "and": "&&", - "or": "||", - "not": "&& !" - }[p.lower()] + yield {"and": "&&", "or": "||", "not": "&& !"}[p.lower()] else: yield "(" for q in p: yield from v(q) yield ")" - return "FILTER(%s)" % " ".join(v(self.params)) + return "FILTER(%s)" % " ".join(v(self.constraints)) + def infer_subtypes(ttlfn): # Disabled currently @@ -462,8 +485,6 @@ def infer_subtypes(ttlfn): if not os.path.exists(ttlfn + ".subclass.nt"): - print("Inferring supertype relationships") - import hashlib import rdflib @@ -508,11 +529,13 @@ def _(): ttlfn += ".subclass.nt" -if platform.system() == "Windows": - JENA_SPARQL = os.path.join(os.environ.get("JENA_HOME"), "bat", "sparql.bat") + +if platform.system() == "Windows" and (jena_home := os.environ.get("JENA_HOME")): + JENA_SPARQL = os.path.join(jena_home, "bat", "sparql.bat") else: JENA_SPARQL = "sparql" + class executor(object): @staticmethod def run(CR, fn, ttlfn): @@ -525,6 +548,8 @@ def run(CR, fn, ttlfn): :return: """ + report = io.StringIO() + def dict_to_list(headers): return lambda di: [di[h] for h in headers] @@ -534,19 +559,28 @@ def execute(query, *args): print(query, file=f) proc = subprocess.Popen( - [JENA_SPARQL, "--data=" + ttlfn, "--query=" + sparqlfn, "--results=CSV"], + [ + JENA_SPARQL, + "--data=" + ttlfn, + "--query=" + sparqlfn, + "--results=CSV", + ], stdout=subprocess.PIPE, - stderr=subprocess.PIPE) + stderr=subprocess.PIPE, + ) stdout, stderr = proc.communicate() - csvf = io.StringIO(stdout.decode('utf-8')) + csvf = io.StringIO(stdout.decode("utf-8")) return list(csv.DictReader(csvf)) root_query = convertor.root(CR.entity) roots = execute(root_query, 0) - print("\nFile contains %d elements of type %s" % (len(roots), CR.entity)) + print( + "\nFile contains %d elements of type %s" % (len(roots), CR.entity), + file=report, + ) passing_all = {} @@ -556,7 +590,9 @@ def execute(query, *args): try: # Full MVD with multiple concepts is_template = False - concept_enumerator = list(itertools.chain([CR.applicability()], CR.concepts())) + concept_enumerator = list( + itertools.chain([CR.applicability()], CR.concepts()) + ) except: is_template = True concept_enumerator = [CR] @@ -566,44 +602,71 @@ def execute(query, *args): num_columns += 1 if is_template or ci > 1: - print("\n%s" % C.name) + print("\n%s" % C.name, file=report) else: - print("\nApplicability") + print("\nApplicability", file=report) query = convertor.convert(C) - print("\nSPARQL query") - print("============") - print(query) + print("\nSPARQL query", file=report) + print("============", file=report) + print(query, file=report) passing = execute(query, ci, 1) - passing_guids = set(r['GlobalId'] for r in passing) - - print("\nElements passing") - print(tabulate.tabulate(list(map(dict_to_list(query.args), passing)), query.args, tablefmt="grid")) - - print("\nElements failing concept") + passing_guids = set(r["GlobalId"] for r in passing) + + print("\nElements passing", file=report) + print( + tabulate.tabulate( + list(map(dict_to_list(query.args), passing)), + query.args, + tablefmt="grid", + ), + file=report, + ) + + print("\nElements failing concept", file=report) hd = ["URI", "GlobalId"] - print(tabulate.tabulate( - list(map(dict_to_list(hd), [r for r in roots if r["GlobalId"] not in passing_guids])), hd, - tablefmt="grid")) + print( + tabulate.tabulate( + list( + map( + dict_to_list(hd), + [r for r in roots if r["GlobalId"] not in passing_guids], + ) + ), + hd, + tablefmt="grid", + ), + file=report, + ) passing_all[ci] = passing_guids - print("\nSummary") + print("\nSummary", file=report) for ci, C in enumerate(concept_enumerator): - print("(%d) %s" % (ci+(0 if is_template else 0), C.name)) + print("(%d) %s" % (ci + (0 if is_template else 0), C.name), file=report) def get_stats(guid): v = lambda i: guid in passing_all[i] st = [guid] + ["x" if v(i) else "" for i in range(num_columns)] if not is_template: - st += ["x" if not v(0) or all(v(i) for i in range(1, num_columns)) else " "] + st += [ + "x" if not v(0) or all(v(i) for i in range(1, num_columns)) else " " + ] return st hd = ["GlobalId"] + list(map(str, range(num_columns))) if not is_template: hd += ["Valid"] - print(tabulate.tabulate(list(map(get_stats, map(operator.itemgetter("GlobalId"), roots))), hd, tablefmt="grid")) \ No newline at end of file + print( + tabulate.tabulate( + list(map(get_stats, map(operator.itemgetter("GlobalId"), roots))), + hd, + tablefmt="grid", + ), + file=report, + ) + return report.getvalue() diff --git a/tests/test_graphviz.py b/tests/test_graphviz.py new file mode 100644 index 0000000..0b844ad --- /dev/null +++ b/tests/test_graphviz.py @@ -0,0 +1,163 @@ +# This file was generated with the assistance of an AI coding tool. + +import pytest + +import ifcopenshell +from ifcopenshell.mvd import rule, template + + +def find_rule(parsed: rule, tag: str) -> rule: + if parsed.tag == tag: + return parsed + for child in parsed.nodes: + try: + return find_rule(child, tag) + except LookupError: + pass + raise LookupError(tag) + + +def test_from_graphviz_parses_edges_and_bindings() -> None: + source = """\ +Text outside the supported block is ignored. + +``` +concept { + IfcObject:ObjectType -> IfcLabel + IfcObject:IsTypedBy -> IfcRelDefinesByType:RelatedObjects + IfcRelDefinesByType:RelatingType -> IfcTypeObject + IfcObject:ObjectType[binding="UserDefinedType"] + IfcTypeObject:PredefinedType[binding="TypePredefinedType"] +} +``` +""" + + parsed = template.from_graphviz(source, name="Object Predefined Type") + + assert parsed.entity == "IfcObject" + assert parsed.name == "Object Predefined Type" + assert [item.attribute for item in parsed.rules] == ["ObjectType", "IsTypedBy"] + assert parsed.rules[0].bind == "UserDefinedType" + assert parsed.rules[1].nodes[0].attribute == "IfcRelDefinesByType" + assert parsed.rules[1].nodes[0].nodes[0].attribute == "RelatingType" + + +def test_from_graphviz_parses_constraint_using_attribute_binding() -> None: + source = """\ +``` +concept { + IfcProduct:Representation -> IfcProductDefinitionShape + IfcProductDefinitionShape:Representations -> IfcShapeRepresentation + IfcShapeRepresentation:RepresentationIdentifier -> IfcLabel_0 + IfcLabel_0 -> constraint_0 + constraint_0[label="=Body"] + IfcShapeRepresentation:RepresentationIdentifier[binding="Identifier"] +} +``` +""" + + parsed = template.from_graphviz(source) + constraint = find_rule(parsed.rules[0], "Constraint") + expression = constraint.attribute[0][0] + + assert constraint.tag == "Constraint" + assert expression.a == "Identifier" + assert expression.b == "Value" + assert expression.c == "Body" + + +def test_from_graphviz_expands_named_template_references() -> None: + referenced = template( + entity="IfcSurfaceStyle", + name="Surface Color Style", + rules=(rule("AttributeRule", "Name", bind="StyleName"),), + ) + source = """\ +``` +concept { + IfcSurfaceStyle -> Surface_Color_Style +} +``` +""" + + parsed = template.from_graphviz( + source, + name="Surface Style", + references={"Surface Color Style": referenced}, + ) + + assert parsed.entity == "IfcSurfaceStyle" + assert parsed.rules == referenced.rules + + +def test_from_graphviz_only_reads_fenced_blocks() -> None: + source = """\ +concept { + IfcWrong:Name -> IfcLabel +} + +``` +concept { + IfcWall:Name -> IfcLabel +} +``` +""" + + assert template.from_graphviz(source).entity == "IfcWall" + + with pytest.raises(ValueError, match="No fenced Graphviz concept block"): + template.from_graphviz("concept { IfcWall:Name -> IfcLabel }") + + +def test_from_graphviz_reports_unknown_reference() -> None: + source = """\ +``` +concept { + IfcSurfaceStyle -> Surface_Color_Style +} +``` +""" + + with pytest.raises(ValueError, match="Unknown Graphviz template reference: Surface_Color_Style"): + template.from_graphviz(source) + + +def test_from_graphviz_entity_leaf_filters_ifc_type() -> None: + source = """\ +``` +concept { + IfcRelDefinesByProperties:RelatingPropertyDefinition -> IfcPropertySetDefinitionSet +} +``` +""" + parsed = template.from_graphviz(source) + + class entity(ifcopenshell.entity_instance): + def __init__(self, ifc_class: str): + self.ifc_class = ifc_class + + def is_a(self, ifc_class: str) -> bool: + return self.ifc_class == ifc_class + + class relationship: + def __init__(self, relating_property_definition: entity): + self.RelatingPropertyDefinition = relating_property_definition + + assert parsed.extract(relationship(entity("IfcPropertySetDefinitionSet"))) + assert not parsed.extract(relationship(entity("IfcPropertySet"))) + + +def test_from_graphviz_leaf_uses_parent_attribute_binding() -> None: + source = """\ +``` +concept { + IfcPointByDistanceExpression:BasisCurve -> IfcCurve + IfcPointByDistanceExpression:BasisCurve[binding="BasisCurve"] +} +``` +""" + + parsed = template.from_graphviz(source) + leaf = parsed.rules[0].nodes[0] + + assert parsed.binding_for(leaf) == "BasisCurve" diff --git a/tests/test_mvd.py b/tests/test_mvd.py new file mode 100644 index 0000000..023a7a0 --- /dev/null +++ b/tests/test_mvd.py @@ -0,0 +1,151 @@ +# This file was generated with the assistance of an AI coding tool. + +from dataclasses import FrozenInstanceError +from pathlib import Path + +import pytest + +from ifcopenshell.mvd import ( + concept_or_applicability, + concept_root, + parse, + rule, + template, +) +from ifcopenshell.mvd.__main__ import build_parser +from ifcopenshell.mvd.mvdxml_expression import parse as parse_expression +from ifcopenshell.mvd import model + + +EXAMPLES = Path(__file__).parents[1] / "mvd_examples" + + +def test_parse_readme_example() -> None: + parsed = parse(EXAMPLES / "wall_extraction.mvdxml") + + assert len(parsed) == 1 + root = parsed[0] + assert isinstance(root, concept_root) + assert root.name == "IfcWall" + assert root.entity == "IfcWall" + assert [concept.name for concept in root.concepts()] == [ + "APexternal", + "voids", + "mat", + "name", + "area", + ] + + +def test_parsed_templates_are_immutable_and_accessed_once() -> None: + root = parse(EXAMPLES / "wall_extraction.mvdxml")[0] + assert isinstance(root, concept_root) + concept = next(root.concepts()) + + assert concept.template() is concept.template() + with pytest.raises(FrozenInstanceError): + concept.template().name = "Changed" + + +def test_extract_returns_native_containers_without_printing( + capsys: pytest.CaptureFixture[str], +) -> None: + name_rule = rule("AttributeRule", "Name") + parsed_template = template( + "IfcWall", "Name", (rule("EntityRule", "IfcWall", (name_rule,)),) + ) + concept = concept_or_applicability("name", parsed_template) + + class wall: + GlobalId = "wall-guid" + Name = "Wall A" + + extracted = concept.extract([wall()]) + + assert extracted == {"wall-guid": "Wall A"} + assert capsys.readouterr() == ("", "") + + +def test_parse_reports_missing_template_reference(tmp_path: Path) -> None: + filename = tmp_path / "missing-template.mvdxml" + filename.write_text( + """\ + + + + + + +