"""Static traffic-signal geometry for the main Blender scene. The anchor file is generated by the intermediates stage. Cesium consumes the same anchors for its dynamic lenses and countdown digits, so this module only creates the durable structure around them. """ import math import os from osmassets.mesh import MeshBatch DEFAULT_LAYOUT = { "poleHeightMeters": 6.7, "poleRadiusMeters": 0.13, "armWidthMeters": 0.21, "mastHeightMeters": 6.25, "headCenterHeightMeters": 6.25, "headWidthMeters": 0.68, "headDepthMeters": 0.30, "headBodyHeightMeters": 1.62, "lensRadiusMeters": 0.22, "lensDepthMeters": 0.07, "lensFaceOffsetMeters": 0.18, "lensVerticalOffsetsMeters": [0.49, -0.01, -0.51], "countdownLateralMeters": 1.15, "countdownFaceOffsetMeters": 0.05, "countdownWidthMeters": 0.82, "countdownDepthMeters": 0.14, "countdownHeightMeters": 0.56, "countdownVerticalOffsetMeters": 0.0, } COUNTDOWN_VALUES = tuple("%02d" % value for value in range(20)) COUNTDOWN_FONT_PATH = os.path.normpath(os.path.join( os.path.dirname(__file__), "..", "..", "assets", "fonts", "7LED-1.ttf")) def assemble(signal_data, projector, collection, materials): """Add batched static signal structures and return the accepted count.""" metal = MeshBatch("Traffic Signal Metal", collection, materials["metal"]) housing = MeshBatch("Traffic Signal Housing", collection, materials["housing"]) lenses = { state: MeshBatch("Traffic Signal %s Lens" % state.title(), collection, material) for state, material in materials["lenses"].items() } layout = _layout(signal_data.get("layout")) count = 0 for signal in signal_data.get("signals", []): if not _valid_signal(signal): continue pose = signal.get("pose") if _valid_pose(signal.get("pose")) else None if pose: x, y = projector.xy((pose["pole"]["longitude"], pose["pole"]["latitude"])) head_x, head_y = projector.xy((pose["head"]["longitude"], pose["head"]["latitude"])) face_heading = math.radians(pose["head"]["faceHeadingDegrees"]) face = (math.sin(face_heading), math.cos(face_heading)) lateral = (-math.cos(face_heading), math.sin(face_heading)) else: x, y = projector.xy((signal["longitude"], signal["latitude"])) heading = math.radians(signal["headingDegrees"]) longitudinal = (math.sin(heading), math.cos(heading)) lateral = (math.cos(heading), -math.sin(heading)) face = (-longitudinal[0], -longitudinal[1]) mast_reach = float(signal.get("mastReachMeters") or 4.5) head_x, head_y = _offset(x, y, lateral, -mast_reach) _add_cylinder(metal, x, y, layout["poleHeightMeters"] / 2, layout["poleRadiusMeters"], layout["poleHeightMeters"]) _add_box(metal, (x, y), (head_x, head_y), layout["armWidthMeters"] / 2, layout["mastHeightMeters"] - layout["armWidthMeters"] / 2, layout["armWidthMeters"]) _add_oriented_box( housing, head_x, head_y, lateral, face, layout["headWidthMeters"], layout["headDepthMeters"], layout["headCenterHeightMeters"], layout["headBodyHeightMeters"], ) for index, state in enumerate(("red", "yellow", "green")): if pose: lens_x, lens_y = projector.xy((pose["lenses"][index]["longitude"], pose["lenses"][index]["latitude"])) lens_z = pose["lenses"][index]["height"] else: lens_x, lens_y = _offset(head_x, head_y, face, layout["lensFaceOffsetMeters"]) lens_z = layout["headCenterHeightMeters"] + layout["lensVerticalOffsetsMeters"][index] _add_lens( lenses[state], lens_x, lens_y, lens_z, lateral, face, layout["lensRadiusMeters"], layout["lensDepthMeters"], 10, ) if pose: board_x, board_y = projector.xy((pose["countdown"]["longitude"], pose["countdown"]["latitude"])) board_z = pose["countdown"]["height"] else: board_x, board_y = _offset(head_x, head_y, lateral, layout["countdownLateralMeters"]) board_x, board_y = _offset(board_x, board_y, face, layout["countdownFaceOffsetMeters"]) board_z = layout["mastHeightMeters"] + layout["countdownVerticalOffsetMeters"] _add_oriented_box(housing, board_x, board_y, lateral, face, layout["countdownWidthMeters"], layout["countdownDepthMeters"], board_z, layout["countdownHeightMeters"]) count += 1 metal.finish() housing.finish() for batch in lenses.values(): batch.finish() return count def assemble_dynamic(signal_data, projector, collection, materials): """Build phase meshes plus instanced font countdowns for Cesium.""" layout = _layout(signal_data.get("layout")) objects = [] countdown_materials = materials.get("countdown") or {} if not countdown_materials: raise RuntimeError("Traffic signal countdown material is not configured") countdown_meshes = _countdown_meshes(countdown_materials) for signal in signal_data.get("signals", []): if not _valid_signal(signal) or not _valid_pose(signal.get("pose")): continue pose = signal["pose"] face_heading = math.radians(pose["head"]["faceHeadingDegrees"]) face = (math.sin(face_heading), math.cos(face_heading)) lateral = (-math.cos(face_heading), math.sin(face_heading)) # The static lenses already occupy the head face. Dynamic emissive # covers must sit just in front of them or the static material wins the # depth test and masks every phase change. active_lens_depth = min(0.025, layout["lensDepthMeters"]) active_lens_radius = layout["lensRadiusMeters"] * 0.88 active_lens_offset = (layout["lensDepthMeters"] + active_lens_depth) / 2 + 0.003 for state in ("red", "yellow", "green"): batch = MeshBatch("TrafficSignalDynamic_%s_%s" % (signal["id"], state), collection, materials[state]) for index in (0, 1, 2): point = pose["lenses"][index] if point["state"] == state: x, y = projector.xy((point["longitude"], point["latitude"])) x, y = _offset(x, y, face, active_lens_offset) _add_lens(batch, x, y, point["height"], lateral, face, active_lens_radius, active_lens_depth, 10) obj = batch.finish() if obj: objects.append(obj) board = pose["countdown"] board_x, board_y = projector.xy((board["longitude"], board["latitude"])) board_z = board["height"] text_x, text_y = _offset( board_x, board_y, face, layout["countdownDepthMeters"] / 2 + 0.008) phase_group = int(signal.get("phaseGroup") or 0) % 2 for value, mesh in countdown_meshes[phase_group].items(): objects.append(_countdown_instance( "TrafficSignalDynamic_%s_countdown_%s" % (signal["id"], value), mesh, collection, text_x, text_y, board_z, lateral, face)) return objects def _countdown_meshes(materials): """Create 20 inverted font meshes per phase group, shared by all signals.""" try: import bpy except ImportError: # Geometry unit tests run in CPython without Blender. Their lens checks # remain useful while the actual font conversion is Blender-only. return {group: {} for group in materials} if not os.path.exists(COUNTDOWN_FONT_PATH): raise RuntimeError("Traffic signal countdown font not found: %s" % COUNTDOWN_FONT_PATH) font = bpy.data.fonts.load(COUNTDOWN_FONT_PATH, check_existing=True) meshes = {group: {} for group in materials} for group, material in materials.items(): for value in COUNTDOWN_VALUES: meshes[group][value] = _inverted_countdown_mesh(value, group, font, material) return meshes def _inverted_countdown_mesh(value, group, font, material): """Turn 7LED's dark glyph cut-out into the emissive number geometry.""" import bpy curve = bpy.data.curves.new("TrafficSignalCountdown_%s_%s" % (group, value), "FONT") curve.body = value curve.font = font curve.align_x = "CENTER" curve.align_y = "CENTER" curve.size = 0.44 curve.extrude = 0.004 curve.resolution_u = 1 text = bpy.data.objects.new("TrafficSignalCountdownTemplate_%s_%s" % (group, value), curve) bpy.context.scene.collection.objects.link(text) bpy.context.view_layer.objects.active = text text.select_set(True) bpy.ops.object.convert(target="CURVE") glyph = bpy.context.view_layer.objects.active vertices = [] faces = [] depth = 0.008 for spline in glyph.data.splines: if _spline_area(spline) >= 0: continue loop = _sample_bezier_loop(spline) if len(loop) < 3: continue start = len(vertices) vertices.extend((x, y, -depth / 2) for x, y in loop) vertices.extend((x, y, depth / 2) for x, y in loop) count = len(loop) faces.append(tuple(reversed(range(start, start + count)))) faces.append(tuple(range(start + count, start + count * 2))) for index in range(count): next_index = (index + 1) % count faces.append((start + index, start + next_index, start + count + next_index, start + count + index)) if not vertices: raise RuntimeError("7LED font contains no digit cut-outs for %s" % value) mesh = bpy.data.meshes.new("TrafficSignalCountdownMesh_%s_%s" % (group, value)) mesh.from_pydata(vertices, [], faces) mesh.materials.append(material) mesh.update() mesh.name = "TrafficSignalCountdownMesh_%s_%s" % (group, value) bpy.data.objects.remove(glyph, do_unlink=True) return mesh def _spline_area(spline): if spline.type != "BEZIER" or len(spline.bezier_points) < 3: return 0 points = spline.bezier_points return sum( point.co.x * points[(index + 1) % len(points)].co.y - points[(index + 1) % len(points)].co.x * point.co.y for index, point in enumerate(points) ) / 2 def _sample_bezier_loop(spline, samples_per_edge=8): points = spline.bezier_points result = [] for index, start in enumerate(points): end = points[(index + 1) % len(points)] p0 = start.co p1 = start.handle_right p2 = end.handle_left p3 = end.co for step in range(samples_per_edge): t = step / samples_per_edge inverse = 1 - t result.append(( inverse ** 3 * p0.x + 3 * inverse ** 2 * t * p1.x + 3 * inverse * t ** 2 * p2.x + t ** 3 * p3.x, inverse ** 3 * p0.y + 3 * inverse ** 2 * t * p1.y + 3 * inverse * t ** 2 * p2.y + t ** 3 * p3.y, )) return result def _countdown_instance(name, mesh, collection, x, y, z, across, face): import bpy from mathutils import Matrix obj = bpy.data.objects.new(name, mesh) collection.objects.link(obj) # Text geometry starts in the local XY plane. Map X across the board, Y # upward, and its front normal toward the same approach-facing axis as the # static housing and dynamic lenses. obj.matrix_world = Matrix((( (across[0], 0.0, face[0], x), (across[1], 0.0, face[1], y), (0.0, 1.0, 0.0, z), (0.0, 0.0, 0.0, 1.0), ))) return obj def _valid_signal(signal): if not isinstance(signal, dict): return False try: return all(math.isfinite(float(signal.get(key))) for key in ("longitude", "latitude", "headingDegrees")) except (TypeError, ValueError): return False def _valid_pose(pose): try: return (isinstance(pose, dict) and len(pose.get("lenses", [])) == 3 and all(math.isfinite(float(pose[key]["longitude"])) and math.isfinite(float(pose[key]["latitude"])) for key in ("pole", "head", "countdown"))) except (KeyError, TypeError, ValueError): return False def _layout(value): layout = dict(DEFAULT_LAYOUT) if not isinstance(value, dict): return layout for key, default in DEFAULT_LAYOUT.items(): candidate = value.get(key) if isinstance(default, list): if (isinstance(candidate, list) and len(candidate) == len(default) and all(isinstance(item, (int, float)) and math.isfinite(item) for item in candidate)): layout[key] = candidate elif (isinstance(candidate, (int, float)) and math.isfinite(candidate) and (key == "countdownVerticalOffsetMeters" or candidate > 0)): layout[key] = candidate return layout def _offset(x, y, direction, distance): return x + direction[0] * distance, y + direction[1] * distance def _add_box(batch, start, end, width, base, height): dx, dy = end[0] - start[0], end[1] - start[1] length = math.hypot(dx, dy) if length <= 0: return across = (-dy / length, dx / length) half = width / 2 ring = [ (start[0] + across[0] * half, start[1] + across[1] * half), (end[0] + across[0] * half, end[1] + across[1] * half), (end[0] - across[0] * half, end[1] - across[1] * half), (start[0] - across[0] * half, start[1] - across[1] * half), ] batch.add_prism(ring, base, height) def _add_oriented_box(batch, x, y, across, depth, width, thickness, center_z, height): half_width = width / 2 half_depth = thickness / 2 ring = [ (x + across[0] * sx * half_width + depth[0] * sy * half_depth, y + across[1] * sx * half_width + depth[1] * sy * half_depth) for sx, sy in ((-1, -1), (1, -1), (1, 1), (-1, 1)) ] batch.add_prism(ring, center_z - height / 2, height) def _add_cylinder(batch, x, y, center_z, radius, height, sides=8): ring = [ (x + math.cos(math.tau * index / sides) * radius, y + math.sin(math.tau * index / sides) * radius) for index in range(sides) ] batch.add_prism(ring, center_z - height / 2, height) def _add_lens(batch, x, y, z, across, face, radius, depth, sides): """Add a shallow round lens flush with the head's approach-facing surface.""" start = len(batch.vertices) for face_offset in (-depth / 2, depth / 2): for index in range(sides): theta = math.tau * index / sides batch.vertices.append(( x + face[0] * face_offset + across[0] * math.cos(theta) * radius, y + face[1] * face_offset + across[1] * math.cos(theta) * radius, z + math.sin(theta) * radius, )) batch.faces.append(tuple(range(start, start + sides))) batch.faces.append(tuple(range(start + sides, start + sides * 2))) for index in range(sides): next_index = (index + 1) % sides a = start + index b = start + next_index c = start + sides + next_index d = start + sides + index batch.faces.append((a, b, c, d))