feat: add cesium traffic signal countdowns
This commit is contained in:
369
blender/osmassets/traffic_signals.py
Normal file
369
blender/osmassets/traffic_signals.py
Normal file
@@ -0,0 +1,369 @@
|
||||
"""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))
|
||||
Reference in New Issue
Block a user