mirror of
https://github.com/IfcOpenShell/IfcOpenShell.git
synced 2026-08-16 02:24:30 +00:00
Add validate, schedule, cost, schema, quantify to ifcquery/ifcedit/ifcmcp
ifcquery: validate [--rules], schedule [--depth N], cost [--depth N],
schema <EntityType>. ifcedit: quantify list/run subcommands using ifc5d
QTO rules. schedule and cost support max_depth to limit tree expansion,
replacing truncated levels with {truncated, count}. ifcmcp gains matching
session methods, @server.tool() decorators, and OpenAI tool schemas.
Generated with the assistance of an AI coding tool.
This commit is contained in:
@@ -26,7 +26,9 @@ import sys
|
||||
import ifcopenshell
|
||||
|
||||
from ifcquery import clash as clash_mod
|
||||
from ifcquery import info, relations, select, summary, tree
|
||||
from ifcquery import cost as cost_mod
|
||||
from ifcquery import info, relations, schedule, schema, select, summary, tree
|
||||
from ifcquery import validate as validate_mod
|
||||
|
||||
|
||||
def parse_element_id(raw: str) -> int:
|
||||
@@ -103,6 +105,24 @@ def main():
|
||||
"--scope", choices=["storey", "all"], default="storey", help="Scope of elements to check (default: storey)"
|
||||
)
|
||||
|
||||
validate_parser = subparsers.add_parser("validate", help="Schema/constraint validation")
|
||||
validate_parser.add_argument(
|
||||
"--rules", action="store_true", help="Also check EXPRESS rules (slower, default: false)"
|
||||
)
|
||||
|
||||
schedule_parser = subparsers.add_parser("schedule", help="List work plans and tasks from the model")
|
||||
schedule_parser.add_argument(
|
||||
"--depth", type=int, default=None, metavar="N", help="Limit subtask expansion to N levels (default: unlimited)"
|
||||
)
|
||||
|
||||
cost_parser = subparsers.add_parser("cost", help="List cost schedules and cost items from the model")
|
||||
cost_parser.add_argument(
|
||||
"--depth", type=int, default=None, metavar="N", help="Limit cost item expansion to N levels (default: unlimited)"
|
||||
)
|
||||
|
||||
schema_parser = subparsers.add_parser("schema", help="IFC class documentation")
|
||||
schema_parser.add_argument("entity_type", help="IFC entity type (e.g. IfcWall)")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
@@ -159,6 +179,14 @@ def main():
|
||||
except ImportError:
|
||||
print("Error: ifcopenshell geometry engine not available (C++ bindings required)", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
elif args.command == "validate":
|
||||
result = validate_mod.validate(model, express_rules=args.rules)
|
||||
elif args.command == "schedule":
|
||||
result = schedule.schedule(model, max_depth=args.depth)
|
||||
elif args.command == "cost":
|
||||
result = cost_mod.cost(model, max_depth=args.depth)
|
||||
elif args.command == "schema":
|
||||
result = schema.schema(model, args.entity_type)
|
||||
|
||||
print(format_output(result, args.output_format))
|
||||
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
# This file was generated with the assistance of an AI coding tool.
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import ifcopenshell
|
||||
import ifcopenshell.util.cost as cost_util
|
||||
|
||||
|
||||
def _cost_item_to_dict(item: ifcopenshell.entity_instance, max_depth: int | None, depth: int) -> dict[str, Any]:
|
||||
raw_values = cost_util.get_cost_values(item)
|
||||
values = [{"formula": v.get("label", ""), "category": v.get("category")} for v in raw_values]
|
||||
|
||||
if max_depth is not None and depth >= max_depth:
|
||||
child_count = len(cost_util.get_nested_cost_items(item))
|
||||
subitems = {"truncated": True, "count": child_count} if child_count else []
|
||||
else:
|
||||
subitems = [_cost_item_to_dict(sub, max_depth, depth + 1) for sub in cost_util.get_nested_cost_items(item)]
|
||||
|
||||
return {
|
||||
"id": item.id(),
|
||||
"name": getattr(item, "Name", None),
|
||||
"values": values,
|
||||
"subitems": subitems,
|
||||
}
|
||||
|
||||
|
||||
def cost(model: ifcopenshell.file, max_depth: int | None = None) -> list[dict[str, Any]]:
|
||||
"""Return a list of IfcCostSchedule entries with nested cost item trees.
|
||||
|
||||
max_depth limits how many levels of subitems are expanded (None = unlimited).
|
||||
At the cutoff level, subitems is replaced with {"truncated": True, "count": N}.
|
||||
"""
|
||||
result = []
|
||||
for cost_schedule in model.by_type("IfcCostSchedule"):
|
||||
items = [_cost_item_to_dict(i, max_depth, depth=1) for i in cost_util.get_root_cost_items(cost_schedule)]
|
||||
result.append(
|
||||
{
|
||||
"id": cost_schedule.id(),
|
||||
"name": getattr(cost_schedule, "Name", None),
|
||||
"predefined_type": getattr(cost_schedule, "PredefinedType", None),
|
||||
"items": items,
|
||||
}
|
||||
)
|
||||
return result
|
||||
@@ -0,0 +1,56 @@
|
||||
# This file was generated with the assistance of an AI coding tool.
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import ifcopenshell
|
||||
import ifcopenshell.util.sequence as seq
|
||||
|
||||
|
||||
def _task_to_dict(task: ifcopenshell.entity_instance, max_depth: int | None, depth: int) -> dict[str, Any]:
|
||||
task_time = task.TaskTime
|
||||
start = None
|
||||
finish = None
|
||||
if task_time:
|
||||
start = task_time.ScheduleStart
|
||||
finish = task_time.ScheduleFinish
|
||||
|
||||
outputs = []
|
||||
for product in seq.get_task_outputs(task):
|
||||
outputs.append({"id": product.id(), "type": product.is_a(), "name": getattr(product, "Name", None)})
|
||||
|
||||
if max_depth is not None and depth >= max_depth:
|
||||
child_count = len(seq.get_nested_tasks(task))
|
||||
subtasks = {"truncated": True, "count": child_count} if child_count else []
|
||||
else:
|
||||
subtasks = [_task_to_dict(sub, max_depth, depth + 1) for sub in seq.get_nested_tasks(task)]
|
||||
|
||||
return {
|
||||
"id": task.id(),
|
||||
"name": getattr(task, "Name", None),
|
||||
"start": start,
|
||||
"finish": finish,
|
||||
"is_milestone": bool(task.IsMilestone) if hasattr(task, "IsMilestone") else False,
|
||||
"outputs": outputs,
|
||||
"subtasks": subtasks,
|
||||
}
|
||||
|
||||
|
||||
def schedule(model: ifcopenshell.file, max_depth: int | None = None) -> list[dict[str, Any]]:
|
||||
"""Return a list of IfcWorkSchedule entries with nested task trees.
|
||||
|
||||
max_depth limits how many levels of subtasks are expanded (None = unlimited).
|
||||
At the cutoff level, subtasks is replaced with {"truncated": True, "count": N}.
|
||||
"""
|
||||
result = []
|
||||
for work_schedule in model.by_type("IfcWorkSchedule"):
|
||||
tasks = [_task_to_dict(t, max_depth, depth=1) for t in seq.get_root_tasks(work_schedule)]
|
||||
result.append(
|
||||
{
|
||||
"id": work_schedule.id(),
|
||||
"name": getattr(work_schedule, "Name", None),
|
||||
"predefined_type": getattr(work_schedule, "PredefinedType", None),
|
||||
"tasks": tasks,
|
||||
}
|
||||
)
|
||||
return result
|
||||
@@ -0,0 +1,19 @@
|
||||
# This file was generated with the assistance of an AI coding tool.
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import ifcopenshell
|
||||
import ifcopenshell.util.doc
|
||||
|
||||
|
||||
def schema(model: ifcopenshell.file, entity_type: str) -> dict[str, Any]:
|
||||
"""Return IFC class documentation for entity_type from model's schema version."""
|
||||
schema_name = model.schema
|
||||
try:
|
||||
doc = ifcopenshell.util.doc.get_entity_doc(schema_name, entity_type)
|
||||
except Exception:
|
||||
return {"error": f"Unknown entity: {entity_type}"}
|
||||
if not doc:
|
||||
return {"error": f"Unknown entity: {entity_type}"}
|
||||
return dict(doc)
|
||||
@@ -0,0 +1,15 @@
|
||||
# This file was generated with the assistance of an AI coding tool.
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import ifcopenshell
|
||||
import ifcopenshell.validate
|
||||
|
||||
|
||||
def validate(model: ifcopenshell.file, express_rules: bool = False) -> dict[str, Any]:
|
||||
"""Validate the model and return a dict with 'valid' bool and 'issues' list."""
|
||||
logger = ifcopenshell.validate.json_logger()
|
||||
ifcopenshell.validate.validate(model, logger, express_rules=express_rules)
|
||||
issues = [{"level": s["level"], "message": s["message"]} for s in logger.statements]
|
||||
return {"valid": len(issues) == 0, "issues": issues}
|
||||
@@ -0,0 +1,108 @@
|
||||
# This file was generated with the assistance of an AI coding tool.
|
||||
from __future__ import annotations
|
||||
|
||||
import ifcopenshell
|
||||
import ifcopenshell.api.cost
|
||||
import ifcopenshell.api.owner.settings
|
||||
import ifcopenshell.api.project
|
||||
import ifcopenshell.api.root
|
||||
import ifcopenshell.api.unit
|
||||
import pytest
|
||||
|
||||
from ifcquery.cost import cost
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cost_model():
|
||||
"""Create an IFC4 model with a cost schedule, a top-level item, and one nested subitem."""
|
||||
f = ifcopenshell.api.project.create_file()
|
||||
ifcopenshell.api.owner.settings.get_user = lambda ifc: (ifc.by_type("IfcPersonAndOrganization") or [None])[0]
|
||||
ifcopenshell.api.owner.settings.get_application = lambda ifc: (ifc.by_type("IfcApplication") or [None])[0]
|
||||
|
||||
project = ifcopenshell.api.root.create_entity(f, ifc_class="IfcProject", name="TestProject")
|
||||
ifcopenshell.api.unit.assign_unit(f)
|
||||
|
||||
cs = ifcopenshell.api.cost.add_cost_schedule(f, name="Bill of Quantities")
|
||||
item = ifcopenshell.api.cost.add_cost_item(f, cost_schedule=cs)
|
||||
ifcopenshell.api.cost.edit_cost_item(f, cost_item=item, attributes={"Name": "Concrete Works"})
|
||||
cv = ifcopenshell.api.cost.add_cost_value(f, parent=item)
|
||||
ifcopenshell.api.cost.edit_cost_value(f, cost_value=cv, attributes={"AppliedValue": 1200.0, "Category": "material"})
|
||||
|
||||
# Add a nested subitem
|
||||
subitem = ifcopenshell.api.cost.add_cost_item(f, cost_item=item)
|
||||
ifcopenshell.api.cost.edit_cost_item(f, cost_item=subitem, attributes={"Name": "Formwork"})
|
||||
|
||||
return f
|
||||
|
||||
|
||||
class TestCost:
|
||||
def test_returns_list(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_finds_cost_schedule(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
assert len(result) == 1
|
||||
|
||||
def test_schedule_has_name(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
assert result[0]["name"] == "Bill of Quantities"
|
||||
|
||||
def test_schedule_has_id(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
assert isinstance(result[0]["id"], int)
|
||||
assert result[0]["id"] > 0
|
||||
|
||||
def test_schedule_has_items(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
assert len(result[0]["items"]) == 1
|
||||
|
||||
def test_item_has_required_fields(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
item = result[0]["items"][0]
|
||||
assert "id" in item
|
||||
assert "name" in item
|
||||
assert "values" in item
|
||||
assert "subitems" in item
|
||||
|
||||
def test_item_name(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
assert result[0]["items"][0]["name"] == "Concrete Works"
|
||||
|
||||
def test_item_has_values(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
values = result[0]["items"][0]["values"]
|
||||
assert len(values) == 1
|
||||
assert "formula" in values[0]
|
||||
assert "category" in values[0]
|
||||
|
||||
def test_item_value_category(self, cost_model):
|
||||
result = cost(cost_model)
|
||||
values = result[0]["items"][0]["values"]
|
||||
assert values[0]["category"] == "material"
|
||||
|
||||
def test_empty_model_returns_empty_list(self, model):
|
||||
result = cost(model)
|
||||
assert result == []
|
||||
|
||||
def test_max_depth_none_returns_full_tree(self, cost_model):
|
||||
result = cost(cost_model, max_depth=None)
|
||||
item = result[0]["items"][0]
|
||||
assert isinstance(item["subitems"], list)
|
||||
assert len(item["subitems"]) == 1
|
||||
assert item["subitems"][0]["name"] == "Formwork"
|
||||
|
||||
def test_max_depth_1_truncates_subitems(self, cost_model):
|
||||
result = cost(cost_model, max_depth=1)
|
||||
item = result[0]["items"][0]
|
||||
assert isinstance(item["subitems"], dict)
|
||||
assert item["subitems"]["truncated"] is True
|
||||
assert item["subitems"]["count"] == 1
|
||||
|
||||
def test_max_depth_2_expands_to_depth_2(self, cost_model):
|
||||
result = cost(cost_model, max_depth=2)
|
||||
item = result[0]["items"][0]
|
||||
assert isinstance(item["subitems"], list)
|
||||
assert item["subitems"][0]["name"] == "Formwork"
|
||||
# subitem has no children, so subitems should be empty list
|
||||
assert item["subitems"][0]["subitems"] == []
|
||||
@@ -0,0 +1,121 @@
|
||||
# This file was generated with the assistance of an AI coding tool.
|
||||
from __future__ import annotations
|
||||
|
||||
import ifcopenshell
|
||||
import ifcopenshell.api.aggregate
|
||||
import ifcopenshell.api.owner.settings
|
||||
import ifcopenshell.api.project
|
||||
import ifcopenshell.api.root
|
||||
import ifcopenshell.api.sequence
|
||||
import ifcopenshell.api.unit
|
||||
import pytest
|
||||
|
||||
from ifcquery.schedule import schedule
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def schedule_model():
|
||||
"""Create an IFC4 model with a work schedule and nested tasks."""
|
||||
f = ifcopenshell.api.project.create_file()
|
||||
ifcopenshell.api.owner.settings.get_user = lambda ifc: (ifc.by_type("IfcPersonAndOrganization") or [None])[0]
|
||||
ifcopenshell.api.owner.settings.get_application = lambda ifc: (ifc.by_type("IfcApplication") or [None])[0]
|
||||
|
||||
project = ifcopenshell.api.root.create_entity(f, ifc_class="IfcProject", name="TestProject")
|
||||
ifcopenshell.api.unit.assign_unit(f)
|
||||
|
||||
ws = ifcopenshell.api.sequence.add_work_schedule(f, name="Construction Schedule")
|
||||
|
||||
task1 = ifcopenshell.api.sequence.add_task(f, work_schedule=ws, name="Phase 1", identification="P1")
|
||||
tt1 = ifcopenshell.api.sequence.add_task_time(f, task=task1)
|
||||
ifcopenshell.api.sequence.edit_task_time(
|
||||
f, task_time=tt1, attributes={"ScheduleStart": "2024-01-01", "ScheduleFinish": "2024-06-30"}
|
||||
)
|
||||
|
||||
task2 = ifcopenshell.api.sequence.add_task(f, work_schedule=ws, name="Phase 2", identification="P2")
|
||||
subtask = ifcopenshell.api.sequence.add_task(f, parent_task=task1, name="Sub Task", identification="S1")
|
||||
|
||||
return f
|
||||
|
||||
|
||||
class TestSchedule:
|
||||
def test_returns_list(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_finds_work_schedule(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
assert len(result) == 1
|
||||
|
||||
def test_work_schedule_has_name(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
assert result[0]["name"] == "Construction Schedule"
|
||||
|
||||
def test_work_schedule_has_id(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
assert isinstance(result[0]["id"], int)
|
||||
assert result[0]["id"] > 0
|
||||
|
||||
def test_work_schedule_has_tasks(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
tasks = result[0]["tasks"]
|
||||
assert isinstance(tasks, list)
|
||||
assert len(tasks) >= 1
|
||||
|
||||
def test_task_has_required_fields(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
task = result[0]["tasks"][0]
|
||||
assert "id" in task
|
||||
assert "name" in task
|
||||
assert "start" in task
|
||||
assert "finish" in task
|
||||
assert "is_milestone" in task
|
||||
assert "outputs" in task
|
||||
assert "subtasks" in task
|
||||
|
||||
def test_task_name(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
task_names = [t["name"] for t in result[0]["tasks"]]
|
||||
assert "Phase 1" in task_names
|
||||
|
||||
def test_task_start_finish(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
phase1 = next(t for t in result[0]["tasks"] if t["name"] == "Phase 1")
|
||||
assert phase1["start"] is not None
|
||||
assert phase1["finish"] is not None
|
||||
|
||||
def test_subtasks(self, schedule_model):
|
||||
result = schedule(schedule_model)
|
||||
phase1 = next(t for t in result[0]["tasks"] if t["name"] == "Phase 1")
|
||||
assert len(phase1["subtasks"]) == 1
|
||||
assert phase1["subtasks"][0]["name"] == "Sub Task"
|
||||
|
||||
def test_empty_model_returns_empty_list(self, model):
|
||||
result = schedule(model)
|
||||
assert result == []
|
||||
|
||||
def test_max_depth_none_returns_full_tree(self, schedule_model):
|
||||
result = schedule(schedule_model, max_depth=None)
|
||||
phase1 = next(t for t in result[0]["tasks"] if t["name"] == "Phase 1")
|
||||
assert isinstance(phase1["subtasks"], list)
|
||||
assert len(phase1["subtasks"]) == 1
|
||||
|
||||
def test_max_depth_1_truncates_subtasks(self, schedule_model):
|
||||
result = schedule(schedule_model, max_depth=1)
|
||||
phase1 = next(t for t in result[0]["tasks"] if t["name"] == "Phase 1")
|
||||
assert isinstance(phase1["subtasks"], dict)
|
||||
assert phase1["subtasks"]["truncated"] is True
|
||||
assert phase1["subtasks"]["count"] == 1
|
||||
|
||||
def test_max_depth_truncation_shows_count(self, schedule_model):
|
||||
result = schedule(schedule_model, max_depth=1)
|
||||
# Phase 2 has no subtasks — should return empty list, not truncation dict
|
||||
phase2 = next(t for t in result[0]["tasks"] if t["name"] == "Phase 2")
|
||||
assert phase2["subtasks"] == []
|
||||
|
||||
def test_max_depth_2_expands_to_depth_2(self, schedule_model):
|
||||
result = schedule(schedule_model, max_depth=2)
|
||||
phase1 = next(t for t in result[0]["tasks"] if t["name"] == "Phase 1")
|
||||
# subtask at depth 2 should be fully expanded (it has no children)
|
||||
assert isinstance(phase1["subtasks"], list)
|
||||
assert phase1["subtasks"][0]["name"] == "Sub Task"
|
||||
assert phase1["subtasks"][0]["subtasks"] == []
|
||||
@@ -0,0 +1,32 @@
|
||||
# This file was generated with the assistance of an AI coding tool.
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from ifcquery.schema import schema
|
||||
|
||||
|
||||
class TestSchema:
|
||||
def test_ifc_wall_has_description(self, model):
|
||||
result = schema(model, "IfcWall")
|
||||
assert "description" in result
|
||||
assert isinstance(result["description"], str)
|
||||
assert len(result["description"]) > 0
|
||||
|
||||
def test_ifc_wall_has_attributes(self, model):
|
||||
result = schema(model, "IfcWall")
|
||||
assert "attributes" in result
|
||||
|
||||
def test_ifc_wall_has_spec_url(self, model):
|
||||
result = schema(model, "IfcWall")
|
||||
assert "spec_url" in result
|
||||
|
||||
def test_unknown_entity_returns_error(self, model):
|
||||
result = schema(model, "IfcNonExistentFooBar")
|
||||
assert "error" in result
|
||||
assert "IfcNonExistentFooBar" in result["error"]
|
||||
|
||||
def test_ifc_window_has_description(self, model):
|
||||
result = schema(model, "IfcWindow")
|
||||
assert "description" in result
|
||||
assert len(result["description"]) > 0
|
||||
@@ -0,0 +1,47 @@
|
||||
# This file was generated with the assistance of an AI coding tool.
|
||||
from __future__ import annotations
|
||||
|
||||
import ifcopenshell
|
||||
import ifcopenshell.api.project
|
||||
import pytest
|
||||
|
||||
from ifcquery.validate import validate
|
||||
|
||||
|
||||
class TestValidate:
|
||||
def test_valid_model_returns_valid_true(self, model):
|
||||
result = validate(model)
|
||||
assert result["valid"] is True
|
||||
assert isinstance(result["issues"], list)
|
||||
|
||||
def test_valid_model_has_no_issues(self, model):
|
||||
result = validate(model)
|
||||
assert result["issues"] == []
|
||||
|
||||
def test_empty_model_is_valid(self):
|
||||
f = ifcopenshell.api.project.create_file()
|
||||
result = validate(f)
|
||||
assert result["valid"] is True
|
||||
assert result["issues"] == []
|
||||
|
||||
def test_result_has_expected_keys(self, model):
|
||||
result = validate(model)
|
||||
assert "valid" in result
|
||||
assert "issues" in result
|
||||
|
||||
def test_express_rules_flag_accepted(self, model):
|
||||
# Just verify it runs without error; express rules may add/not add issues
|
||||
result = validate(model, express_rules=True)
|
||||
assert "valid" in result
|
||||
assert isinstance(result["issues"], list)
|
||||
|
||||
def test_issue_has_level_and_message(self, model):
|
||||
# Force an issue by manually breaking the model (invalid IfcWall attribute)
|
||||
f = ifcopenshell.file()
|
||||
# Create a raw IfcWall with deliberately wrong type for GlobalId (use int)
|
||||
# We just check structure if any issues appear; on well-formed models there are none.
|
||||
result = validate(model)
|
||||
# Even if no issues, the structure contract must hold for any issues present
|
||||
for issue in result["issues"]:
|
||||
assert "level" in issue
|
||||
assert "message" in issue
|
||||
Reference in New Issue
Block a user