leistungsbilanz-ts/dynamo/02_export_electrical_fixture_parameter_inventory.py

517 lines
16 KiB
Python

"""Export all Electrical Fixtures instance/type parameters from Revit 2026.
Optional Dynamo inputs:
IN[0]: output directory or complete .json file path
IN[1]: include empty parameters (default True)
IN[2]: maximum aggregated sample values (default 5)
The script is read-only and performs no Revit transaction.
"""
import datetime
import json
import os
import platform
import sys
import tempfile
import traceback
import clr
clr.AddReference("RevitAPI")
clr.AddReference("RevitServices")
from Autodesk.Revit.DB import ( # noqa: E402
BuiltInCategory,
FilteredElementCollector,
ModelPathUtils,
StorageType,
)
from RevitServices.Persistence import DocumentManager # noqa: E402
def safe_text(value):
if value is None:
return None
try:
return str(value)
except Exception:
return None
def element_id_text(element_id):
if element_id is None:
return None
try:
return str(element_id.Value)
except Exception:
try:
return str(element_id.IntegerValue)
except Exception:
return safe_text(element_id)
def forge_type_id_text(value):
if value is None:
return None
try:
return value.TypeId
except Exception:
return safe_text(value)
def parameter_value(parameter):
result = {"hasValue": False, "raw": None, "display": None}
try:
result["hasValue"] = bool(parameter.HasValue)
except Exception:
pass
try:
storage_type = parameter.StorageType
if storage_type == StorageType.String:
result["raw"] = parameter.AsString()
elif storage_type == StorageType.Integer:
result["raw"] = int(parameter.AsInteger())
elif storage_type == StorageType.Double:
result["raw"] = float(parameter.AsDouble())
elif storage_type == StorageType.ElementId:
result["raw"] = element_id_text(parameter.AsElementId())
except Exception as error:
result["readError"] = safe_text(error)
try:
result["display"] = parameter.AsValueString()
except Exception:
pass
return result
def describe_parameter(parameter, scope):
definition = None
try:
definition = parameter.Definition
except Exception:
pass
name = None
data_type = None
group_type = None
if definition is not None:
try:
name = definition.Name
except Exception:
pass
try:
data_type = forge_type_id_text(definition.GetDataType())
except Exception:
pass
try:
group_type = forge_type_id_text(definition.GetGroupTypeId())
except Exception:
pass
is_shared = False
try:
is_shared = bool(parameter.IsShared)
except Exception:
pass
shared_guid = None
if is_shared:
try:
shared_guid = str(parameter.GUID)
except Exception:
pass
unit_type = None
try:
unit_type = forge_type_id_text(parameter.GetUnitTypeId())
except Exception:
pass
try:
storage_type = str(parameter.StorageType)
except Exception:
storage_type = None
try:
is_read_only = bool(parameter.IsReadOnly)
except Exception:
is_read_only = None
try:
user_modifiable = bool(parameter.UserModifiable)
except Exception:
user_modifiable = None
return {
"scope": scope,
"name": name,
"parameterId": element_id_text(getattr(parameter, "Id", None)),
"isShared": is_shared,
"sharedGuid": shared_guid,
"storageType": storage_type,
"dataTypeId": data_type,
"groupTypeId": group_type,
"unitTypeId": unit_type,
"isReadOnly": is_read_only,
"userModifiable": user_modifiable,
"value": parameter_value(parameter),
}
def has_meaningful_value(parameter_description):
value = parameter_description.get("value", {})
return bool(value.get("hasValue")) or value.get("raw") not in (None, "") or value.get(
"display"
) not in (None, "")
def read_parameters(element, scope, include_empty):
result = []
try:
for parameter in element.Parameters:
description = describe_parameter(parameter, scope)
if include_empty or has_meaningful_value(description):
result.append(description)
except Exception as error:
return [], [safe_text(error)]
result.sort(
key=lambda parameter: (
(parameter.get("name") or "").casefold(),
parameter.get("parameterId") or "",
)
)
return result, []
def read_space(element, document):
try:
space = element.Space
except Exception as error:
return None, safe_text(error)
if space is None:
return None, None
level_name = None
try:
level = document.GetElement(space.LevelId)
level_name = None if level is None else safe_text(level.Name)
except Exception:
pass
return {
"uniqueId": safe_text(space.UniqueId),
"elementId": element_id_text(space.Id),
"number": safe_text(getattr(space, "Number", None)),
"name": safe_text(getattr(space, "Name", None)),
"levelName": level_name,
}, None
def read_family_identity(element, document):
symbol = None
try:
symbol = element.Symbol
except Exception:
try:
symbol = document.GetElement(element.GetTypeId())
except Exception:
pass
family_name = None
type_name = None
type_unique_id = None
type_element_id = None
if symbol is not None:
try:
family_name = safe_text(symbol.Family.Name)
except Exception:
family_name = safe_text(getattr(symbol, "FamilyName", None))
type_name = safe_text(getattr(symbol, "Name", None))
type_unique_id = safe_text(getattr(symbol, "UniqueId", None))
type_element_id = element_id_text(getattr(symbol, "Id", None))
return {
"familyName": family_name,
"typeName": type_name,
"typeUniqueId": type_unique_id,
"typeElementId": type_element_id,
}, symbol
def parameter_inventory_key(parameter):
stable_id = parameter.get("sharedGuid") or parameter.get("parameterId") or ""
return "|".join(
(
parameter.get("scope") or "",
stable_id,
parameter.get("name") or "",
parameter.get("dataTypeId") or "",
)
)
def sample_value_key(value):
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
def add_to_inventory(inventory, parameter, max_samples):
key = parameter_inventory_key(parameter)
entry = inventory.get(key)
if entry is None:
entry = {
"scope": parameter.get("scope"),
"name": parameter.get("name"),
"parameterId": parameter.get("parameterId"),
"isShared": parameter.get("isShared"),
"sharedGuid": parameter.get("sharedGuid"),
"storageType": parameter.get("storageType"),
"dataTypeId": parameter.get("dataTypeId"),
"groupTypeId": parameter.get("groupTypeId"),
"unitTypeId": parameter.get("unitTypeId"),
"occurrenceCount": 0,
"populatedCount": 0,
"sampleValues": [],
"_sampleKeys": set(),
}
inventory[key] = entry
entry["occurrenceCount"] += 1
if has_meaningful_value(parameter):
entry["populatedCount"] += 1
value = parameter.get("value")
value_key = sample_value_key(value)
if len(entry["sampleValues"]) < max_samples and value_key not in entry["_sampleKeys"]:
entry["_sampleKeys"].add(value_key)
entry["sampleValues"].append(value)
def finalize_inventory(inventory):
result = []
for entry in inventory.values():
clean_entry = dict(entry)
clean_entry.pop("_sampleKeys", None)
result.append(clean_entry)
return sorted(
result,
key=lambda entry: (
entry.get("scope") or "",
(entry.get("name") or "").casefold(),
entry.get("sharedGuid") or entry.get("parameterId") or "",
),
)
def loaded_assembly_versions():
result = {}
try:
from System import AppDomain
for assembly in AppDomain.CurrentDomain.GetAssemblies():
try:
name = assembly.GetName()
simple_name = str(name.Name)
if simple_name in (
"DynamoCore",
"DynamoCoreWpf",
"DynamoRevitDS",
"RevitAPI",
"RevitServices",
):
result[simple_name] = str(name.Version)
except Exception:
continue
except Exception:
pass
return dict(sorted(result.items()))
def resolve_output_path(configured_path):
timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
default_directory = os.path.join(tempfile.gettempdir(), "leistungsbilanz-dynamo")
raw_path = safe_text(configured_path)
if raw_path is None or not raw_path.strip():
directory = default_directory
file_path = os.path.join(
directory, "electrical-fixture-parameters-" + timestamp + ".json"
)
else:
expanded = os.path.abspath(os.path.expandvars(os.path.expanduser(raw_path.strip())))
if expanded.lower().endswith(".json"):
file_path = expanded
directory = os.path.dirname(file_path)
else:
directory = expanded
file_path = os.path.join(
directory, "electrical-fixture-parameters-" + timestamp + ".json"
)
if not directory:
directory = os.getcwd()
if not os.path.isdir(directory):
os.makedirs(directory)
return file_path
def write_json(file_path, payload):
temporary_path = file_path + ".tmp"
with open(temporary_path, "w", encoding="utf-8", newline="\n") as output:
json.dump(payload, output, ensure_ascii=False, indent=2, sort_keys=True)
output.write("\n")
os.replace(temporary_path, file_path)
def get_input(index, default=None):
values = globals().get("IN", [])
try:
value = values[index]
return default if value is None else value
except Exception:
return default
def build_report(include_empty, max_samples):
document = DocumentManager.Instance.CurrentDBDocument
if document is None:
raise RuntimeError("No active Revit document is available.")
if document.IsFamilyDocument:
raise RuntimeError("Open a Revit project document, not a family document.")
application = document.Application
collector = (
FilteredElementCollector(document)
.OfCategory(BuiltInCategory.OST_ElectricalFixtures)
.WhereElementIsNotElementType()
)
source_elements = list(collector)
source_elements.sort(key=lambda element: safe_text(element.UniqueId) or "")
elements = []
types_by_unique_id = {}
inventory = {}
errors = []
elements_without_space = 0
for element in source_elements:
element_errors = []
try:
family_identity, symbol = read_family_identity(element, document)
instance_parameters, parameter_errors = read_parameters(
element, "instance", include_empty
)
element_errors.extend(parameter_errors)
for parameter in instance_parameters:
add_to_inventory(inventory, parameter, max_samples)
type_unique_id = family_identity.get("typeUniqueId")
if symbol is not None and type_unique_id and type_unique_id not in types_by_unique_id:
type_parameters, type_errors = read_parameters(symbol, "type", include_empty)
element_errors.extend(type_errors)
for parameter in type_parameters:
add_to_inventory(inventory, parameter, max_samples)
types_by_unique_id[type_unique_id] = {
**family_identity,
"parameters": type_parameters,
}
space, space_error = read_space(element, document)
if space_error:
element_errors.append("MEP Space: " + space_error)
if space is None:
elements_without_space += 1
elements.append(
{
"uniqueId": safe_text(element.UniqueId),
"elementId": element_id_text(element.Id),
"categoryName": safe_text(
None if element.Category is None else element.Category.Name
),
"family": family_identity,
"space": space,
"instanceParameters": instance_parameters,
"warnings": element_errors,
}
)
except Exception as error:
errors.append(
{
"uniqueId": safe_text(getattr(element, "UniqueId", None)),
"elementId": element_id_text(getattr(element, "Id", None)),
"error": safe_text(error),
}
)
return {
"reportSchemaVersion": 1,
"generatedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"readOnly": True,
"complete": len(errors) == 0,
"scope": {
"builtInCategory": "OST_ElectricalFixtures",
"wholeDocument": True,
"elementTypesExcluded": True,
"includeEmptyParameters": include_empty,
},
"environment": {
"revitVersionNumber": safe_text(getattr(application, "VersionNumber", None)),
"revitVersionName": safe_text(getattr(application, "VersionName", None)),
"revitSubVersionNumber": safe_text(
getattr(application, "SubVersionNumber", None)
),
"pythonImplementation": safe_text(getattr(sys.implementation, "name", None)),
"pythonVersion": platform.python_version(),
"assemblies": loaded_assembly_versions(),
},
"document": {
"title": safe_text(document.Title),
"pathName": safe_text(document.PathName),
"projectInformationUniqueId": safe_text(document.ProjectInformation.UniqueId),
},
"summary": {
"elementCount": len(source_elements),
"exportedElementCount": len(elements),
"typeCount": len(types_by_unique_id),
"parameterDefinitionCount": len(inventory),
"elementsWithoutMepSpace": elements_without_space,
"elementErrorCount": len(errors),
},
"parameterInventory": finalize_inventory(inventory),
"types": sorted(
types_by_unique_id.values(),
key=lambda entry: (
(entry.get("familyName") or "").casefold(),
(entry.get("typeName") or "").casefold(),
entry.get("typeUniqueId") or "",
),
),
"elements": elements,
"errors": errors,
}
try:
include_empty_input = get_input(1, True)
include_empty = bool(include_empty_input)
try:
max_samples = int(get_input(2, 5))
except Exception:
max_samples = 5
max_samples = max(0, min(max_samples, 50))
report = build_report(include_empty, max_samples)
output_path = resolve_output_path(get_input(0))
write_json(output_path, report)
OUT = {
"ok": True,
"filePath": output_path,
"complete": report["complete"],
"summary": report["summary"],
"errors": report["errors"],
}
except Exception as error:
OUT = {
"ok": False,
"error": safe_text(error),
"traceback": traceback.format_exc(),
}