# Bioclimatic Explorer: send this Revit model to the page, from Dynamo. Paste into a Python Script node (CPython 3), # connect a String node with the link code shown on the page (site card, Link a program) to IN[0] and a Boolean # to IN[1]; set it to True to send. Nothing is installed. # Generated by tools/build-relay.py from relay/src/; edit there. # ---- Bioclimatic Explorer link: the common part (relay/src/core.py; built into each script by tools/build-relay.py) ---- # Python 3 only, standard library only: AES-256-GCM written out here (NIST FIPS 197 and SP 800-38D), so nothing has # to be installed in Rhino's or Dynamo's Python. The model is compressed (gzip), encrypted with the link's key and # written to the link's room; the server keeps only ciphertext. The page that made the link decrypts it. import base64, gzip, hashlib, json, os, struct, time # AES (FIPS 197): the S-box built from the field inverse and the affine map, then four T-tables for the rounds. def _aes_tables(): exp, log = [0] * 512, [0] * 256 x = 1 for i in range(255): exp[i] = x; log[x] = i x ^= (x << 1) ^ (0x11B if x & 0x80 else 0) x &= 0xFF for i in range(255, 512): exp[i] = exp[i - 255] sbox = [0] * 256 for i in range(256): inv = 0 if i == 0 else exp[255 - log[i]] s = inv for k in range(1, 5): s ^= ((inv << k) | (inv >> (8 - k))) & 0xFF sbox[i] = s ^ 0x63 xt = lambda a: ((a << 1) ^ (0x1B if a & 0x80 else 0)) & 0xFF T0 = [0] * 256 for i in range(256): s = sbox[i]; s2 = xt(s); s3 = s2 ^ s T0[i] = (s2 << 24) | (s << 16) | (s << 8) | s3 ror = lambda w, n: ((w >> n) | (w << (32 - n))) & 0xFFFFFFFF return sbox, T0, [ror(t, 8) for t in T0], [ror(t, 16) for t in T0], [ror(t, 24) for t in T0] _SBOX, _T0, _T1, _T2, _T3 = _aes_tables() def _expand(key): nk, nr = len(key) // 4, len(key) // 4 + 6 w = list(struct.unpack(">%dI" % nk, key)) rcon = 1 for i in range(nk, 4 * (nr + 1)): t = w[i - 1] if i % nk == 0: t = ((t << 8) | (t >> 24)) & 0xFFFFFFFF t = (_SBOX[t >> 24] << 24) | (_SBOX[(t >> 16) & 255] << 16) | (_SBOX[(t >> 8) & 255] << 8) | _SBOX[t & 255] t ^= rcon << 24 rcon = ((rcon << 1) ^ (0x1B if rcon & 0x80 else 0)) & 0xFF elif nk > 6 and i % nk == 4: t = (_SBOX[t >> 24] << 24) | (_SBOX[(t >> 16) & 255] << 16) | (_SBOX[(t >> 8) & 255] << 8) | _SBOX[t & 255] w.append(w[i - nk] ^ t) return w, nr def _block(w, nr, s0, s1, s2, s3): T0, T1, T2, T3, S = _T0, _T1, _T2, _T3, _SBOX s0 ^= w[0]; s1 ^= w[1]; s2 ^= w[2]; s3 ^= w[3] k = 4 for _ in range(nr - 1): t0 = T0[s0 >> 24] ^ T1[(s1 >> 16) & 255] ^ T2[(s2 >> 8) & 255] ^ T3[s3 & 255] ^ w[k] t1 = T0[s1 >> 24] ^ T1[(s2 >> 16) & 255] ^ T2[(s3 >> 8) & 255] ^ T3[s0 & 255] ^ w[k + 1] t2 = T0[s2 >> 24] ^ T1[(s3 >> 16) & 255] ^ T2[(s0 >> 8) & 255] ^ T3[s1 & 255] ^ w[k + 2] t3 = T0[s3 >> 24] ^ T1[(s0 >> 16) & 255] ^ T2[(s1 >> 8) & 255] ^ T3[s2 & 255] ^ w[k + 3] s0, s1, s2, s3 = t0, t1, t2, t3 k += 4 o0 = ((S[s0 >> 24] << 24) | (S[(s1 >> 16) & 255] << 16) | (S[(s2 >> 8) & 255] << 8) | S[s3 & 255]) ^ w[k] o1 = ((S[s1 >> 24] << 24) | (S[(s2 >> 16) & 255] << 16) | (S[(s3 >> 8) & 255] << 8) | S[s0 & 255]) ^ w[k + 1] o2 = ((S[s2 >> 24] << 24) | (S[(s3 >> 16) & 255] << 16) | (S[(s0 >> 8) & 255] << 8) | S[s1 & 255]) ^ w[k + 2] o3 = ((S[s3 >> 24] << 24) | (S[(s0 >> 16) & 255] << 16) | (S[(s1 >> 8) & 255] << 8) | S[s2 & 255]) ^ w[k + 3] return (o0 << 96) | (o1 << 64) | (o2 << 32) | o3 def _enc_int(w, nr, x): return _block(w, nr, (x >> 96) & 0xFFFFFFFF, (x >> 64) & 0xFFFFFFFF, (x >> 32) & 0xFFFFFFFF, x & 0xFFFFFFFF) # GCM (SP 800-38D): the hash key's products tabled byte by byte (GHASH is linear), the counter mode from inc32(J0). _R = 0xE1 << 120 def _gmul(x, y): z, v = 0, y for i in range(127, -1, -1): if (x >> i) & 1: z ^= v v = (v >> 1) ^ _R if v & 1 else v >> 1 return z def aes_gcm_encrypt(key, iv, plain): """AES-GCM with a 96-bit IV and no additional data: ciphertext followed by the 16-byte tag (WebCrypto's form).""" w, nr = _expand(key) H = _enc_int(w, nr, 0) bit = [_gmul(1 << s, H) for s in range(128)] tab = [] for j in range(16): t = [0] * 256 for b in range(256): acc = 0 for k in range(8): if b >> k & 1: acc ^= bit[8 * (15 - j) + k] t[b] = acc tab.append(t) def ghash_mul(x): z = 0 for j in range(16): z ^= tab[j][(x >> (8 * (15 - j))) & 255] return z j0 = (int.from_bytes(iv, "big") << 32) | 1 n = len(plain) parts = [] hi, lo = j0 & ~0xFFFFFFFF, j0 & 0xFFFFFFFF y = 0 for i in range(0, n, 16): lo = (lo + 1) & 0xFFFFFFFF ks = _enc_int(w, nr, hi | lo) chunk = plain[i:i + 16] m = len(chunk) if m == 16: c = int.from_bytes(chunk, "big") ^ ks cb = c.to_bytes(16, "big") else: cb = (int.from_bytes(chunk, "big") ^ (ks >> (8 * (16 - m)))).to_bytes(m, "big") c = int.from_bytes(cb.ljust(16, b"\0"), "big") parts.append(cb) y = ghash_mul(y ^ c) out = b"".join(parts) y = ghash_mul(y ^ (n * 8)) # len(A) = 0 in the high half, len(C) in bits in the low half tag = (_enc_int(w, nr, j0) ^ y).to_bytes(16, "big") return out + tag def _b64u(b): return base64.urlsafe_b64encode(b).decode().rstrip("=") def _unb64u(t): return base64.urlsafe_b64decode(t + "=" * (-len(t) % 4)) def parse_code(code): """The page's link code: /rooms/#..""" code = (code or "").strip() try: url, frag = code.split("#", 1) base, rid = url.rstrip("/").rsplit("/rooms/", 1) key, token = frag.split(".", 1) k = _unb64u(key) if len(k) != 32 or len(rid) != 12: raise ValueError() return {"base": base, "id": rid, "key": k, "token": token} except Exception: raise ValueError("not a link code: copy it from the page (Link a program), it reads /rooms/#.") def mesh_object(name, verts, tris): """One object: verts a flat list x, y, z (model units), tris a flat list of vertex indices, three a triangle.""" return {"name": name, "v": base64.b64encode(struct.pack("<%df" % len(verts), *verts)).decode(), "f": base64.b64encode(struct.pack("<%dI" % len(tris), *tris)).decode()} MAX = 1400000 # the room keeps a document under 1.5 MB of text def pack(code, objects, app, name="", unit=1.0, up="z", lat=None, lon=None, north=None, unit_name=""): """The message, compressed and encrypted: the text to write to the room.""" c = parse_code(code) msg = {"cc": "model", "v": 1, "app": app, "name": name, "unit": unit, "unitName": unit_name, "up": up, "lat": lat, "lon": lon, "north": north, "t": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "objects": objects} plain = gzip.compress(json.dumps(msg, separators=(",", ":")).encode(), 6) iv = os.urandom(12) doc = _b64u(iv + aes_gcm_encrypt(c["key"], iv, plain)) if len(doc) > MAX: raise ValueError("the model is too large to send (%.1f MB compressed and encrypted; the limit is 1.4 MB): send fewer objects, or simpler meshes" % (len(doc) / 1e6)) return c, doc def _http(method, url, body=None): data = json.dumps(body).encode() if body is not None else None try: import urllib.request req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=60) as r: return r.status, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: return e.code, json.loads(e.read() or b"{}") except Exception as err: # Python without certificates (some hosts' embedded Python): .NET's HTTP client, where .NET is there if "CERTIFICATE" not in str(err).upper() and "SSL" not in str(err).upper(): raise import clr clr.AddReference("System.Net.Http") from System.Net.Http import HttpClient, StringContent, HttpMethod, HttpRequestMessage from System.Text import Encoding cl = HttpClient() rq = HttpRequestMessage(HttpMethod(method), url) if data is not None: rq.Content = StringContent(data.decode(), Encoding.UTF8, "application/json") rs = cl.SendAsync(rq).Result return int(rs.StatusCode), json.loads(rs.Content.ReadAsStringAsync().Result or "{}") def send(code, objects, app, **meta): """Packs the model and writes it to the link's room (the version read first, a conflict retried).""" c, doc = pack(code, objects, app, **meta) room = "%s/rooms/%s" % (c["base"], c["id"]) for _ in range(4): st, o = _http("GET", room + "?since=-2&pv=-2") if st == 404: raise ValueError("this link no longer exists: make a new one on the page") st, o = _http("POST", room, {"w": c["token"], "base": o.get("v", 0), "doc": doc}) if st == 200: return {"version": o.get("v"), "bytes": len(doc)} if st == 403: raise ValueError("the link code's token is wrong: copy the code again") if st != 409: raise ValueError("the server answered %s: %s" % (st, o.get("error", ""))) raise ValueError("the room kept changing; try again") # ---- end of the common part ------------------------------------------------------------------------------------------ # ---- Revit through Dynamo: the model --------------------------------------------------------------------------------- # In a Python Script node (CPython 3) with IN[0] the link code (a String node) and IN[1] a Boolean to send; OUT says what # was sent. The building's elements (walls, floors, roofs, columns, curtain panels and mullions, windows, doors, stairs, # railings, masses, generic models; their solids' faces triangulated by Revit), one object per element. Units feet; # the place from the project's site location; north from the project position's angle (from memory, unchecked: the # page lets north be set by hand). import clr, math clr.AddReference("RevitAPI") clr.AddReference("RevitServices") from Autodesk.Revit import DB from RevitServices.Persistence import DocumentManager from System.Collections.Generic import List CATS = ["OST_Walls", "OST_Floors", "OST_Roofs", "OST_Columns", "OST_StructuralColumns", "OST_StructuralFraming", "OST_CurtainWallPanels", "OST_CurtainWallMullions", "OST_Windows", "OST_Doors", "OST_Stairs", "OST_StairsRailing", "OST_Mass", "OST_GenericModel"] def _add(m, verts, tris): o = len(verts) // 3 for p in m.Vertices: verts += [p.X, p.Y, p.Z] for i in range(m.NumTriangles): t = m.get_Triangle(i) tris += [o + int(t.get_Index(0)), o + int(t.get_Index(1)), o + int(t.get_Index(2))] def _collect(geo, verts, tris): for g in geo: if isinstance(g, DB.Solid): if g.Volume <= 0: continue for f in g.Faces: m = f.Triangulate() if m: _add(m, verts, tris) elif isinstance(g, DB.GeometryInstance): _collect(g.GetInstanceGeometry(), verts, tris) elif isinstance(g, DB.Mesh): _add(g, verts, tris) def run(code): doc = DocumentManager.Instance.CurrentDBDocument cats = List[DB.BuiltInCategory]([getattr(DB.BuiltInCategory, c) for c in CATS if hasattr(DB.BuiltInCategory, c)]) els = DB.FilteredElementCollector(doc).WherePasses(DB.ElementMulticategoryFilter(cats)).WhereElementIsNotElementType().ToElements() opt = DB.Options(); opt.DetailLevel = DB.ViewDetailLevel.Medium; opt.IncludeNonVisibleObjects = False out = [] for e in els: geo = e.get_Geometry(opt) if geo is None: continue verts, tris = [], [] _collect(geo, verts, tris) if tris: out.append(mesh_object("%s %s" % (e.Category.Name if e.Category else "", e.Id), verts, tris)) if not out: return "Nothing to send: no walls, floors, roofs or other building elements with solids" site = doc.SiteLocation lat, lon = math.degrees(site.Latitude), math.degrees(site.Longitude) north = None try: north = (math.degrees(doc.ActiveProjectLocation.GetProjectPosition(DB.XYZ.Zero).Angle)) % 360 except Exception: pass r = send(code, out, "Revit", name=doc.Title or "", unit=0.3048, up="z", lat=lat, lon=lon, north=north, unit_name="feet") return "Sent %d elements (%d kB) to the page." % (len(out), r["bytes"] // 1000) try: OUT = run(IN[0]) if (len(IN) < 2 or IN[1]) else "Set the Boolean to True to send" except ValueError as err: OUT = str(err)