Skip to content
All posts

Unreal Engine

How to build view-specific proxies in Unreal Engine

In this blog we will use the Simplygon Subsystem to build optimized level geometry for every view position in an Unreal Engine world. Static meshes are divided into distance buckets, processed with aggregation or reduction, and culled against cameras placed around each view position.

How to build view-specific proxies in Unreal Engine

Disclaimer: The code in this post is written using version 10.4.446.0 of Simplygon and Unreal Engine 5.7.4. If you encounter this post at a later stage, some of the API calls might have changed. However, the core concepts should still remain valid.

Problem to solve

We have a game that takes place around very restricted camera positions. This could for example be a virtual reality game.

The player can only move the camera within 1 meter in the specific position and are restricted to look in one direction.

We want to generate a specialized representation of the world for each ViewPosition actor. In this image those are represented by white boxes.

World with view positions marked

Solution

Our idea is that every view position will have it's own set of meshes optimized specifically for it. Close to it we will have aggregated geometry, and further away reduced geometry. We divide the scene by distance into different buckets.

We will also use visibility culling, where buckets closer to the view position will occlude buckets further away.

View position
  -> near aggregation bucket
  -> reduction bucket 1
  -> reduction bucket 2
  -> reduction bucket 3
  -> far reduction bucket

The nearest bucket uses aggregation to just combine and cull geometry, while every farther bucket uses reduction driven by a maximum-deviation target.

Configure optimization buckets

Each bucket has an inclusive distance threshold and an optimization method. Reduction buckets also specify a maximum allowed geometric deviation.

class OptimizationMethod(Enum):
    AGGREGATION = "aggregation"
    REDUCTION = "reduction"


@dataclass(frozen=True)
class BucketSettings:
    method: OptimizationMethod
    distance_threshold: float
    max_deviation: float = 10.0


BUCKET_SETTINGS = [
    BucketSettings(OptimizationMethod.AGGREGATION, distance_threshold=2000.0),
    BucketSettings(
        OptimizationMethod.REDUCTION,
        distance_threshold=5000.0,
        max_deviation=20.0,
    ),
    BucketSettings(
        OptimizationMethod.REDUCTION,
        distance_threshold=15000.0,
        max_deviation=60.0,
    ),
    BucketSettings(
        OptimizationMethod.REDUCTION,
        distance_threshold=30000.0,
        max_deviation=100.0,
    ),
    BucketSettings(
        OptimizationMethod.REDUCTION,
        distance_threshold=math.inf,
        max_deviation=110.0,
    ),
]

Distances use Unreal units, which are centimeters by default. The final infinite threshold ensures that every eligible mesh is assigned. The script validates that thresholds increase and that every reduction bucket has a positive maximum deviation.

Find view positions and source geometry

We use an actor called ViewPosition to mark out the different views. We have added an arrow to it so we can see in which position it faces.

View position actor

VIEW_BLUEPRINT_CLASS_PATH = ""
VIEW_BLUEPRINT_CLASS_NAME = "ViewPosition"

The optimizer first finds all view actors.

def _find_view_actors(self) -> List[unreal.Actor]:
    actors = unreal.get_editor_subsystem(
        unreal.EditorActorSubsystem
    ).get_all_level_actors()

    if VIEW_BLUEPRINT_CLASS_PATH:
        view_class = unreal.load_class(None, VIEW_BLUEPRINT_CLASS_PATH)
        if view_class is None:
            raise RuntimeError(
                f"Could not load view class '{VIEW_BLUEPRINT_CLASS_PATH}'."
            )
        return [actor for actor in actors if isinstance(actor, view_class)]

    expected_names = {
        VIEW_BLUEPRINT_CLASS_NAME,
        f"{VIEW_BLUEPRINT_CLASS_NAME}_C",
        f"BP_{VIEW_BLUEPRINT_CLASS_NAME}_C",
    }
    return [
        actor
        for actor in actors
        if actor.get_class().get_name() in expected_names
    ]
}

It then gathers static mesh components from the current editor world. View actors, known sky actors, movable components, and actors generated by an earlier optimizer run are excluded.

def _find_static_mesh_components(
    self, view_actors: List[unreal.Actor]
) -> List[unreal.StaticMeshComponent]:
    view_paths = {actor.get_path_name() for actor in view_actors}
    world = unreal.get_editor_subsystem(
        unreal.UnrealEditorSubsystem
    ).get_editor_world()
    if world is None:
        raise RuntimeError("Could not get the current editor world.")
    actors = unreal.GameplayStatics.get_all_actors_of_class(world, unreal.Actor)
    components = []
    for actor in actors:
        if (
            actor.get_path_name() in view_paths
            or actor.actor_has_tag(GENERATED_ACTOR_TAG)
            or actor.get_class().get_name() in EXCLUDED_ACTOR_CLASS_NAMES
        ):
            continue
        for component in actor.get_components_by_class(unreal.StaticMeshComponent):
            if (
                component.get_editor_property("static_mesh") is not None
                and component.get_editor_property("mobility")
                == unreal.ComponentMobility.STATIC
            ):
                components.append(component)
    return components

Assign meshes to jobs

For each view, the script measures the distance from the view location to every eligible static mesh component. The component is placed in the first bucket whose threshold contains that distance.

for component in components:
    distance = (component.get_world_location() - view_location).length()
    for bucket_index, settings in enumerate(BUCKET_SETTINGS):
        if distance <= settings.distance_threshold:
            buckets[bucket_index].append(component)
            break

The resulting OptimizationJob also stores the view's location and orientation vectors. Those vectors are later used to construct the visibility camera positions.

Before processing starts, the script writes a bucket report to the Unreal log. This is useful for verifying thresholds and finding unexpectedly assigned components.

LogPython: import simplygon_view_optimizer as view_optimizer
LogPython: view_optimizer.optimize_views()
LogPython: ========== View Bucket Report ==========
LogPython: View 1 | Bucket 1 | <= 2000 cm | aggregation | 214 mesh components
LogPython:   SM_building_border_187 (/Game/Stylized_Egypt/Maps/Stylized_Egypt_Demo.Stylized_Egypt_Demo:PersistentLevel.SM_building_border_213) | StaticMeshComponent0=1805 cm
LogPython:   SM_building_border_404 (/Game/Stylized_Egypt/Maps/Stylized_Egypt_Demo.Stylized_Egypt_Demo:PersistentLevel.SM_building_border_340) | StaticMeshComponent0=699 cm

...

LogPython: View 1 | Bucket 2 | <= 5000 cm | reduction | 992 mesh components
LogPython:   SM_block_11 (/Game/Stylized_Egypt/Maps/Stylized_Egypt_Demo.Stylized_Egypt_Demo:PersistentLevel.SM_block_11) | StaticMeshComponent0=2382 cm
LogPython:   SM_building_border_03 (/Game/Stylized_Egypt/Maps/Stylized_Egypt_Demo.Stylized_Egypt_Demo:PersistentLevel.SM_building_border_03_5) | StaticMeshComponent0=3890 cm
LogPython:   SM_building_border_04 (/Game/Stylized_Egypt/Maps/Stylized_Egypt_Demo.Stylized_Egypt_Demo:PersistentLevel.SM_building_border_04_47) | StaticMeshComponent0=4997 cm

...

LogPython: View 2 | Bucket 4 | <= 30000 cm | reduction | 2 mesh components
LogPython:   SM_pyramid_03 (/Game/Stylized_Egypt/Maps/Stylized_Egypt_Demo.Stylized_Egypt_Demo:PersistentLevel.SM_pyramid_03_20) | StaticMeshComponent0=18107 cm
LogPython:   SM_pyramid_4 (/Game/Stylized_Egypt/Maps/Stylized_Egypt_Demo.Stylized_Egypt_Demo:PersistentLevel.SM_pyramid_4) | StaticMeshComponent0=19751 cm
LogPython: View 2 | Bucket 5 | <= infinity cm | reduction | 1 mesh components
LogPython:   SM_pyramid_5 (/Game/Stylized_Egypt/Maps/Stylized_Egypt_Demo.Stylized_Egypt_Demo:PersistentLevel.SM_pyramid_0) | StaticMeshComponent0=47116 cm
LogPython: ========== End View Bucket Report ==========
LogPython: Starting 10 Simplygon optimization jobs.

Export from Unreal Engine

The Simplygon Subsystem exports the static mesh components for one populated bucket to an intermediate .sgscene file.

job_id = subsystem.export_static_mesh_components_to_scene(
    output_path,
    job.components,
    get_default_scene_export_settings(False),
)
self.export_jobs[job_id.to_string()] = job

Export is asynchronous. The returned task ID is associated with its job, and _on_export_completed resumes processing when Unreal signals completion. Materials are not baked because the generated geometry keeps the source material setup.

Build the visibility scene

After export, the script opens the scene with the Simplygon API and creates three selection sets:

  • Process selection set: geometry in the current distance bucket;
  • Occluder selection set: already processed, closer buckets for the same view; and
  • Camera selection set: the camera used by Simplygon visibility processing.

The current bucket is recursively added to the process selection set. Processed scenes from closer buckets are appended to the scene and placed in the occluder selection set. This allows nearby optimized geometry to hide geometry in farther buckets without processing the nearby geometry again.

One Simplygon perspective camera contains nine positions: the view location and the eight corners of a small box around it. All positions look along the view actor's forward vector.

camera_offsets = [unreal.Vector()]
for forward_side in (-1.0, 1.0):
    for right_side in (-1.0, 1.0):
        for up_side in (-1.0, 1.0):
            camera_offsets.append(
                (
                    job.view_forward * forward_side
                    + job.view_right * right_side
                    + job.view_up * up_side
                )
                * VISIBILITY_CAMERA_BOX_HALF_EXTENT
            )

The box gives visibility processing a small movement volume instead of optimizing for one exact point. The camera uses a 170-degree horizontal field of view, while VISIBILITY_CAMERA_RESOLUTION controls its pixel field of view.

Create aggregation and reduction pipelines

The nearest bucket uses aggregation instead of remeshing. It combines nearby source meshes while still using the common visibility settings.

pipeline = sg_utils.create_aggregation_pipeline(self.context)
process_settings = pipeline.GetAggregationSettings()

Farther buckets use reduction with maximum deviation as the target. Triangle ratio, triangle count, and on-screen-size targets are disabled so maximum deviation is the only reduction target.

pipeline = self.context.get_api_handle().CreateReductionPipeline()
reduction_settings = pipeline.GetReductionSettings()
reduction_settings.SetMergeGeometries(True)
reduction_settings.SetReductionTargetTriangleRatioEnabled(False)
reduction_settings.SetReductionTargetTriangleCountEnabled(False)
reduction_settings.SetReductionTargetOnScreenSizeEnabled(False)
reduction_settings.SetReductionTargetMaxDeviationEnabled(True)
reduction_settings.SetReductionTargetMaxDeviation(settings.max_deviation)

Both pipelines receive the same process, occluder, and camera selection sets. Conservative occlusion culling is enabled, and geometry not belonging to the process selection is discarded from the output.

process_settings.SetKeepUnprocessedSceneMeshes(False)
visibility_settings.SetConservativeMode(True)
visibility_settings.SetCullOccludedGeometry(True)

Process without blocking Unreal Editor callbacks

RunSceneFromFile is a blocking Simplygon operation. The optimizer runs it in a Python worker thread so Unreal's editor callbacks can continue handling export and import events.

self.pipeline_thread = threading.Thread(
    target=self._run_pipeline,
    args=(pipeline, job, visibility_scene_path, processed_path),
    name=f"Simplygon_{job.label}",
    daemon=True,
)
self.pipeline_thread.start()

The worker places its result in a thread-safe queue. A Slate post-tick callback polls that queue on the editor thread. This keeps Unreal API work on the editor side while the blocking Simplygon process runs separately.

try:
    job, processed_path, error_code, error_message = (
        self.pipeline_results.get_nowait()
    )
except queue.Empty:
    return

Import to Unreal Engine

When processing succeeds, the resulting scene is registered as an occluder for the next bucket and imported into the current editor world.

self.occluder_scene_paths.setdefault(view_path, []).append(processed_path)
subsystem.import_scene_as_actors(
    processed_path, world, unreal.SgSceneImportSettings()
)

Imported actors receive the SimplygonView tag. The optimizer identifies them from both the import result and a before-import actor snapshot, then verifies that the tag was applied. The source discovery step checks this tag, preventing generated actors from becoming input if the script is run again.

Visibility culling can remove every vertex in a bucket. In that case Simplygon completes successfully but returns no imported actor paths. This is a valid empty result rather than a tagging failure, so the optimizer logs a warning and continues with the next bucket. If Simplygon reports actor paths but none of those actors can be tagged, processing still stops with an error.

def _on_import_completed(self, result: unreal.ImportSceneResult) -> None:
    subsystem = self.context.get_simplygon_subsystem()
    subsystem.on_import_scene_completed.remove_callable(self._on_import_completed)
    job = self.import_job
    self.import_job = None
    if job is None:
        self._fail("Simplygon import completed without an active job.")
        return
    if result.has_error:
        self._fail(f"Import failed for {job.label}: {result.error_message}")
        return

    imported_actor_paths = list(result.imported_actor_paths)
    if not imported_actor_paths:
        self.pre_import_actor_paths.clear()
        unreal.log_warning(
            f"Completed view optimization bucket {job.label}, but Simplygon "
            "produced no importable actors. All geometry may have been removed "
            "by visibility culling."
        )
        self._start_next_export()
        return

    tagged_actor_count = self._tag_imported_actors(imported_actor_paths)
    if tagged_actor_count == 0:
        self._fail(
            f"Simplygon reported {len(imported_actor_paths)} imported actor(s) "
            f"for {job.label}, but could not apply the "
            f"'{GENERATED_ACTOR_TAG}' exclusion tag."
        )
        return

    unreal.log(
        f"Completed view optimization bucket: {job.label}. "
        f"Tagged {tagged_actor_count} generated actor(s)."
    )
    self._start_next_export()

After import completes, the optimizer starts the next populated bucket. When no jobs remain, dirty packages are saved.

Run the optimizer

Place the script and its helper modules where Unreal's Python environment can import them. For this project the script is located at:

Content/Python/simplygon_view_optimizer.py

Set VIEW_BLUEPRINT_CLASS_PATH if the generated class name is ambiguous, then run the following from the Unreal Editor Python console:

import simplygon_view_optimizer
simplygon_view_optimizer.optimize_views()

During development, reload the module before another run to pick up source changes:

import importlib
import simplygon_view_optimizer

importlib.reload(simplygon_view_optimizer)
simplygon_view_optimizer.optimize_views()

Result

Each populated bucket is processed for its view position. A bucket produces imported optimized actors when visible geometry remains; fully culled buckets complete without creating actors. Near geometry is aggregated, distant geometry is progressively reduced, and geometry outside the configured camera visibility is removed.

Let us look at the different views.

View 1 original
View 1 original
View 1 optimized
View 1 optimized
View 2 original
View 2 original
View 2 optimized
View 2 optimized
View 1 original
View 1 original
View 1 optimized
View 1 optimized

The strongest validation is to compare the source and generated views from the full camera movement volume, not only from the center position. Silhouettes and important openings should remain stable as the camera moves between the nine sampled positions.

Complete script

Script should be saved in Content/Python/simplygon_view_optimizer.py

"""Build Simplygon-optimized level views for every ViewPosition actor.

Run ``optimize_views()`` from the Unreal Editor Python console.
Distances are Unreal units (centimeters by default).
"""

import math
import queue
import threading
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Dict, List, Optional, Set

import unreal
from simplygon10 import Simplygon

import simplygon_utils as sg_utils
from simplygon_common import get_default_scene_export_settings


# Use the generated class path when multiple Blueprints have the same name, for
# example: "/Game/Blueprints/ViewPosition.ViewPosition_C".
VIEW_BLUEPRINT_CLASS_PATH = ""
VIEW_BLUEPRINT_CLASS_NAME = "ViewPosition"
EXCLUDED_ACTOR_CLASS_NAMES = {
    "BP_Sky_Sphere_C",
    "BP_SkySphere_C",
}

GENERATED_ACTOR_TAG = "SimplygonView"
VISIBILITY_CAMERA_SELECTION_SET = "ViewVisibilityCamera"
VISIBILITY_PROCESS_SELECTION_SET = "ViewVisibilityProcess"
VISIBILITY_OCCLUDER_SELECTION_SET = "ViewVisibilityOccluders"
VISIBILITY_HORIZONTAL_FOV_DEGREES = 170.0
VISIBILITY_CAMERA_RESOLUTION = 2048
VISIBILITY_CAMERA_BOX_HALF_EXTENT = 100.0


class OptimizationMethod(Enum):
    AGGREGATION = "aggregation"
    REDUCTION = "reduction"


@dataclass(frozen=True)
class BucketSettings:
    method: OptimizationMethod
    distance_threshold: float
    max_deviation: float = 10.0


# Distance thresholds are inclusive upper bounds in Unreal units (normally cm).
# Keep math.inf last so every mesh is assigned.
BUCKET_SETTINGS = [
    BucketSettings(OptimizationMethod.AGGREGATION, distance_threshold=2000.0),
    BucketSettings(
        OptimizationMethod.REDUCTION,
        distance_threshold=5000.0,
        max_deviation=20.0,
    ),
    BucketSettings(
        OptimizationMethod.REDUCTION,
        distance_threshold=15000.0,
        max_deviation=60.0,
    ),
    BucketSettings(
        OptimizationMethod.REDUCTION,
        distance_threshold=30000.0,
        max_deviation=100.0,
    ),
    BucketSettings(
        OptimizationMethod.REDUCTION,
        distance_threshold=math.inf,
        max_deviation=110.0,
    ),
]


@dataclass
class OptimizationJob:
    view_actor: unreal.Actor
    view_location: unreal.Vector
    view_forward: unreal.Vector
    view_right: unreal.Vector
    view_up: unreal.Vector
    bucket_index: int
    components: List[unreal.StaticMeshComponent]

    @property
    def label(self) -> str:
        return f"{self.view_actor.get_actor_label()}_Bucket_{self.bucket_index + 1}"


class ViewOptimizer:
    """Builds Simplygon-optimized views one bucket at a time."""

    def __init__(self) -> None:
        self.context = None
        self.jobs: List[OptimizationJob] = []
        self.export_jobs: Dict[str, OptimizationJob] = {}
        self.import_job: Optional[OptimizationJob] = None
        self.pre_import_actor_paths: Set[str] = set()
        self.source_components: List[unreal.StaticMeshComponent] = []
        self.occluder_scene_paths: Dict[str, List[str]] = {}
        self.pipeline_results = queue.Queue()
        self.pipeline_thread = None
        self.tick_handle = None
        self.failed = False

    def run(self) -> None:
        self._validate_settings()
        view_actors = self._find_view_actors()
        if not view_actors:
            raise RuntimeError(
                f"No Blueprint actors of class '{VIEW_BLUEPRINT_CLASS_NAME}' were found."
            )

        components = self._find_static_mesh_components(view_actors)
        if not components:
            raise RuntimeError("No StaticMeshComponents were found in the current level.")

        self.source_components = components
        self.jobs = self._create_jobs(view_actors, components)
        self._log_bucket_assignments()
        self.jobs = [job for job in self.jobs if job.components]
        if not self.jobs:
            raise RuntimeError("No populated distance buckets were found.")

        unreal.log(f"Starting {len(self.jobs)} Simplygon optimization jobs.")
        self.context = sg_utils.SimplygonUEContext()
        if not self.context.is_api_handle_valid():
            raise RuntimeError("Could not initialize the Simplygon API.")
        if not self.context.is_subsystem_valid():
            raise RuntimeError("Could not initialize the Simplygon subsystem.")
        self._start_next_export()

    def _validate_settings(self) -> None:
        if not BUCKET_SETTINGS:
            raise ValueError("At least one bucket must be configured.")
        if any(
            current.distance_threshold <= previous.distance_threshold
            for previous, current in zip(
                BUCKET_SETTINGS, BUCKET_SETTINGS[1:]
            )
        ):
            raise ValueError("Bucket distance thresholds must be strictly increasing.")
        if not math.isinf(BUCKET_SETTINGS[-1].distance_threshold):
            unreal.log_warning(
                "Meshes beyond the final bucket distance threshold will be skipped."
            )
        for bucket_index, settings in enumerate(BUCKET_SETTINGS):
            if (
                settings.method == OptimizationMethod.REDUCTION
                and settings.max_deviation <= 0.0
            ):
                raise ValueError(
                    f"Bucket {bucket_index} max_deviation must be greater than zero."
                )


    def _find_view_actors(self) -> List[unreal.Actor]:
        actors = unreal.get_editor_subsystem(
            unreal.EditorActorSubsystem
        ).get_all_level_actors()

        if VIEW_BLUEPRINT_CLASS_PATH:
            view_class = unreal.load_class(None, VIEW_BLUEPRINT_CLASS_PATH)
            if view_class is None:
                raise RuntimeError(
                    f"Could not load view class '{VIEW_BLUEPRINT_CLASS_PATH}'."
                )
            return [actor for actor in actors if isinstance(actor, view_class)]

        expected_names = {
            VIEW_BLUEPRINT_CLASS_NAME,
            f"{VIEW_BLUEPRINT_CLASS_NAME}_C",
            f"BP_{VIEW_BLUEPRINT_CLASS_NAME}_C",
        }
        return [
            actor
            for actor in actors
            if actor.get_class().get_name() in expected_names
        ]

    def _find_static_mesh_components(
        self, view_actors: List[unreal.Actor]
    ) -> List[unreal.StaticMeshComponent]:
        view_paths = {actor.get_path_name() for actor in view_actors}
        world = unreal.get_editor_subsystem(
            unreal.UnrealEditorSubsystem
        ).get_editor_world()
        if world is None:
            raise RuntimeError("Could not get the current editor world.")
        actors = unreal.GameplayStatics.get_all_actors_of_class(world, unreal.Actor)
        components = []
        for actor in actors:
            if (
                actor.get_path_name() in view_paths
                or actor.actor_has_tag(GENERATED_ACTOR_TAG)
                or actor.get_class().get_name() in EXCLUDED_ACTOR_CLASS_NAMES
            ):
                continue
            for component in actor.get_components_by_class(unreal.StaticMeshComponent):
                if (
                    component.get_editor_property("static_mesh") is not None
                    and component.get_editor_property("mobility")
                    == unreal.ComponentMobility.STATIC
                ):
                    components.append(component)
        return components

    def _create_jobs(
        self,
        view_actors: List[unreal.Actor],
        components: List[unreal.StaticMeshComponent],
    ) -> List[OptimizationJob]:
        jobs = []
        for view_actor in view_actors:
            view_location = view_actor.get_actor_location()
            view_forward = view_actor.get_actor_forward_vector()
            view_right = view_actor.get_actor_right_vector()
            view_up = view_actor.get_actor_up_vector()
            buckets = [[] for _ in BUCKET_SETTINGS]
            for component in components:
                distance = (component.get_world_location() - view_location).length()
                for bucket_index, settings in enumerate(BUCKET_SETTINGS):
                    if distance <= settings.distance_threshold:
                        buckets[bucket_index].append(component)
                        break

            for bucket_index, bucket_components in enumerate(buckets):
                jobs.append(
                    OptimizationJob(
                        view_actor=view_actor,
                        view_location=view_location,
                        view_forward=view_forward,
                        view_right=view_right,
                        view_up=view_up,
                        bucket_index=bucket_index,
                        components=bucket_components,
                    )
                )
        return jobs

    def _log_bucket_assignments(self) -> None:
        unreal.log("========== View Bucket Report ==========")
        for job in self.jobs:
            threshold = BUCKET_SETTINGS[job.bucket_index].distance_threshold
            threshold_text = "infinity" if math.isinf(threshold) else f"{threshold:.0f}"
            method = BUCKET_SETTINGS[job.bucket_index].method.value
            unreal.log(
                f"{job.view_actor.get_actor_label()} | Bucket "
                f"{job.bucket_index + 1} | <= {threshold_text} cm | {method} | "
                f"{len(job.components)} mesh components"
            )

            actors = {}
            for component in job.components:
                owner = component.get_owner()
                actor_path = owner.get_path_name()
                distance = (
                    component.get_world_location() - job.view_location
                ).length()
                actors.setdefault(actor_path, []).append((component, distance))

            for actor_path in sorted(actors):
                actor_components = actors[actor_path]
                owner = actor_components[0][0].get_owner()
                distances = ", ".join(
                    f"{component.get_name()}={distance:.0f} cm"
                    for component, distance in actor_components
                )
                unreal.log(
                    f"  {owner.get_actor_label()} ({actor_path}) | {distances}"
                )
        unreal.log("========== End View Bucket Report ==========")

    def _start_next_export(self) -> None:
        if not self.jobs:
            self._finish()
            return

        job = self.jobs.pop(0)
        output_path = self._scene_path("Source", job)
        subsystem = self.context.get_simplygon_subsystem()
        subsystem.on_export_scene_completed.add_callable(self._on_export_completed)
        job_id = subsystem.export_static_mesh_components_to_scene(
            output_path,
            job.components,
            get_default_scene_export_settings(False),
        )
        self.export_jobs[job_id.to_string()] = job
        unreal.log(f"Exporting view optimization bucket: {job.label}")

    def _on_export_completed(self, result: unreal.ExportSceneResult) -> None:
        subsystem = self.context.get_simplygon_subsystem()
        subsystem.on_export_scene_completed.remove_callable(self._on_export_completed)
        job = self.export_jobs.pop(result.job_id.to_string(), None)
        if job is None:
            self._fail("Simplygon export completed for an unknown job.")
            return
        if result.has_error:
            self._fail(f"Export failed for {job.label}: {result.error_message}")
            return

        self._process_and_import(job, result.output_file_path)

    def _process_and_import(
        self, job: OptimizationJob, source_scene_path: str
    ) -> None:
        if not Path(source_scene_path).exists():
            self._fail(f"Exported scene was not found: {source_scene_path}")
            return

        visibility_scene_path = self._scene_path("VisibilitySource", job)
        self._add_visibility_camera(job, source_scene_path, visibility_scene_path)
        pipeline = self._create_pipeline(job)
        processed_path = self._scene_path("Processed", job)
        unreal.log(
            f"Running {BUCKET_SETTINGS[job.bucket_index].method.value} with "
            f"{VISIBILITY_HORIZONTAL_FOV_DEGREES:.0f}-degree visibility culling "
            f"for {job.label}"
        )
        self.tick_handle = unreal.register_slate_post_tick_callback(
            self._poll_pipeline_result
        )
        self.pipeline_thread = threading.Thread(
            target=self._run_pipeline,
            args=(pipeline, job, visibility_scene_path, processed_path),
            name=f"Simplygon_{job.label}",
            daemon=True,
        )
        self.pipeline_thread.start()

    def _add_visibility_camera(
        self,
        job: OptimizationJob,
        source_scene_path: str,
        visibility_scene_path: str,
    ) -> None:
        api = self.context.get_api_handle()
        importer = api.CreateSceneImporter()
        importer.SetImportFilePath(source_scene_path)
        import_result = importer.Run()
        if Simplygon.Failed(import_result):
            raise RuntimeError(f"Could not load exported scene for {job.label}.")

        scene = importer.GetScene()
        root_node = scene.GetRootNode()
        process_selection_set_name = self._selection_set_name(
            VISIBILITY_PROCESS_SELECTION_SET, job
        )
        occluder_selection_set_name = self._selection_set_name(
            VISIBILITY_OCCLUDER_SELECTION_SET, job
        )
        camera_selection_set_name = self._selection_set_name(
            VISIBILITY_CAMERA_SELECTION_SET, job
        )
        process_selection_set = api.CreateSelectionSet()
        process_selection_set.SetName(process_selection_set_name)
        for child_index in range(root_node.GetChildCount()):
            self._add_node_tree_to_selection_set(
                root_node.GetChild(child_index), process_selection_set
            )

        occluder_selection_set = api.CreateSelectionSet()
        occluder_selection_set.SetName(occluder_selection_set_name)
        view_path = job.view_actor.get_path_name()
        occluder_scene_paths = self.occluder_scene_paths.get(view_path, [])
        for occluder_scene_path in occluder_scene_paths:
            occluder_importer = api.CreateSceneImporter()
            occluder_importer.SetImportFilePath(occluder_scene_path)
            occluder_import_result = occluder_importer.Run()
            if Simplygon.Failed(occluder_import_result):
                raise RuntimeError(
                    f"Could not load occluder scene: {occluder_scene_path}"
                )

            first_occluder_child = root_node.GetChildCount()
            scene.Append(occluder_importer.GetScene())
            for child_index in range(
                first_occluder_child, root_node.GetChildCount()
            ):
                self._add_node_tree_to_selection_set(
                    root_node.GetChild(child_index), occluder_selection_set
                )

        selection_set_table = scene.GetSelectionSetTable()
        selection_set_table.AddSelectionSet(process_selection_set)
        selection_set_table.AddSelectionSet(occluder_selection_set)

        selection_set = api.CreateSelectionSet()
        selection_set.SetName(camera_selection_set_name)
        camera = api.CreateSceneCamera()
        camera.SetName(camera_selection_set_name)
        camera.SetCameraType(Simplygon.ECameraType_Perspective)
        camera.SetUseNormalizedCoordinates(False)

        field_of_view = math.radians(VISIBILITY_HORIZONTAL_FOV_DEGREES)
        camera.SetFieldOfView(field_of_view)
        camera.SetPixelFieldOfView( field_of_view / VISIBILITY_CAMERA_RESOLUTION)

        # Create cameras for a box around the view location
        camera_offsets = [unreal.Vector()]
        for forward_side in (-1.0, 1.0):
            for right_side in (-1.0, 1.0):
                for up_side in (-1.0, 1.0):
                    camera_offsets.append(
                        (
                            job.view_forward * forward_side
                            + job.view_right * right_side
                            + job.view_up * up_side
                        )
                        * VISIBILITY_CAMERA_BOX_HALF_EXTENT
                    )

        for camera_offset in camera_offsets:
            camera_position = job.view_location + camera_offset
            target = camera_position + job.view_forward * 100.0
            camera.GetCameraPositions().AddTuple(
                [camera_position.x, camera_position.z, camera_position.y]
            )
            camera.GetTargetPositions().AddTuple([target.x, target.z, target.y])

        root_node.AddChild(camera)
        selection_set.AddItem(camera.GetNodeGUID())
        selection_set_table.AddSelectionSet(selection_set)
        if not camera.ValidateCamera():
            raise RuntimeError(f"Visibility camera is invalid for {job.label}.")

        exporter = api.CreateSceneExporter()
        exporter.SetExportFilePath(visibility_scene_path)
        exporter.SetScene(scene)
        export_result = exporter.Run()
        if Simplygon.Failed(export_result):
            raise RuntimeError(f"Could not save visibility scene for {job.label}.")

        unreal.log(
            f"Prepared {job.label} with {len(occluder_scene_paths)} closer "
            "bucket scene(s) as occluders."
        )

    def _add_node_tree_to_selection_set(self, node, selection_set) -> None:
        selection_set.AddItem(node.GetNodeGUID())
        for child_index in range(node.GetChildCount()):
            self._add_node_tree_to_selection_set(
                node.GetChild(child_index), selection_set
            )

    def _selection_set_name(self, base_name: str, job: OptimizationJob) -> str:
        return f"{base_name}_Bucket_{job.bucket_index + 1}"

    def _run_pipeline(
        self, pipeline, job: OptimizationJob, source_scene_path: str, processed_path: str
    ) -> None:
        try:
            error_code = pipeline.RunSceneFromFile(
                source_scene_path,
                processed_path,
                Simplygon.EPipelineRunMode_RunInNewProcess,
            )
            self.pipeline_results.put((job, processed_path, error_code, None))
        except Exception as error:
            self.pipeline_results.put((job, processed_path, None, str(error)))

    def _poll_pipeline_result(self, delta_time: float) -> None:
        try:
            job, processed_path, error_code, error_message = (
                self.pipeline_results.get_nowait()
            )
        except queue.Empty:
            return

        unreal.unregister_slate_post_tick_callback(self.tick_handle)
        self.tick_handle = None
        self.pipeline_thread = None

        if error_message is not None:
            self._fail(f"Simplygon processing crashed for {job.label}: {error_message}")
            return
        if Simplygon.Failed(error_code):
            self._log_simplygon_messages()
            self._fail(f"Simplygon processing failed for {job.label}.")
            return

        view_path = job.view_actor.get_path_name()
        self.occluder_scene_paths.setdefault(view_path, []).append(processed_path)
        unreal.log(f"Simplygon processing completed for {job.label}. Importing...")
        self._import_processed_scene(job, processed_path)

    def _import_processed_scene(
        self, job: OptimizationJob, processed_path: str
    ) -> None:
        world = unreal.get_editor_subsystem(
            unreal.UnrealEditorSubsystem
        ).get_editor_world()
        if world is None:
            self._fail("Could not get the current editor world.")
            return

        self.import_job = job
        self.pre_import_actor_paths = {
            actor.get_path_name()
            for actor in unreal.get_editor_subsystem(
                unreal.EditorActorSubsystem
            ).get_all_level_actors()
        }
        subsystem = self.context.get_simplygon_subsystem()
        subsystem.on_import_scene_completed.add_callable(self._on_import_completed)
        subsystem.import_scene_as_actors(
            processed_path, world, unreal.SgSceneImportSettings()
        )

    def _create_pipeline(self, job: OptimizationJob):
        """Create an optimization pipeline with camera-based visibility culling."""
        settings = BUCKET_SETTINGS[job.bucket_index]
        common = {
            "observe_progress": False,
        }

        if settings.method == OptimizationMethod.AGGREGATION:
            pipeline = sg_utils.create_aggregation_pipeline(self.context, **common)
            process_settings = pipeline.GetAggregationSettings()
        elif settings.method == OptimizationMethod.REDUCTION:
            pipeline = self.context.get_api_handle().CreateReductionPipeline()
            reduction_settings = pipeline.GetReductionSettings()
            process_settings = reduction_settings
            reduction_settings.SetMergeGeometries(True)
            reduction_settings.SetReductionTargetTriangleRatioEnabled(False)
            reduction_settings.SetReductionTargetTriangleCountEnabled(False)
            reduction_settings.SetReductionTargetOnScreenSizeEnabled(False)
            reduction_settings.SetReductionTargetMaxDeviationEnabled(True)
            reduction_settings.SetReductionTargetMaxDeviation(
                settings.max_deviation
            )
        else:
            raise ValueError(f"Unsupported optimization method: {settings.method}")

        process_settings.SetProcessSelectionSetName(
            self._selection_set_name(VISIBILITY_PROCESS_SELECTION_SET, job)
        )
        process_settings.SetKeepUnprocessedSceneMeshes(False)

        visibility_settings = pipeline.GetVisibilitySettings()
        visibility_settings.SetCameraSelectionSetName(
            self._selection_set_name(VISIBILITY_CAMERA_SELECTION_SET, job)
        )
        visibility_settings.SetOccluderSelectionSetName(
            self._selection_set_name(VISIBILITY_OCCLUDER_SELECTION_SET, job)
        )
        visibility_settings.SetConservativeMode(True)
        visibility_settings.SetCullOccludedGeometry(True)
        return pipeline

    def _on_import_completed(self, result: unreal.ImportSceneResult) -> None:
        subsystem = self.context.get_simplygon_subsystem()
        subsystem.on_import_scene_completed.remove_callable(self._on_import_completed)
        job = self.import_job
        self.import_job = None
        if job is None:
            self._fail("Simplygon import completed without an active job.")
            return
        if result.has_error:
            self._fail(f"Import failed for {job.label}: {result.error_message}")
            return

        imported_actor_paths = list(result.imported_actor_paths)
        if not imported_actor_paths:
            self.pre_import_actor_paths.clear()
            unreal.log_warning(
                f"Completed view optimization bucket {job.label}, but Simplygon "
                "produced no importable actors. All geometry may have been removed "
                "by visibility culling."
            )
            self._start_next_export()
            return

        tagged_actor_count = self._tag_imported_actors(imported_actor_paths)
        if tagged_actor_count == 0:
            self._fail(
                f"Simplygon reported {len(imported_actor_paths)} imported actor(s) "
                f"for {job.label}, but could not apply the "
                f"'{GENERATED_ACTOR_TAG}' exclusion tag."
            )
            return

        unreal.log(
            f"Completed view optimization bucket: {job.label}. "
            f"Tagged {tagged_actor_count} generated actor(s)."
        )
        self._start_next_export()

    def _tag_imported_actors(self, imported_actor_paths) -> int:
        reported_paths = {str(path) for path in imported_actor_paths}
        actors = unreal.get_editor_subsystem(
            unreal.EditorActorSubsystem
        ).get_all_level_actors()
        tagged_actor_count = 0
        for actor in actors:
            actor_path = actor.get_path_name()
            if (
                actor_path in self.pre_import_actor_paths
                and actor_path not in reported_paths
            ):
                continue
            tags = list(actor.get_editor_property("tags"))
            if GENERATED_ACTOR_TAG not in tags:
                tags.append(GENERATED_ACTOR_TAG)
                actor.set_editor_property("tags", tags)
            if actor.actor_has_tag(GENERATED_ACTOR_TAG):
                tagged_actor_count += 1

        self.pre_import_actor_paths.clear()
        return tagged_actor_count

    def _finish(self) -> None:
        if self.failed:
            return
        #for component in self.source_components:
        #    component.set_visibility(False, True)
        #    component.set_hidden_in_game(True)
        unreal.EditorLoadingAndSavingUtils.save_dirty_packages(True, True)
        unreal.log(
            "View optimization completed. Original StaticMeshComponents are hidden."
        )

    def _scene_path(self, prefix: str, job: OptimizationJob) -> str:
        safe_label = "".join(
            character if character.isalnum() or character in "_-" else "_"
            for character in job.label
        )
        return (
            unreal.Paths.project_intermediate_dir()
            + f"/Simplygon/Views/{prefix}_{safe_label}.sgscene"
        )

    def _log_simplygon_messages(self) -> None:
        api = self.context.get_api_handle()
        for getter, logger in (
            (api.GetWarningMessages, unreal.log_warning),
            (api.GetErrorMessages, unreal.log_error),
        ):
            messages = api.CreateStringArray()
            getter(messages)
            for index in range(messages.GetItemCount()):
                logger(messages.GetItem(index))
        api.ClearWarningMessages()
        api.ClearErrorMessages()

    def _fail(self, message: str) -> None:
        self.failed = True
        unreal.log_error(message)


_active_optimizer: Optional[ViewOptimizer] = None


def optimize_views() -> ViewOptimizer:
    """Build optimized assets for every populated view bucket."""
    global _active_optimizer
    _active_optimizer = ViewOptimizer()
    _active_optimizer.run()
    return _active_optimizer
All posts