Spaces:
Sleeping
Sleeping
File size: 3,668 Bytes
51982d6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | import importlib.util
import os
import glob
import uuid
import time
import logging
import ifcopenshell
logger = logging.getLogger("ifcore")
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
TEAM_FIELDS = [
"element_id", "element_type", "element_name", "element_name_long",
"check_status", "actual_value", "required_value", "comment", "log",
]
def discover_checks():
checks = []
pattern = os.path.join(BASE_DIR, "teams", "*", "tools", "checker_*.py")
for path in sorted(glob.glob(pattern)):
parts = path.replace(BASE_DIR + os.sep, "").split(os.sep)
team = parts[1]
module_name = os.path.splitext(os.path.basename(path))[0]
try:
spec = importlib.util.spec_from_file_location(f"teams.{team}.{module_name}", path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
for attr in dir(mod):
if attr.startswith("check_") and callable(getattr(mod, attr)):
checks.append((team, attr, getattr(mod, attr)))
except Exception as exc:
logger.warning(f"[discover] skipping {team}/{module_name}: {exc}")
return checks
def _aggregate_status(elements):
statuses = [e.get("check_status", "blocked") for e in elements]
if any(s == "fail" for s in statuses):
return "fail"
if any(s == "warning" for s in statuses):
return "warning"
if all(s in ("pass", "log") for s in statuses):
return "pass"
return "unknown"
def _build_summary(elements):
p = sum(1 for e in elements if e.get("check_status") == "pass")
f = sum(1 for e in elements if e.get("check_status") == "fail")
w = sum(1 for e in elements if e.get("check_status") == "warning")
b = sum(1 for e in elements if e.get("check_status") == "blocked")
total = len(elements)
parts = []
if p: parts.append(f"{p} pass")
if f: parts.append(f"{f} fail")
if w: parts.append(f"{w} warning")
if b: parts.append(f"{b} blocked")
return f"{total} elements: {', '.join(parts)}" if parts else f"{total} elements"
def run_all_checks(ifc_path, job_id, project_id):
model = ifcopenshell.open(ifc_path)
checks = discover_checks()
check_results = []
element_results = []
for team, func_name, func in checks:
check_id = str(uuid.uuid4())
try:
elements = func(model)
status = _aggregate_status(elements)
summary = _build_summary(elements)
check_results.append({
"id": check_id,
"job_id": job_id,
"project_id": project_id,
"check_name": func_name,
"team": team,
"status": status,
"summary": summary,
"has_elements": 1 if elements else 0,
"created_at": int(time.time() * 1000),
})
for el in elements:
row = {"id": str(uuid.uuid4()), "check_result_id": check_id}
for field in TEAM_FIELDS:
row[field] = el.get(field)
element_results.append(row)
except Exception as exc:
check_results.append({
"id": check_id,
"job_id": job_id,
"project_id": project_id,
"check_name": func_name,
"team": team,
"status": "error",
"summary": str(exc)[:200],
"has_elements": 0,
"created_at": int(time.time() * 1000),
})
return {"check_results": check_results, "element_results": element_results}
|