> This is the Markdown version of the page. For the complete Lattice Developers documentation index, fetch https://developer.anduril.com/llms.txt. Append .md to any page URL for its clean Markdown.

# Create video streams

> Publish a camera asset and start an RTSP, MPEG-TS or SRT ingress stream.

Publish a camera asset entity, start an ingress stream over RTSP, MPEG-TS or SRT, and attach the stream to the entity
so operators can view the feed in the Lattice UI.

## Before you begin

* To publish a camera asset and start a stream, [set up](/guides/getting-started/set-up) your Lattice environment.
* See an overview of the [Video Manager API and stream lifecycle](/guides/video/overview) in Lattice.

> **gRPC authentication**
>
> These gRPC examples authenticate with a static environment token attached as request metadata.
> If you are using OAuth 2.0 client credentials instead, [set up the token refresh module](/guides/getting-started/authenticate#client-credentials) before running the examples on this page.

## Start a stream

Lattice supports the following video streaming protocols:

#### RTSP

If you don't already have an RTSP source, stand one up with [MediaMTX](https://github.com/bluenviron/mediamtx):

#### Install MediaMTX

Download a pre-built MediaMTX binary for your platform from the [MediaMTX releases page](https://github.com/bluenviron/mediamtx/releases)
and place it on the host that publishes the camera feed.

#### Configure the RTSP listener

In your `mediamtx.yml`, set the RTSP listener address and define a `/cam` or similar path that reads from the local camera.
Set `source` to match your camera; see the [MediaMTX paths documentation](https://github.com/bluenviron/mediamtx#configuration) for the available source types.

**`mediamtx.yml`**

```yaml title="mediamtx.yml"
rtspAddress: :<your-port>

paths:
  cam:
    source: <your-camera-source>
```

#### Start MediaMTX

Start the server in the foreground to verify connectivity, then move it to a systemd service for production use.

```bash
./mediamtx mediamtx.yml
```

You see `[RTSP] listener opened on :<your-port>` once the server is ready.

#### Verify the stream

From any host on the same network, confirm that the feed is playable. Replace `<rtsp-host>` with the address of the host running MediaMTX.

```bash
ffplay rtsp://<rtsp-host>:<your-port>/cam
```

A playback window confirms the source is ready for Lattice ingestion.

Before attaching a video to an asset, the asset must exist as a Lattice entity. Publish a camera entity with the
`TEMPLATE_ASSET` ontology and a `PlatformType` of `CAMERA`, then refresh it on a heartbeat so it does not expire:

#### Publish a camera

Construct an `Entity` with the components Lattice needs to render a friendly land-based camera asset on the common
operational picture (COP). `ExpiryTime` is set five minutes out and paired with a five-second publish loop, so the entity stays alive as long as the publisher runs.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to publish a camera asset entity.
// Attaching a video to an asset requires the asset to already exist as a Lattice entity, so
// publish the camera on a heartbeat before starting an ingress stream.
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"time"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
	"github.com/google/uuid"
)

const (
	cameraName          = "camera-01"
	integrationName     = "lattice_video_integration"
	dataType            = "camera_video_feed"
	publishInterval     = 5 * time.Second
	entityExpiryTtlSecs = 60
)

func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")

	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	// Stable entity ID derived from the camera's hardware identifier so the same
	// physical camera maps to the same Lattice entity across restarts.
	entityId := uuid.NewSHA1(uuid.NameSpaceURL, []byte(cameraName)).String()

	creationTime := time.Now().UTC()

	// Continuously publish the entity so it does not expire.
	for {
		latestTimestamp := time.Now().UTC()
		ctx := context.Background()

		entity := Lattice.Entity{
			EntityID:    &entityId,
			Description: Lattice.String("EO camera streaming live video"),
			Aliases: &Lattice.Aliases{
				Name: Lattice.String("Camera 01"),
				AlternateIDs: []*Lattice.AlternateID{
					{
						ID:   Lattice.String(cameraName),
						Type: Lattice.AlternateIDTypeAltIDTypeSerialNumber.Ptr(),
					},
				},
			},
			IsLive:      Lattice.Bool(true),
			CreatedTime: Lattice.Time(creationTime),
			// ExpiryTime is set in the future and paired with a shorter publish loop,
			// so the entity stays alive as long as the publisher runs.
			ExpiryTime: Lattice.Time(latestTimestamp.Add(entityExpiryTtlSecs * time.Second)),
			Ontology: &Lattice.Ontology{
				Template:     Lattice.OntologyTemplateTemplateAsset.Ptr(),
				PlatformType: Lattice.String("CAMERA"),
			},
			MilView: &Lattice.MilView{
				Disposition: Lattice.MilViewDispositionDispositionFriendly.Ptr(),
				Environment: Lattice.MilViewEnvironmentEnvironmentLand.Ptr(),
				Nationality: Lattice.MilViewNationalityNationalityUnitedStatesOfAmerica.Ptr(),
			},
			// Overall classification of the entity's data.
			DataClassification: &Lattice.Classification{
				Default: &Lattice.ClassificationInformation{
					Level: Lattice.ClassificationInformationLevelClassificationLevelsUnclassified.Ptr(),
				},
			},
			Location: &Lattice.Location{
				Position: &Lattice.Position{
					LatitudeDegrees:   Lattice.Float64(36.66633492530301),
					LongitudeDegrees:  Lattice.Float64(-116.9117674522639),
					AltitudeHaeMeters: Lattice.Float64(780.0),
				},
			},
			Provenance: &Lattice.Provenance{
				IntegrationName:  Lattice.String(integrationName),
				DataType:         Lattice.String(dataType),
				SourceUpdateTime: Lattice.Time(latestTimestamp),
			},
			Health: &Lattice.Health{
				ConnectionStatus: Lattice.HealthConnectionStatusConnectionStatusOnline.Ptr(),
				HealthStatus:     Lattice.HealthHealthStatusHealthStatusHealthy.Ptr(),
				ActiveAlerts:     []*Lattice.Alert{},
				UpdateTime:       Lattice.Time(latestTimestamp),
				// Per-subsystem health rolls up into the entity's overall
				// health_status; report at least the camera sensor itself.
				Components: []*Lattice.ComponentHealth{
					{
						ID:         Lattice.String("camera-01-eo"),
						Name:       Lattice.String("EO camera sensor"),
						Health:     Lattice.ComponentHealthHealthHealthStatusHealthy.Ptr(),
						UpdateTime: Lattice.Time(latestTimestamp),
					},
				},
			},
			// Declare the camera as a sensor so Lattice can render the sensor cone.
			Sensors: &Lattice.Sensors{
				Sensors: []*Lattice.Sensor{
					{
						SensorID:          Lattice.String("camera-01-eo"),
						SensorType:        Lattice.SensorSensorTypeSensorTypeCamera.Ptr(),
						OperationalState:  Lattice.SensorOperationalStateOperationalStateOperational.Ptr(),
						SensorDescription: Lattice.String("EO camera, H.264"),
						FieldsOfView: []*Lattice.FieldOfView{
							{
								CenterRayPose: &Lattice.EntityManagerPose{
									Orientation: &Lattice.Quaternion{
										X: Lattice.Float64(0.0),
										Y: Lattice.Float64(0.0),
										Z: Lattice.Float64(0.0),
										W: Lattice.Float64(1.0),
									},
								},
								HorizontalFov: Lattice.Float64(1.204), // ~69 deg (typical EO horizontal FoV)
								VerticalFov:   Lattice.Float64(0.954), // ~54.6 deg (typical EO vertical FoV)
								Range:         Lattice.Float64(150.0),
								Mode:          Lattice.FieldOfViewModeSensorModeSearch.Ptr(),
							},
						},
					},
				},
			},
			TaskCatalog: &Lattice.TaskCatalog{
				TaskDefinitions: []*Lattice.TaskDefinition{
					{TaskSpecificationURL: Lattice.String("type.googleapis.com/<org>.<package>.<version>.<task>")},
				},
			},
		}

		// Publish the entity
		_, err := LatticeClient.Entities.PublishEntity(ctx, &entity)
		if err != nil {
			fmt.Printf("Error publishing entity: %v\n", err)
		} else {
			fmt.Printf("Publishing asset with entity ID: %s\n", entityId)
		}

		time.Sleep(publishInterval)
	}
}

```

```java title={"Java (REST)"}
package org.example;

import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.List;
import java.util.UUID;

import com.anduril.Lattice;
import com.anduril.types.Alert;
import com.anduril.types.AlternateId;
import com.anduril.types.AlternateIdType;
import com.anduril.types.Aliases;
import com.anduril.types.Classification;
import com.anduril.types.ClassificationInformation;
import com.anduril.types.ClassificationInformationLevel;
import com.anduril.types.ComponentHealth;
import com.anduril.types.ComponentHealthHealth;
import com.anduril.types.Entity;
import com.anduril.types.EntityManagerPose;
import com.anduril.types.FieldOfView;
import com.anduril.types.FieldOfViewMode;
import com.anduril.types.Health;
import com.anduril.types.HealthConnectionStatus;
import com.anduril.types.HealthHealthStatus;
import com.anduril.types.Location;
import com.anduril.types.MilView;
import com.anduril.types.MilViewDisposition;
import com.anduril.types.MilViewEnvironment;
import com.anduril.types.MilViewNationality;
import com.anduril.types.Ontology;
import com.anduril.types.OntologyTemplate;
import com.anduril.types.Position;
import com.anduril.types.Provenance;
import com.anduril.types.Quaternion;
import com.anduril.types.Sensor;
import com.anduril.types.SensorOperationalState;
import com.anduril.types.SensorSensorType;
import com.anduril.types.Sensors;
import com.anduril.types.TaskCatalog;
import com.anduril.types.TaskDefinition;

/**
 * Publishes a camera asset entity with the Lattice SDK. Attaching a video to an asset requires the
 * asset to already exist as a Lattice entity, so publish the camera on a heartbeat before starting
 * an ingress stream.
 */
public class PublishCamera {
    private static final String CAMERA_NAME = "camera-01";
    private static final String INTEGRATION_NAME = "lattice_video_integration";
    private static final String DATA_TYPE = "camera_video_feed";
    private static final long PUBLISH_INTERVAL_MS = 5000;
    private static final long ENTITY_EXPIRY_TTL_SEC = 60;

    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        // Check required variables and ensure URL has https://
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Stable entity ID derived from the camera's hardware identifier so the same
            // physical camera maps to the same Lattice entity across restarts.
            String entityId = UUID.nameUUIDFromBytes(CAMERA_NAME.getBytes()).toString();
            Instant creationTime = Instant.now();

            // Continuously publish the camera so it does not expire while the publisher runs.
            while (true) {
                try {
                    Instant latestTimestamp = Instant.now();

                    // Construct an Entity with the components Lattice needs to render a friendly
                    // land-based camera asset on the common operational picture (COP).
                    Entity entity = Entity.builder()
                        .entityId(entityId)
                        .description("EO camera streaming live video")
                        .aliases(Aliases.builder()
                            .name("Camera 01")
                            .alternateIds(List.of(AlternateId.builder()
                                .id(CAMERA_NAME)
                                .type(AlternateIdType.ALT_ID_TYPE_SERIAL_NUMBER)
                                .build()))
                            .build())
                        .isLive(true)
                        .createdTime(OffsetDateTime.ofInstant(creationTime, ZoneOffset.UTC))
                        // ExpiryTime is set in the future and paired with a shorter publish loop,
                        // so the entity stays alive as long as the publisher runs.
                        .expiryTime(OffsetDateTime.ofInstant(
                            latestTimestamp.plus(ENTITY_EXPIRY_TTL_SEC, ChronoUnit.SECONDS), ZoneOffset.UTC))
                        .ontology(Ontology.builder()
                            .template(OntologyTemplate.TEMPLATE_ASSET)
                            .platformType("CAMERA")
                            .build())
                        .milView(MilView.builder()
                            .disposition(MilViewDisposition.DISPOSITION_FRIENDLY)
                            .environment(MilViewEnvironment.ENVIRONMENT_LAND)
                            .nationality(MilViewNationality.NATIONALITY_UNITED_STATES_OF_AMERICA)
                            .build())
                        // Overall classification of the entity's data.
                        .dataClassification(Classification.builder()
                            .default_(ClassificationInformation.builder()
                                .level(ClassificationInformationLevel.CLASSIFICATION_LEVELS_UNCLASSIFIED)
                                .build())
                            .build())
                        .location(Location.builder()
                            .position(Position.builder()
                                .latitudeDegrees(36.66633492530301)
                                .longitudeDegrees(-116.9117674522639)
                                .altitudeHaeMeters(780.0)
                                .build())
                            .build())
                        .provenance(Provenance.builder()
                            .integrationName(INTEGRATION_NAME)
                            .dataType(DATA_TYPE)
                            .sourceUpdateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                            .build())
                        .health(Health.builder()
                            .connectionStatus(HealthConnectionStatus.CONNECTION_STATUS_ONLINE)
                            .healthStatus(HealthHealthStatus.HEALTH_STATUS_HEALTHY)
                            .activeAlerts(Collections.<Alert>emptyList())
                            .updateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                            // Per-subsystem health rolls up into the entity's overall
                            // health_status; report at least the camera sensor itself.
                            .components(List.of(ComponentHealth.builder()
                                .id("camera-01-eo")
                                .name("EO camera sensor")
                                .health(ComponentHealthHealth.HEALTH_STATUS_HEALTHY)
                                .updateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                                .build()))
                            .build())
                        // Declare the camera as a sensor so Lattice can render the sensor cone.
                        .sensors(Sensors.builder()
                            .sensors(List.of(Sensor.builder()
                                .sensorId("camera-01-eo")
                                .sensorType(SensorSensorType.SENSOR_TYPE_CAMERA)
                                .operationalState(SensorOperationalState.OPERATIONAL_STATE_OPERATIONAL)
                                .sensorDescription("EO camera, H.264")
                                .fieldsOfView(List.of(FieldOfView.builder()
                                    .centerRayPose(EntityManagerPose.builder()
                                        .orientation(Quaternion.builder()
                                            .x(0.0)
                                            .y(0.0)
                                            .z(0.0)
                                            .w(1.0)
                                            .build())
                                        .build())
                                    .horizontalFov(1.204f) // ~69 deg (typical EO horizontal FoV)
                                    .verticalFov(0.954f) // ~54.6 deg (typical EO vertical FoV)
                                    .range(150.0f)
                                    .mode(FieldOfViewMode.SENSOR_MODE_SEARCH)
                                    .build()))
                                .build()))
                            .build())
                        .taskCatalog(TaskCatalog.builder()
                            .taskDefinitions(List.of(TaskDefinition.builder()
                                .taskSpecificationUrl("type.googleapis.com/<org>.<package>.<version>.<task>")
                                .build()))
                            .build())
                        .build();

                    // Publish the entity
                    client.entities().publishEntity(entity);
                    System.out.println("Publishing asset with entity ID: " + entityId);

                    Thread.sleep(PUBLISH_INTERVAL_MS);
                } catch (Exception e) {
                    System.err.println("Error publishing entity: " + e.getMessage());
                    Thread.sleep(PUBLISH_INTERVAL_MS);
                }
            }
        } catch (Exception e) {
            System.err.println("Failed to initialize: " + e.getMessage());
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the public Lattice REST SDK (anduril-lattice-sdk) to publish a
# camera asset entity. Attaching a video to an asset requires the asset to already exist as a
# Lattice entity, so publish the camera on a heartbeat before starting an ingress stream.

import os
import sys
import time
import uuid
from datetime import datetime, timedelta, timezone

from anduril import (
    Aliases,
    AlternateId,
    Classification,
    ClassificationInformation,
    ComponentHealth,
    FieldOfView,
    Health,
    Lattice,
    Location,
    MilView,
    Ontology,
    Position,
    Provenance,
    Quaternion,
    Sensor,
    Sensors,
    EntityManagerPose,
    TaskCatalog,
    TaskDefinition,
)

CAMERA_NAME = "camera-01"
INTEGRATION_NAME = "lattice_video_integration"
DATA_TYPE = "camera_video_feed"
PUBLISH_INTERVAL_SEC = 5
ENTITY_EXPIRY_TTL_SEC = 60


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")

    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")
    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables.", file=sys.stderr)
        sys.exit(1)

    client = Lattice(
        base_url=f"https://{lattice_endpoint}",
        token=lambda: str(environment_token),
        # Remove the following header if you are not developing on Sandboxes.
        headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"},
    )

    # Stable entity_id derived from the camera's hardware identifier so the
    # same physical camera maps to the same Lattice entity across restarts.
    entity_id = str(uuid.uuid5(uuid.NAMESPACE_URL, CAMERA_NAME))

    creation_time = datetime.now(timezone.utc)

    try:
        while True:
            latest_timestamp = datetime.now(timezone.utc)

            client.entities.publish_entity(
                entity_id=entity_id,
                description="EO camera streaming live video",
                aliases=Aliases(
                    name="Camera 01",
                    alternate_ids=[
                        AlternateId(id=CAMERA_NAME, type="ALT_ID_TYPE_SERIAL_NUMBER"),
                    ],
                ),
                is_live=True,
                created_time=creation_time,
                # ExpiryTime is set in the future and paired with a shorter publish loop,
                # so the entity stays alive as long as the publisher runs.
                expiry_time=latest_timestamp + timedelta(seconds=ENTITY_EXPIRY_TTL_SEC),
                ontology=Ontology(template="TEMPLATE_ASSET", platform_type="CAMERA"),
                mil_view=MilView(
                    disposition="DISPOSITION_FRIENDLY",
                    environment="ENVIRONMENT_LAND",
                    nationality="NATIONALITY_UNITED_STATES_OF_AMERICA",
                ),
                # Overall classification of the entity's data.
                data_classification=Classification(
                    default=ClassificationInformation(
                        level="CLASSIFICATION_LEVELS_UNCLASSIFIED",
                    ),
                ),
                location=Location(
                    position=Position(
                        latitude_degrees=36.66633492530301,
                        longitude_degrees=-116.9117674522639,
                        altitude_hae_meters=780.0,
                    ),
                ),
                provenance=Provenance(
                    integration_name=INTEGRATION_NAME,
                    data_type=DATA_TYPE,
                    source_update_time=latest_timestamp,
                ),
                health=Health(
                    connection_status="CONNECTION_STATUS_ONLINE",
                    health_status="HEALTH_STATUS_HEALTHY",
                    active_alerts=[],
                    update_time=latest_timestamp,
                    # Per-subsystem health rolls up into the entity's overall
                    # health_status; report at least the camera sensor itself.
                    components=[
                        ComponentHealth(
                            id="camera-01-eo",
                            name="EO camera sensor",
                            health="HEALTH_STATUS_HEALTHY",
                            update_time=latest_timestamp,
                        ),
                    ],
                ),
                # Declare the camera as a sensor so Lattice can render the sensor cone.
                sensors=Sensors(
                    sensors=[
                        Sensor(
                            sensor_id="camera-01-eo",
                            sensor_type="SENSOR_TYPE_CAMERA",
                            operational_state="OPERATIONAL_STATE_OPERATIONAL",
                            sensor_description="EO camera, H.264",
                            fields_of_view=[
                                FieldOfView(
                                    center_ray_pose=EntityManagerPose(
                                        orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0),
                                    ),
                                    horizontal_fov=1.204,  # ~69 deg (typical EO horizontal FoV)
                                    vertical_fov=0.954,  # ~54.6 deg (typical EO vertical FoV)
                                    range=150.0,
                                    mode="SENSOR_MODE_SEARCH",
                                ),
                            ],
                        ),
                    ],
                ),
                task_catalog=TaskCatalog(
                    task_definitions=[
                        TaskDefinition(task_specification_url="type.googleapis.com/<org>.<package>.<version>.<task>"),
                    ],
                ),
            )
            print(f"Publishing asset with entity ID: {entity_id}")
            time.sleep(PUBLISH_INTERVAL_SEC)

    except (KeyboardInterrupt, SystemExit):
        print(">>> Exiting...")
    except Exception as error:
        print(f"Error publishing entity: {error}", file=sys.stderr)


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
import { LatticeClient } from "@anduril-industries/lattice-sdk";
import { v5 as uuidv5 } from "uuid";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;

// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

const cameraName = "camera-01";
const integrationName = "lattice_video_integration";
const dataType = "camera_video_feed";
const publishIntervalMs = 5000;
const entityExpiryTtlSecs = 60;

// Stable entity ID derived from the camera's hardware identifier so the same
// physical camera maps to the same Lattice entity across restarts.
const entityId = uuidv5(cameraName, uuidv5.URL);
const creationTime = new Date().toISOString();

async function App() {
    const latestTimestamp = new Date();
    try {
        // Attaching a video to an asset requires the asset to already exist as a Lattice
        // entity, so publish the camera on a heartbeat before starting an ingress stream.
        await client.entities.publishEntity(
            {
                entityId: entityId,
                description: "EO camera streaming live video",
                aliases: {
                    name: "Camera 01",
                    alternateIds: [
                        { id: cameraName, type: "ALT_ID_TYPE_SERIAL_NUMBER" }
                    ]
                },
                isLive: true,
                createdTime: creationTime,
                // ExpiryTime is set in the future and paired with a shorter publish loop,
                // so the entity stays alive as long as the publisher runs.
                expiryTime: new Date(latestTimestamp.getTime() + entityExpiryTtlSecs * 1000).toISOString(),
                ontology: {
                    template: "TEMPLATE_ASSET",
                    platformType: "CAMERA"
                },
                milView: {
                    disposition: "DISPOSITION_FRIENDLY",
                    environment: "ENVIRONMENT_LAND",
                    nationality: "NATIONALITY_UNITED_STATES_OF_AMERICA"
                },
                // Overall classification of the entity's data.
                dataClassification: {
                    default: {
                        level: "CLASSIFICATION_LEVELS_UNCLASSIFIED"
                    }
                },
                location: {
                    position: {
                        latitudeDegrees: 36.66633492530301,
                        longitudeDegrees: -116.9117674522639,
                        altitudeHaeMeters: 780.0
                    }
                },
                provenance: {
                    integrationName: integrationName,
                    dataType: dataType,
                    sourceUpdateTime: latestTimestamp.toISOString()
                },
                health: {
                    connectionStatus: "CONNECTION_STATUS_ONLINE",
                    healthStatus: "HEALTH_STATUS_HEALTHY",
                    activeAlerts: [],
                    updateTime: latestTimestamp.toISOString(),
                    // Per-subsystem health rolls up into the entity's overall
                    // healthStatus; report at least the camera sensor itself.
                    components: [
                        {
                            id: "camera-01-eo",
                            name: "EO camera sensor",
                            health: "HEALTH_STATUS_HEALTHY",
                            updateTime: latestTimestamp.toISOString()
                        }
                    ]
                },
                // Declare the camera as a sensor so Lattice can render the sensor cone.
                sensors: {
                    sensors: [
                        {
                            sensorId: "camera-01-eo",
                            sensorType: "SENSOR_TYPE_CAMERA",
                            operationalState: "OPERATIONAL_STATE_OPERATIONAL",
                            sensorDescription: "EO camera, H.264",
                            fieldsOfView: [
                                {
                                    centerRayPose: {
                                        orientation: { x: 0.0, y: 0.0, z: 0.0, w: 1.0 }
                                    },
                                    horizontalFov: 1.204, // ~69 deg (typical EO horizontal FoV)
                                    verticalFov: 0.954, // ~54.6 deg (typical EO vertical FoV)
                                    range: 150.0,
                                    mode: "SENSOR_MODE_SEARCH"
                                }
                            ]
                        }
                    ]
                },
                taskCatalog: {
                    taskDefinitions: [
                        { taskSpecificationUrl: "type.googleapis.com/<org>.<package>.<version>.<task>" }
                    ]
                }
            }
        );
        console.log(`Publishing asset with entity ID: ${entityId}`);
    } catch (error) {
        console.log(`Encountered the following error while publishing entity: ${error}`);
    }
}

(async function runIndefinitely() {
    while (true) {
        await App();
        await new Promise(resolve => setTimeout(resolve, publishIntervalMs));
    }
})();

```

```go title={"Go (gRPC)"}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/types/known/timestamppb"
	"google.golang.org/protobuf/types/known/wrapperspb"
	ontologyv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/ontology/v1"
	tasksv2 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/tasks/v2"
	_type "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/type"
)

const (
	cameraName      = "camera-01"
	integrationName = "lattice_video_integration"
	dataType        = "camera_video_feed"
	publishInterval = 5 * time.Second
	entityExpiryTTL = 60 * time.Second
	publishTimeout  = 10 * time.Second
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}

	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}
	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	client := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	// Stable entity_id derived from the camera's hardware identifier so the
	// same physical camera maps to the same Lattice entity across restarts.
	entityId := uuid.NewSHA1(uuid.NameSpaceURL, []byte(cameraName)).String()

	creationTimestamp := timestamppb.New(time.Now().UTC())

	for {
		latestTimestamp := time.Now().UTC()
		ctx, cancel := context.WithTimeout(context.Background(), publishTimeout)

		entity := &entitymanagerv1.Entity{
			EntityId:    entityId,
			Description: "EO camera streaming live video",
			Aliases: &entitymanagerv1.Aliases{
				Name: "Camera 01",
				AlternateIds: []*entitymanagerv1.AlternateId{
					{
						Id:   cameraName,
						Type: entitymanagerv1.AltIdType_ALT_ID_TYPE_SERIAL_NUMBER,
					},
				},
			},
			IsLive:      true,
			CreatedTime: creationTimestamp,
			ExpiryTime:  timestamppb.New(latestTimestamp.Add(entityExpiryTTL)),
			Ontology: &entitymanagerv1.Ontology{
				Template:     entitymanagerv1.Template_TEMPLATE_ASSET,
				PlatformType: "CAMERA",
			},
			MilView: &entitymanagerv1.MilView{
				Disposition: ontologyv1.Disposition_DISPOSITION_FRIENDLY,
				Environment: ontologyv1.Environment_ENVIRONMENT_LAND,
				Nationality: ontologyv1.Nationality_NATIONALITY_UNITED_STATES_OF_AMERICA,
			},
			// Overall classification of the entity's data.
			DataClassification: &entitymanagerv1.Classification{
				Default: &entitymanagerv1.ClassificationInformation{
					Level: entitymanagerv1.ClassificationLevels_CLASSIFICATION_LEVELS_UNCLASSIFIED,
				},
			},
			Location: &entitymanagerv1.Location{
				Position: &entitymanagerv1.Position{
					LatitudeDegrees:   36.66633492530301,
					LongitudeDegrees:  -116.9117674522639,
					AltitudeHaeMeters: wrapperspb.Double(780.0),
				},
			},
			Provenance: &entitymanagerv1.Provenance{
				IntegrationName:  integrationName,
				DataType:         dataType,
				SourceUpdateTime: timestamppb.New(latestTimestamp),
			},
			Health: &entitymanagerv1.Health{
				ConnectionStatus: entitymanagerv1.ConnectionStatus_CONNECTION_STATUS_ONLINE,
				HealthStatus:     entitymanagerv1.HealthStatus_HEALTH_STATUS_HEALTHY,
				ActiveAlerts:     []*entitymanagerv1.Alert{},
				UpdateTime:       timestamppb.New(latestTimestamp),
				// Per-subsystem health rolls up into the entity's overall
				// health_status; report at least the camera sensor itself.
				Components: []*entitymanagerv1.ComponentHealth{
					{
						Id:         "camera-01-eo",
						Name:       "EO camera sensor",
						Health:     entitymanagerv1.HealthStatus_HEALTH_STATUS_HEALTHY,
						UpdateTime: timestamppb.New(latestTimestamp),
					},
				},
			},
			Sensors: &entitymanagerv1.Sensors{
				Sensors: []*entitymanagerv1.Sensor{
					{
						SensorId:          "camera-01-eo",
						SensorType:        entitymanagerv1.SensorType_SENSOR_TYPE_CAMERA,
						OperationalState:  entitymanagerv1.OperationalState_OPERATIONAL_STATE_OPERATIONAL,
						SensorDescription: "EO camera, H.264",
						FieldsOfView: []*entitymanagerv1.FieldOfView{
							{
								CenterRayPose: &entitymanagerv1.Pose{
									Orientation: &_type.Quaternion{
										X: 0.0,
										Y: 0.0,
										Z: 0.0,
										W: 1.0,
									},
								},
								HorizontalFov: 1.204, // ~69 deg (typical EO horizontal FoV)
								VerticalFov:   0.954, // ~54.6 deg (typical EO vertical FoV)
								Range:         wrapperspb.Float(150.0),
								Mode:          entitymanagerv1.SensorMode_SENSOR_MODE_SEARCH,
							},
						},
					},
				},
			},
			TaskCatalog: &tasksv2.TaskCatalog{
				TaskDefinitions: []*tasksv2.TaskDefinition{
					{TaskSpecificationUrl: "type.googleapis.com/<org>.<package>.<version>.<task>"},
				},
			},
		}

		_, err := client.PublishEntity(ctx, &entitymanagerv1.PublishEntityRequest{Entity: entity})
		cancel()

		if err != nil {
			fmt.Printf("Error publishing entity: %v\n", err)
		} else {
			fmt.Printf("Publishing asset with entity ID: %s\n", entityId)
		}

		time.Sleep(publishInterval)
	}
}

```

```py title={"Python (gRPC)"}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
import time
import uuid
from datetime import datetime, timedelta, timezone
from bearer_auth import create_client
from protobuf.wkt import DoubleValue, FloatValue, Timestamp

from anduril.entitymanager.v1.classification_pub_pb import (
    Classification,
    ClassificationInformation,
    ClassificationLevels,
)
from anduril.entitymanager.v1.entity_pub_pb import (
    Aliases,
    AlternateId,
    Entity,
    Provenance,
)
from anduril.entitymanager.v1.entity_manager_api_pub_pb import PublishEntityRequest
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync
from anduril.entitymanager.v1.health_status_pub_pb import (
    ComponentHealth,
    ConnectionStatus,
    Health,
    HealthStatus,
)
from anduril.entitymanager.v1.location_pub_pb import Location, Pose, Position
from anduril.entitymanager.v1.ontology_pub_pb import MilView, Ontology
from anduril.entitymanager.v1.sensors_pub_pb import (
    FieldOfView,
    OperationalState,
    Sensor,
    SensorMode,
    Sensors,
    SensorType,
)
from anduril.entitymanager.v1.types_pub_pb import AltIdType, Template
from anduril.ontology.v1.type_pub_pb import Disposition, Environment, Nationality
from anduril.tasks.v2.catalog_pub_pb import TaskCatalog, TaskDefinition
from anduril.type.coords_pub_pb import Quaternion


CAMERA_NAME = "camera-01"
INTEGRATION_NAME = "lattice_video_integration"
DATA_TYPE = "camera_video_feed"
PUBLISH_INTERVAL_SEC = 5
ENTITY_EXPIRY_TTL_SEC = 60
PUBLISH_TIMEOUT_SEC = 10


def to_timestamp(dt: datetime) -> Timestamp:
    return Timestamp.from_datetime(dt)


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(EntityManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Stable entity_id derived from the camera's hardware identifier so the
    # same physical camera maps to the same Lattice entity across restarts.
    entity_id = str(uuid.uuid5(uuid.NAMESPACE_URL, CAMERA_NAME))

    creation_timestamp = to_timestamp(datetime.now(timezone.utc))

    while True:
        latest = datetime.now(timezone.utc)
        latest_ts = to_timestamp(latest)
        expiry_ts = to_timestamp(latest + timedelta(seconds=ENTITY_EXPIRY_TTL_SEC))

        entity = Entity(
            entity_id=entity_id,
            description="EO camera streaming live video",
            aliases=Aliases(
                name="Camera 01",
                alternate_ids=[
                    AlternateId(id=CAMERA_NAME, type=AltIdType.SERIAL_NUMBER),
                ],
            ),
            is_live=True,
            created_time=creation_timestamp,
            expiry_time=expiry_ts,
            ontology=Ontology(template=Template.ASSET, platform_type="CAMERA"),
            mil_view=MilView(
                disposition=Disposition.FRIENDLY,
                environment=Environment.LAND,
                nationality=Nationality.UNITED_STATES_OF_AMERICA,
            ),
            # Overall classification of the entity's data.
            data_classification=Classification(
                default=ClassificationInformation(
                    level=ClassificationLevels.UNCLASSIFIED,
                ),
            ),
            location=Location(
                position=Position(
                    latitude_degrees=36.66633492530301,
                    longitude_degrees=-116.9117674522639,
                    altitude_hae_meters=DoubleValue(value=780.0),
                ),
            ),
            provenance=Provenance(
                integration_name=INTEGRATION_NAME,
                data_type=DATA_TYPE,
                source_update_time=latest_ts,
            ),
            health=Health(
                connection_status=ConnectionStatus.ONLINE,
                health_status=HealthStatus.HEALTHY,
                active_alerts=[],
                update_time=latest_ts,
                # Per-subsystem health rolls up into the entity's overall
                # health_status; report at least the camera sensor itself.
                components=[
                    ComponentHealth(
                        id="camera-01-eo",
                        name="EO camera sensor",
                        health=HealthStatus.HEALTHY,
                        update_time=latest_ts,
                    ),
                ],
            ),
            sensors=Sensors(
                sensors=[
                    Sensor(
                        sensor_id="camera-01-eo",
                        sensor_type=SensorType.CAMERA,
                        operational_state=OperationalState.OPERATIONAL,
                        sensor_description="EO camera, H.264",
                        fields_of_view=[
                            FieldOfView(
                                center_ray_pose=Pose(
                                    orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0),
                                ),
                                horizontal_fov=1.204,  # ~69 deg (typical EO horizontal FoV)
                                vertical_fov=0.954,  # ~54.6 deg (typical EO vertical FoV)
                                range=FloatValue(value=150.0),
                                mode=SensorMode.SEARCH,
                            ),
                        ],
                    ),
                ],
            ),
            task_catalog=TaskCatalog(
                task_definitions=[
                    TaskDefinition(task_specification_url="type.googleapis.com/<org>.<package>.<version>.<task>"),
                ],
            ),
        )

        try:
            client.publish_entity(
                PublishEntityRequest(entity=entity),
                timeout_ms=PUBLISH_TIMEOUT_SEC * 1000,
            )
            print(f"Publishing asset with entity ID: {entity_id}")
        except Exception as error:
            print(f"Error publishing entity: {error}", file=sys.stderr)

        time.sleep(PUBLISH_INTERVAL_SEC)


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::{
    Aliases, AltIdType, AlternateId, Classification, ClassificationInformation,
    ClassificationLevels, ComponentHealth, ConnectionStatus, Entity, FieldOfView, Health,
    HealthStatus, Location, MilView, Ontology, OperationalState, Pose, Position,
    Provenance, PublishEntityRequest, Sensor, SensorMode, SensorType, Sensors, Template,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::ontology::v1::{
    Disposition, Environment, Nationality,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::r#type::Quaternion;
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::tasks::v2::{
    TaskCatalog, TaskDefinition,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use prost_types::Timestamp;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;
use uuid::Uuid;

use std::env;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

const CAMERA_NAME: &str = "camera-01";
const INTEGRATION_NAME: &str = "lattice_video_integration";
const DATA_TYPE: &str = "camera_video_feed";
const PUBLISH_INTERVAL_SECS: u64 = 5;
const ENTITY_EXPIRY_TTL_SECS: i64 = 60;

fn now_timestamp() -> Timestamp {
    let d = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
    Timestamp {
        seconds: d.as_secs() as i64,
        nanos: d.subsec_nanos() as i32,
    }
}

fn add_seconds(ts: &Timestamp, seconds: i64) -> Timestamp {
    Timestamp {
        seconds: ts.seconds + seconds,
        nanos: ts.nanos,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Stable entity_id derived from the camera's hardware identifier so the
    // same physical camera maps to the same Lattice entity across restarts.
    let entity_id = Uuid::new_v5(&Uuid::NAMESPACE_URL, CAMERA_NAME.as_bytes()).to_string();
    let creation_timestamp = now_timestamp();

    // Refresh the entity on a heartbeat so it does not expire while the publisher runs.
    loop {
        let latest_timestamp = now_timestamp();

        let entity = Entity {
            entity_id: entity_id.clone(),
            description: "EO camera streaming live video".to_string(),
            aliases: Some(Aliases {
                name: "Camera 01".to_string(),
                alternate_ids: vec![AlternateId {
                    id: CAMERA_NAME.to_string(),
                    r#type: AltIdType::SerialNumber as i32,
                }],
            }),
            is_live: true,
            created_time: Some(creation_timestamp.clone()),
            // ExpiryTime is set in the future and paired with a shorter publish loop,
            // so the entity stays alive as long as the publisher runs.
            expiry_time: Some(add_seconds(&latest_timestamp, ENTITY_EXPIRY_TTL_SECS)),
            ontology: Some(Ontology {
                template: Template::Asset as i32,
                platform_type: "CAMERA".to_string(),
                ..Default::default()
            }),
            mil_view: Some(MilView {
                disposition: Disposition::Friendly as i32,
                environment: Environment::Land as i32,
                nationality: Nationality::UnitedStatesOfAmerica as i32,
            }),
            // Overall classification of the entity's data.
            data_classification: Some(Classification {
                default: Some(ClassificationInformation {
                    level: ClassificationLevels::Unclassified as i32,
                    ..Default::default()
                }),
                ..Default::default()
            }),
            location: Some(Location {
                position: Some(Position {
                    latitude_degrees: 36.66633492530301,
                    longitude_degrees: -116.9117674522639,
                    altitude_hae_meters: Some(780.0),
                    ..Default::default()
                }),
                ..Default::default()
            }),
            provenance: Some(Provenance {
                integration_name: INTEGRATION_NAME.to_string(),
                data_type: DATA_TYPE.to_string(),
                source_update_time: Some(latest_timestamp.clone()),
                ..Default::default()
            }),
            health: Some(Health {
                connection_status: ConnectionStatus::Online as i32,
                health_status: HealthStatus::Healthy as i32,
                active_alerts: vec![],
                update_time: Some(latest_timestamp.clone()),
                // Per-subsystem health rolls up into the entity's overall
                // health_status; report at least the camera sensor itself.
                components: vec![ComponentHealth {
                    id: "camera-01-eo".to_string(),
                    name: "EO camera sensor".to_string(),
                    health: HealthStatus::Healthy as i32,
                    update_time: Some(latest_timestamp.clone()),
                    ..Default::default()
                }],
            }),
            // Declare the camera as a sensor so Lattice can render the sensor cone.
            sensors: Some(Sensors {
                sensors: vec![Sensor {
                    sensor_id: "camera-01-eo".to_string(),
                    sensor_type: SensorType::Camera as i32,
                    operational_state: OperationalState::Operational as i32,
                    sensor_description: "EO camera, H.264".to_string(),
                    fields_of_view: vec![FieldOfView {
                        center_ray_pose: Some(Pose {
                            orientation: Some(Quaternion {
                                x: 0.0,
                                y: 0.0,
                                z: 0.0,
                                w: 1.0,
                            }),
                            ..Default::default()
                        }),
                        horizontal_fov: 1.204, // ~69 deg (typical EO horizontal FoV)
                        vertical_fov: 0.954,   // ~54.6 deg (typical EO vertical FoV)
                        range: Some(150.0),
                        mode: SensorMode::Search as i32,
                        ..Default::default()
                    }],
                    ..Default::default()
                }],
            }),
            task_catalog: Some(TaskCatalog {
                task_definitions: vec![TaskDefinition {
                    task_specification_url: "type.googleapis.com/<org>.<package>.<version>.<task>"
                        .to_string(),
                }],
            }),
            ..Default::default()
        };

        let request = Request::new(PublishEntityRequest {
            entity: Some(entity),
        });
        match client.publish_entity(request).await {
            Ok(_) => println!("Publishing asset with entity ID: {}", entity_id),
            Err(e) => eprintln!("Error publishing entity: {}", e),
        }

        tokio::time::sleep(Duration::from_secs(PUBLISH_INTERVAL_SECS)).await;
    }
}

```

> **Info**
>
> The heartbeat loop is the simplest pattern for keeping a transient asset alive. For long-running deployments,
> consider publishing from a supervised service that resets `ExpiryTime` based on hardware health rather than a fixed interval.

#### Set the stream endpoint

The following generates a client-side UUID for `ingress_id`, calls `CreateIngressStream`
with your RTSP stream URL, then attaches the new video to the camera entity using `OverrideEntity`:

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to start an RTSP ingress stream and
// attach it to a camera asset entity.
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"time"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
	"github.com/google/uuid"
)

const (
	// Entity ID to associate the stream with. Replace before running.
	entityID = "<ENTITY-ID>"
	// RTSP source Lattice will pull from.
	rtspURL = "rtsp://<rtsp-host>:<port>/cam"
)

func main() {
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")

	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	ctx := context.Background()

	// Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
	ingressID := uuid.New().String()
	fmt.Printf("Starting RTSP ingress stream %s from %s\n", ingressID, rtspURL)

	// RTSP is a pull protocol: Lattice dials the supplied URL, so no push endpoint
	// is returned.
	createResp, err := LatticeClient.Video.CreateIngressStream(ctx, &Lattice.CreateIngressStreamRequest{
		IngressID: &ingressID,
		Title:     Lattice.String("cam-01"),
		Rtsp: &Lattice.RtspSettings{
			URL: Lattice.String(rtspURL),
		},
	})
	if err != nil {
		fmt.Printf("Failed to create ingress stream: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Successfully started stream with ID: %s\n", *createResp.GetIngressID())

	// Attach the new video to the asset entity. A field_path of media.media replaces
	// only the media list, leaving every other component on the entity untouched.
	_, err = LatticeClient.Entities.OverrideEntity(ctx, &Lattice.EntityOverride{
		EntityID:  entityID,
		FieldPath: "media.media",
		Entity: &Lattice.Entity{
			EntityID: Lattice.String(entityID),
			Media: &Lattice.Media{
				Media: []*Lattice.MediaItem{
					{
						ItemIdentifier: &ingressID,
						Type:           Lattice.MediaItemTypeMediaTypeVideo.Ptr(),
					},
				},
			},
		},
		Provenance: &Lattice.Provenance{
			IntegrationName:  Lattice.String("lattice-video-integration"),
			DataType:         Lattice.String("video-association"),
			SourceUpdateTime: Lattice.Time(time.Now().UTC()),
		},
	})
	if err != nil {
		fmt.Printf("Failed to override entity media: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Associated video %s with entity %s\n", ingressID, entityID)
}

```

```java title={"Java (REST)"}
package org.example;

import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.Collections;
import java.util.UUID;

import com.anduril.Lattice;
import com.anduril.resources.entities.requests.EntityOverride;
import com.anduril.resources.video.requests.CreateIngressStreamRequest;
import com.anduril.types.CreateIngressStreamResponse;
import com.anduril.types.Entity;
import com.anduril.types.Media;
import com.anduril.types.MediaItem;
import com.anduril.types.MediaItemType;
import com.anduril.types.Provenance;
import com.anduril.types.RtspSettings;

/**
 * Starts an RTSP ingress stream with the Lattice SDK and attaches it to a camera asset entity.
 */
public class CreateIngressStreamRtsp {
    // Entity ID to associate the stream with. Replace before running.
    private static final String ENTITY_ID = "<ENTITY-ID>";
    // RTSP source Lattice will pull from.
    private static final String RTSP_URL = "rtsp://<rtsp-host>:<port>/cam";

    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        // Check required variables and ensure the endpoint has its scheme
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
            String ingressId = UUID.randomUUID().toString();
            System.out.println("Starting RTSP ingress stream " + ingressId + " from " + RTSP_URL);

            // RTSP is a pull protocol: Lattice dials the supplied URL, so no push endpoint
            // is returned.
            CreateIngressStreamResponse response = client.video().createIngressStream(
                CreateIngressStreamRequest.builder()
                    .ingressId(ingressId)
                    .title("cam-01")
                    .rtsp(RtspSettings.builder()
                        .url(RTSP_URL)
                        .build())
                    .build());
            System.out.println("Successfully started stream with ID: " + response.getIngressId().orElse(ingressId));

            // Attach the new video to the asset entity. A field_path of media.media replaces
            // only the media list, leaving every other component on the entity untouched.
            MediaItem mediaItem = MediaItem.builder()
                .itemIdentifier(ingressId)
                .type(MediaItemType.MEDIA_TYPE_VIDEO)
                .build();
            EntityOverride entityOverride = EntityOverride.builder()
                .entity(Entity.builder()
                    .entityId(ENTITY_ID)
                    .media(Media.builder()
                        .media(Collections.singletonList(mediaItem))
                        .build())
                    .build())
                .provenance(Provenance.builder()
                    .integrationName("lattice-video-integration")
                    .dataType("video-association")
                    .sourceUpdateTime(OffsetDateTime.now(ZoneOffset.UTC))
                    .build())
                .build();
            client.entities().overrideEntity(ENTITY_ID, "media.media", entityOverride);

            System.out.println("Associated video " + ingressId + " with entity " + ENTITY_ID);
        } catch (Exception e) {
            System.err.println("Exception: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to start an RTSP ingress
# stream and attach it to a camera asset entity.

import os
import sys
import uuid
from datetime import datetime, timezone

from anduril import Entity, Lattice, Media, MediaItem, Provenance, RtspSettings

# Entity ID to associate the stream with. Replace before running.
ENTITY_ID = "<ENTITY-ID>"
# RTSP source Lattice will pull from.
RTSP_URL = "rtsp://<rtsp-host>:<port>/cam"


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")

    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")
    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables.", file=sys.stderr)
        sys.exit(1)

    client = Lattice(
        base_url=f"https://{lattice_endpoint}",
        token=lambda: str(environment_token),
        # Remove the following header if you are not developing on Sandboxes.
        headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"},
    )

    # Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    ingress_id = str(uuid.uuid4())
    print(f"Starting RTSP ingress stream {ingress_id} from {RTSP_URL}")

    # RTSP is a pull protocol: Lattice dials the supplied URL, so no push endpoint
    # is returned.
    try:
        create_resp = client.video.create_ingress_stream(
            ingress_id=ingress_id,
            title="cam-01",
            rtsp=RtspSettings(url=RTSP_URL),
        )
    except Exception as error:
        print(f"Failed to create ingress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully started stream with ID: {create_resp.ingress_id}")

    # Attach the new video to the asset entity. A field_path of media.media replaces
    # only the media list, leaving every other component on the entity untouched.
    try:
        client.entities.override_entity(
            entity_id=ENTITY_ID,
            field_path="media.media",
            entity=Entity(
                entity_id=ENTITY_ID,
                media=Media(
                    media=[
                        MediaItem(
                            item_identifier=ingress_id,
                            type="MEDIA_TYPE_VIDEO",
                        ),
                    ],
                ),
            ),
            provenance=Provenance(
                integration_name="lattice-video-integration",
                data_type="video-association",
                source_update_time=datetime.now(timezone.utc),
            ),
        )
    except Exception as error:
        print(f"Failed to override entity media: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Associated video {ingress_id} with entity {ENTITY_ID}")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
import { LatticeClient } from "@anduril-industries/lattice-sdk";
import { v4 as uuidv4 } from "uuid";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;

// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Entity ID to associate the stream with. Replace before running.
const entityId = "<ENTITY-ID>";
// RTSP source Lattice will pull from.
const rtspUrl = "rtsp://<rtsp-host>:<port>/cam";

async function App() {
    // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    const ingressId = uuidv4();
    console.log(`Starting RTSP ingress stream ${ingressId} from ${rtspUrl}`);

    try {
        // RTSP is a pull protocol: Lattice dials the supplied URL, so no push endpoint
        // is returned.
        const createResp = await client.video.createIngressStream({
            ingressId: ingressId,
            title: "cam-01",
            rtsp: { url: rtspUrl }
        });
        console.log(`Successfully started stream with ID: ${createResp.ingressId}`);

        // Attach the new video to the asset entity. A field_path of media.media replaces
        // only the media list, leaving every other component on the entity untouched.
        await client.entities.overrideEntity({
            entityId,
            fieldPath: "media.media",
            entity: {
                entityId: entityId,
                media: {
                    media: [
                        { itemIdentifier: ingressId, type: "MEDIA_TYPE_VIDEO" }
                    ]
                }
            },
            provenance: {
                integrationName: "lattice-video-integration",
                dataType: "video-association",
                sourceUpdateTime: new Date().toISOString()
            }
        });
        console.log(`Associated video ${ingressId} with entity ${entityId}`);
    } catch (error) {
        console.log(`Encountered the following error while starting the RTSP ingress stream: ${error}`);
    }
}

App();

```

```go title={"Go (gRPC)"}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const (
	// Entity ID to associate the stream with. Replace before running.
	entityID = "<ENTITY-ID>"
	// RTSP source Lattice will pull from.
	rtspURL = "rtsp://<rtsp-host>:<port>/cam"
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}

	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}
	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	videoClient := videomanagerv1grpc.NewVideoManagerAPIClient(conn)
	emClient := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
	ingressID := uuid.NewString()
	fmt.Printf("Starting RTSP ingress stream %s from %s\n", ingressID, rtspURL)

	createResp, err := videoClient.CreateIngressStream(ctx, &videomanagerv1.CreateIngressStreamRequest{
		IngressId: &ingressID,
		Title:     "cam-01",
		Ingress: &videomanagerv1.CreateIngressStreamRequest_Rtsp{
			Rtsp: &videomanagerv1.RtspSettings{
				Url: proto.String(rtspURL),
			},
		},
	})
	if err != nil {
		log.Fatalf("Failed to create ingress stream: %v", err)
	}
	fmt.Printf("Successfully started stream with ID: %s\n", createResp.GetIngressId())

	// Attach the new video to the asset entity. A field_path of media.media replaces
	// only the media list, leaving every other component on the entity untouched.
	_, err = emClient.OverrideEntity(ctx, &entitymanagerv1.OverrideEntityRequest{
		Entity: &entitymanagerv1.Entity{
			EntityId: entityID,
			Media: &entitymanagerv1.Media{
				Media: []*entitymanagerv1.MediaItem{{
					ItemIdentifier: ingressID,
					Type:           entitymanagerv1.MediaType_MEDIA_TYPE_VIDEO,
				}},
			},
		},
		FieldPath: []string{"media.media"},
		Provenance: &entitymanagerv1.Provenance{
			IntegrationName:  "lattice-video-integration",
			DataType:         "video-association",
			SourceUpdateTime: timestamppb.Now(),
		},
	})
	if err != nil {
		log.Fatalf("Failed to override entity media: %v", err)
	}
	fmt.Printf("Associated video %s with entity %s\n", ingressID, entityID)
}

```

```py title={"Python (gRPC)"}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
import uuid
from datetime import datetime, timezone
from bearer_auth import create_client
from protobuf import Oneof
from protobuf.wkt import Timestamp

from anduril.entitymanager.v1.entity_pub_pb import Entity, Provenance
from anduril.entitymanager.v1.entity_manager_api_pub_pb import OverrideEntityRequest
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync
from anduril.entitymanager.v1.media_pub_pb import Media, MediaItem, MediaType

from anduril.videomanager.v1.create_ingress_stream_request_pub_pb import (
    CreateIngressStreamRequest,
)
from anduril.videomanager.v1.rtsp_settings_pub_pb import RtspSettings
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Entity ID to associate the stream with. Replace before running.
ENTITY_ID = "<ENTITY-ID>"
# RTSP source Lattice will pull from.
RTSP_URL = "rtsp://<rtsp-host>:<port>/cam"


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API clients over Connect gRPC with authentication
    video_client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)
    em_client = create_client(EntityManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    ingress_id = str(uuid.uuid4())
    print(f"Starting RTSP ingress stream {ingress_id} from {RTSP_URL}")

    # RTSP is a pull protocol: Lattice dials the supplied URL, so no push endpoint
    # is returned.
    try:
        create_resp = video_client.create_ingress_stream(
            CreateIngressStreamRequest(
                ingress_id=ingress_id,
                title="cam-01",
                ingress=Oneof("rtsp", RtspSettings(url=RTSP_URL)),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to create ingress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully started stream with ID: {create_resp.ingress_id}")

    # Attach the new video to the asset entity. A field_path of media.media replaces
    # only the media list, leaving every other component on the entity untouched.
    now_ts = Timestamp.from_datetime(datetime.now(timezone.utc))

    try:
        em_client.override_entity(
            OverrideEntityRequest(
                entity=Entity(
                    entity_id=ENTITY_ID,
                    media=Media(
                        media=[
                            MediaItem(
                                item_identifier=ingress_id,
                                type=MediaType.VIDEO,
                            ),
                        ],
                    ),
                ),
                field_path=["media.media"],
                provenance=Provenance(
                    integration_name="lattice-video-integration",
                    data_type="video-association",
                    source_update_time=now_ts,
                ),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to override entity media: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Associated video {ingress_id} with entity {ENTITY_ID}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::{
    Entity, Media, MediaItem, MediaType, OverrideEntityRequest, Provenance,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::{
    create_ingress_stream_request::Ingress, CreateIngressStreamRequest, RtspSettings,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use prost_types::Timestamp;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;
use uuid::Uuid;

use std::env;
use std::time::{SystemTime, UNIX_EPOCH};

// Entity ID to associate the stream with. Replace before running.
const ENTITY_ID: &str = "<ENTITY-ID>";
// RTSP source Lattice will pull from.
const RTSP_URL: &str = "rtsp://<rtsp-host>:<port>/cam";

fn now_timestamp() -> Timestamp {
    let d = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
    Timestamp {
        seconds: d.as_secs() as i64,
        nanos: d.subsec_nanos() as i32,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    // Both clients share the same channel and inject the same auth metadata.
    let video_auth = auth_header.clone();
    let video_sandbox = sandbox_header.clone();
    let mut video_client = VideoManagerApiClient::with_interceptor(
        channel.clone(),
        move |mut req: Request<()>| {
            req.metadata_mut().insert("authorization", video_auth.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", video_sandbox.clone());
            Ok(req)
        },
    );
    let mut em_client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    let ingress_id = Uuid::new_v4().to_string();
    println!("Starting RTSP ingress stream {} from {}", ingress_id, RTSP_URL);

    // RTSP is a pull protocol: Lattice dials the supplied URL, so no push endpoint
    // is returned.
    let create_req = Request::new(CreateIngressStreamRequest {
        ingress_id: Some(ingress_id.clone()),
        title: "cam-01".to_string(),
        ingress: Some(Ingress::Rtsp(RtspSettings {
            url: Some(RTSP_URL.to_string()),
        })),
    });
    let create_resp = video_client.create_ingress_stream(create_req).await?.into_inner();
    println!("Successfully started stream with ID: {}", create_resp.ingress_id);

    // Attach the new video to the asset entity. A field_path of media.media replaces
    // only the media list, leaving every other component on the entity untouched.
    let override_req = Request::new(OverrideEntityRequest {
        entity: Some(Entity {
            entity_id: ENTITY_ID.to_string(),
            media: Some(Media {
                media: vec![MediaItem {
                    item_identifier: ingress_id.clone(),
                    r#type: MediaType::Video as i32,
                    relative_path: String::new(),
                }],
            }),
            ..Default::default()
        }),
        field_path: vec!["media.media".to_string()],
        provenance: Some(Provenance {
            integration_name: "lattice-video-integration".to_string(),
            data_type: "video-association".to_string(),
            source_update_time: Some(now_timestamp()),
            ..Default::default()
        }),
    });
    match em_client.override_entity(override_req).await {
        Ok(_) => println!("Associated video {} with entity {}", ingress_id, ENTITY_ID),
        Err(e) => eprintln!("Failed to override entity media: {}", e),
    }

    Ok(())
}

```

#### MPEG-TS

> **Available at the edge only**
>
> MPEG-TS ingress is supported only at the edge, in closed networks. If you connect to Lattice over the public internet, you
> can't start an MPEG-TS stream, though you can still list or inspect an existing `IngressStream` that uses it.

Before attaching a video to an asset, the asset must exist as a Lattice entity. Publish a camera entity with the
`TEMPLATE_ASSET` ontology and a `PlatformType` of `CAMERA`, then refresh it on a heartbeat so it does not expire.

#### Publish a camera

Construct an `Entity` with the components Lattice needs to render a friendly land-based camera asset on the common
operational picture (COP). `ExpiryTime` is set five minutes out and paired with a five-second publish loop, so the entity stays alive as long as the publisher runs.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to publish a camera asset entity.
// Attaching a video to an asset requires the asset to already exist as a Lattice entity, so
// publish the camera on a heartbeat before starting an ingress stream.
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"time"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
	"github.com/google/uuid"
)

const (
	cameraName          = "camera-01"
	integrationName     = "lattice_video_integration"
	dataType            = "camera_video_feed"
	publishInterval     = 5 * time.Second
	entityExpiryTtlSecs = 60
)

func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")

	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	// Stable entity ID derived from the camera's hardware identifier so the same
	// physical camera maps to the same Lattice entity across restarts.
	entityId := uuid.NewSHA1(uuid.NameSpaceURL, []byte(cameraName)).String()

	creationTime := time.Now().UTC()

	// Continuously publish the entity so it does not expire.
	for {
		latestTimestamp := time.Now().UTC()
		ctx := context.Background()

		entity := Lattice.Entity{
			EntityID:    &entityId,
			Description: Lattice.String("EO camera streaming live video"),
			Aliases: &Lattice.Aliases{
				Name: Lattice.String("Camera 01"),
				AlternateIDs: []*Lattice.AlternateID{
					{
						ID:   Lattice.String(cameraName),
						Type: Lattice.AlternateIDTypeAltIDTypeSerialNumber.Ptr(),
					},
				},
			},
			IsLive:      Lattice.Bool(true),
			CreatedTime: Lattice.Time(creationTime),
			// ExpiryTime is set in the future and paired with a shorter publish loop,
			// so the entity stays alive as long as the publisher runs.
			ExpiryTime: Lattice.Time(latestTimestamp.Add(entityExpiryTtlSecs * time.Second)),
			Ontology: &Lattice.Ontology{
				Template:     Lattice.OntologyTemplateTemplateAsset.Ptr(),
				PlatformType: Lattice.String("CAMERA"),
			},
			MilView: &Lattice.MilView{
				Disposition: Lattice.MilViewDispositionDispositionFriendly.Ptr(),
				Environment: Lattice.MilViewEnvironmentEnvironmentLand.Ptr(),
				Nationality: Lattice.MilViewNationalityNationalityUnitedStatesOfAmerica.Ptr(),
			},
			// Overall classification of the entity's data.
			DataClassification: &Lattice.Classification{
				Default: &Lattice.ClassificationInformation{
					Level: Lattice.ClassificationInformationLevelClassificationLevelsUnclassified.Ptr(),
				},
			},
			Location: &Lattice.Location{
				Position: &Lattice.Position{
					LatitudeDegrees:   Lattice.Float64(36.66633492530301),
					LongitudeDegrees:  Lattice.Float64(-116.9117674522639),
					AltitudeHaeMeters: Lattice.Float64(780.0),
				},
			},
			Provenance: &Lattice.Provenance{
				IntegrationName:  Lattice.String(integrationName),
				DataType:         Lattice.String(dataType),
				SourceUpdateTime: Lattice.Time(latestTimestamp),
			},
			Health: &Lattice.Health{
				ConnectionStatus: Lattice.HealthConnectionStatusConnectionStatusOnline.Ptr(),
				HealthStatus:     Lattice.HealthHealthStatusHealthStatusHealthy.Ptr(),
				ActiveAlerts:     []*Lattice.Alert{},
				UpdateTime:       Lattice.Time(latestTimestamp),
				// Per-subsystem health rolls up into the entity's overall
				// health_status; report at least the camera sensor itself.
				Components: []*Lattice.ComponentHealth{
					{
						ID:         Lattice.String("camera-01-eo"),
						Name:       Lattice.String("EO camera sensor"),
						Health:     Lattice.ComponentHealthHealthHealthStatusHealthy.Ptr(),
						UpdateTime: Lattice.Time(latestTimestamp),
					},
				},
			},
			// Declare the camera as a sensor so Lattice can render the sensor cone.
			Sensors: &Lattice.Sensors{
				Sensors: []*Lattice.Sensor{
					{
						SensorID:          Lattice.String("camera-01-eo"),
						SensorType:        Lattice.SensorSensorTypeSensorTypeCamera.Ptr(),
						OperationalState:  Lattice.SensorOperationalStateOperationalStateOperational.Ptr(),
						SensorDescription: Lattice.String("EO camera, H.264"),
						FieldsOfView: []*Lattice.FieldOfView{
							{
								CenterRayPose: &Lattice.EntityManagerPose{
									Orientation: &Lattice.Quaternion{
										X: Lattice.Float64(0.0),
										Y: Lattice.Float64(0.0),
										Z: Lattice.Float64(0.0),
										W: Lattice.Float64(1.0),
									},
								},
								HorizontalFov: Lattice.Float64(1.204), // ~69 deg (typical EO horizontal FoV)
								VerticalFov:   Lattice.Float64(0.954), // ~54.6 deg (typical EO vertical FoV)
								Range:         Lattice.Float64(150.0),
								Mode:          Lattice.FieldOfViewModeSensorModeSearch.Ptr(),
							},
						},
					},
				},
			},
			TaskCatalog: &Lattice.TaskCatalog{
				TaskDefinitions: []*Lattice.TaskDefinition{
					{TaskSpecificationURL: Lattice.String("type.googleapis.com/<org>.<package>.<version>.<task>")},
				},
			},
		}

		// Publish the entity
		_, err := LatticeClient.Entities.PublishEntity(ctx, &entity)
		if err != nil {
			fmt.Printf("Error publishing entity: %v\n", err)
		} else {
			fmt.Printf("Publishing asset with entity ID: %s\n", entityId)
		}

		time.Sleep(publishInterval)
	}
}

```

```java title={"Java (REST)"}
package org.example;

import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.List;
import java.util.UUID;

import com.anduril.Lattice;
import com.anduril.types.Alert;
import com.anduril.types.AlternateId;
import com.anduril.types.AlternateIdType;
import com.anduril.types.Aliases;
import com.anduril.types.Classification;
import com.anduril.types.ClassificationInformation;
import com.anduril.types.ClassificationInformationLevel;
import com.anduril.types.ComponentHealth;
import com.anduril.types.ComponentHealthHealth;
import com.anduril.types.Entity;
import com.anduril.types.EntityManagerPose;
import com.anduril.types.FieldOfView;
import com.anduril.types.FieldOfViewMode;
import com.anduril.types.Health;
import com.anduril.types.HealthConnectionStatus;
import com.anduril.types.HealthHealthStatus;
import com.anduril.types.Location;
import com.anduril.types.MilView;
import com.anduril.types.MilViewDisposition;
import com.anduril.types.MilViewEnvironment;
import com.anduril.types.MilViewNationality;
import com.anduril.types.Ontology;
import com.anduril.types.OntologyTemplate;
import com.anduril.types.Position;
import com.anduril.types.Provenance;
import com.anduril.types.Quaternion;
import com.anduril.types.Sensor;
import com.anduril.types.SensorOperationalState;
import com.anduril.types.SensorSensorType;
import com.anduril.types.Sensors;
import com.anduril.types.TaskCatalog;
import com.anduril.types.TaskDefinition;

/**
 * Publishes a camera asset entity with the Lattice SDK. Attaching a video to an asset requires the
 * asset to already exist as a Lattice entity, so publish the camera on a heartbeat before starting
 * an ingress stream.
 */
public class PublishCamera {
    private static final String CAMERA_NAME = "camera-01";
    private static final String INTEGRATION_NAME = "lattice_video_integration";
    private static final String DATA_TYPE = "camera_video_feed";
    private static final long PUBLISH_INTERVAL_MS = 5000;
    private static final long ENTITY_EXPIRY_TTL_SEC = 60;

    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        // Check required variables and ensure URL has https://
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Stable entity ID derived from the camera's hardware identifier so the same
            // physical camera maps to the same Lattice entity across restarts.
            String entityId = UUID.nameUUIDFromBytes(CAMERA_NAME.getBytes()).toString();
            Instant creationTime = Instant.now();

            // Continuously publish the camera so it does not expire while the publisher runs.
            while (true) {
                try {
                    Instant latestTimestamp = Instant.now();

                    // Construct an Entity with the components Lattice needs to render a friendly
                    // land-based camera asset on the common operational picture (COP).
                    Entity entity = Entity.builder()
                        .entityId(entityId)
                        .description("EO camera streaming live video")
                        .aliases(Aliases.builder()
                            .name("Camera 01")
                            .alternateIds(List.of(AlternateId.builder()
                                .id(CAMERA_NAME)
                                .type(AlternateIdType.ALT_ID_TYPE_SERIAL_NUMBER)
                                .build()))
                            .build())
                        .isLive(true)
                        .createdTime(OffsetDateTime.ofInstant(creationTime, ZoneOffset.UTC))
                        // ExpiryTime is set in the future and paired with a shorter publish loop,
                        // so the entity stays alive as long as the publisher runs.
                        .expiryTime(OffsetDateTime.ofInstant(
                            latestTimestamp.plus(ENTITY_EXPIRY_TTL_SEC, ChronoUnit.SECONDS), ZoneOffset.UTC))
                        .ontology(Ontology.builder()
                            .template(OntologyTemplate.TEMPLATE_ASSET)
                            .platformType("CAMERA")
                            .build())
                        .milView(MilView.builder()
                            .disposition(MilViewDisposition.DISPOSITION_FRIENDLY)
                            .environment(MilViewEnvironment.ENVIRONMENT_LAND)
                            .nationality(MilViewNationality.NATIONALITY_UNITED_STATES_OF_AMERICA)
                            .build())
                        // Overall classification of the entity's data.
                        .dataClassification(Classification.builder()
                            .default_(ClassificationInformation.builder()
                                .level(ClassificationInformationLevel.CLASSIFICATION_LEVELS_UNCLASSIFIED)
                                .build())
                            .build())
                        .location(Location.builder()
                            .position(Position.builder()
                                .latitudeDegrees(36.66633492530301)
                                .longitudeDegrees(-116.9117674522639)
                                .altitudeHaeMeters(780.0)
                                .build())
                            .build())
                        .provenance(Provenance.builder()
                            .integrationName(INTEGRATION_NAME)
                            .dataType(DATA_TYPE)
                            .sourceUpdateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                            .build())
                        .health(Health.builder()
                            .connectionStatus(HealthConnectionStatus.CONNECTION_STATUS_ONLINE)
                            .healthStatus(HealthHealthStatus.HEALTH_STATUS_HEALTHY)
                            .activeAlerts(Collections.<Alert>emptyList())
                            .updateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                            // Per-subsystem health rolls up into the entity's overall
                            // health_status; report at least the camera sensor itself.
                            .components(List.of(ComponentHealth.builder()
                                .id("camera-01-eo")
                                .name("EO camera sensor")
                                .health(ComponentHealthHealth.HEALTH_STATUS_HEALTHY)
                                .updateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                                .build()))
                            .build())
                        // Declare the camera as a sensor so Lattice can render the sensor cone.
                        .sensors(Sensors.builder()
                            .sensors(List.of(Sensor.builder()
                                .sensorId("camera-01-eo")
                                .sensorType(SensorSensorType.SENSOR_TYPE_CAMERA)
                                .operationalState(SensorOperationalState.OPERATIONAL_STATE_OPERATIONAL)
                                .sensorDescription("EO camera, H.264")
                                .fieldsOfView(List.of(FieldOfView.builder()
                                    .centerRayPose(EntityManagerPose.builder()
                                        .orientation(Quaternion.builder()
                                            .x(0.0)
                                            .y(0.0)
                                            .z(0.0)
                                            .w(1.0)
                                            .build())
                                        .build())
                                    .horizontalFov(1.204f) // ~69 deg (typical EO horizontal FoV)
                                    .verticalFov(0.954f) // ~54.6 deg (typical EO vertical FoV)
                                    .range(150.0f)
                                    .mode(FieldOfViewMode.SENSOR_MODE_SEARCH)
                                    .build()))
                                .build()))
                            .build())
                        .taskCatalog(TaskCatalog.builder()
                            .taskDefinitions(List.of(TaskDefinition.builder()
                                .taskSpecificationUrl("type.googleapis.com/<org>.<package>.<version>.<task>")
                                .build()))
                            .build())
                        .build();

                    // Publish the entity
                    client.entities().publishEntity(entity);
                    System.out.println("Publishing asset with entity ID: " + entityId);

                    Thread.sleep(PUBLISH_INTERVAL_MS);
                } catch (Exception e) {
                    System.err.println("Error publishing entity: " + e.getMessage());
                    Thread.sleep(PUBLISH_INTERVAL_MS);
                }
            }
        } catch (Exception e) {
            System.err.println("Failed to initialize: " + e.getMessage());
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the public Lattice REST SDK (anduril-lattice-sdk) to publish a
# camera asset entity. Attaching a video to an asset requires the asset to already exist as a
# Lattice entity, so publish the camera on a heartbeat before starting an ingress stream.

import os
import sys
import time
import uuid
from datetime import datetime, timedelta, timezone

from anduril import (
    Aliases,
    AlternateId,
    Classification,
    ClassificationInformation,
    ComponentHealth,
    FieldOfView,
    Health,
    Lattice,
    Location,
    MilView,
    Ontology,
    Position,
    Provenance,
    Quaternion,
    Sensor,
    Sensors,
    EntityManagerPose,
    TaskCatalog,
    TaskDefinition,
)

CAMERA_NAME = "camera-01"
INTEGRATION_NAME = "lattice_video_integration"
DATA_TYPE = "camera_video_feed"
PUBLISH_INTERVAL_SEC = 5
ENTITY_EXPIRY_TTL_SEC = 60


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")

    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")
    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables.", file=sys.stderr)
        sys.exit(1)

    client = Lattice(
        base_url=f"https://{lattice_endpoint}",
        token=lambda: str(environment_token),
        # Remove the following header if you are not developing on Sandboxes.
        headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"},
    )

    # Stable entity_id derived from the camera's hardware identifier so the
    # same physical camera maps to the same Lattice entity across restarts.
    entity_id = str(uuid.uuid5(uuid.NAMESPACE_URL, CAMERA_NAME))

    creation_time = datetime.now(timezone.utc)

    try:
        while True:
            latest_timestamp = datetime.now(timezone.utc)

            client.entities.publish_entity(
                entity_id=entity_id,
                description="EO camera streaming live video",
                aliases=Aliases(
                    name="Camera 01",
                    alternate_ids=[
                        AlternateId(id=CAMERA_NAME, type="ALT_ID_TYPE_SERIAL_NUMBER"),
                    ],
                ),
                is_live=True,
                created_time=creation_time,
                # ExpiryTime is set in the future and paired with a shorter publish loop,
                # so the entity stays alive as long as the publisher runs.
                expiry_time=latest_timestamp + timedelta(seconds=ENTITY_EXPIRY_TTL_SEC),
                ontology=Ontology(template="TEMPLATE_ASSET", platform_type="CAMERA"),
                mil_view=MilView(
                    disposition="DISPOSITION_FRIENDLY",
                    environment="ENVIRONMENT_LAND",
                    nationality="NATIONALITY_UNITED_STATES_OF_AMERICA",
                ),
                # Overall classification of the entity's data.
                data_classification=Classification(
                    default=ClassificationInformation(
                        level="CLASSIFICATION_LEVELS_UNCLASSIFIED",
                    ),
                ),
                location=Location(
                    position=Position(
                        latitude_degrees=36.66633492530301,
                        longitude_degrees=-116.9117674522639,
                        altitude_hae_meters=780.0,
                    ),
                ),
                provenance=Provenance(
                    integration_name=INTEGRATION_NAME,
                    data_type=DATA_TYPE,
                    source_update_time=latest_timestamp,
                ),
                health=Health(
                    connection_status="CONNECTION_STATUS_ONLINE",
                    health_status="HEALTH_STATUS_HEALTHY",
                    active_alerts=[],
                    update_time=latest_timestamp,
                    # Per-subsystem health rolls up into the entity's overall
                    # health_status; report at least the camera sensor itself.
                    components=[
                        ComponentHealth(
                            id="camera-01-eo",
                            name="EO camera sensor",
                            health="HEALTH_STATUS_HEALTHY",
                            update_time=latest_timestamp,
                        ),
                    ],
                ),
                # Declare the camera as a sensor so Lattice can render the sensor cone.
                sensors=Sensors(
                    sensors=[
                        Sensor(
                            sensor_id="camera-01-eo",
                            sensor_type="SENSOR_TYPE_CAMERA",
                            operational_state="OPERATIONAL_STATE_OPERATIONAL",
                            sensor_description="EO camera, H.264",
                            fields_of_view=[
                                FieldOfView(
                                    center_ray_pose=EntityManagerPose(
                                        orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0),
                                    ),
                                    horizontal_fov=1.204,  # ~69 deg (typical EO horizontal FoV)
                                    vertical_fov=0.954,  # ~54.6 deg (typical EO vertical FoV)
                                    range=150.0,
                                    mode="SENSOR_MODE_SEARCH",
                                ),
                            ],
                        ),
                    ],
                ),
                task_catalog=TaskCatalog(
                    task_definitions=[
                        TaskDefinition(task_specification_url="type.googleapis.com/<org>.<package>.<version>.<task>"),
                    ],
                ),
            )
            print(f"Publishing asset with entity ID: {entity_id}")
            time.sleep(PUBLISH_INTERVAL_SEC)

    except (KeyboardInterrupt, SystemExit):
        print(">>> Exiting...")
    except Exception as error:
        print(f"Error publishing entity: {error}", file=sys.stderr)


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
import { LatticeClient } from "@anduril-industries/lattice-sdk";
import { v5 as uuidv5 } from "uuid";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;

// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

const cameraName = "camera-01";
const integrationName = "lattice_video_integration";
const dataType = "camera_video_feed";
const publishIntervalMs = 5000;
const entityExpiryTtlSecs = 60;

// Stable entity ID derived from the camera's hardware identifier so the same
// physical camera maps to the same Lattice entity across restarts.
const entityId = uuidv5(cameraName, uuidv5.URL);
const creationTime = new Date().toISOString();

async function App() {
    const latestTimestamp = new Date();
    try {
        // Attaching a video to an asset requires the asset to already exist as a Lattice
        // entity, so publish the camera on a heartbeat before starting an ingress stream.
        await client.entities.publishEntity(
            {
                entityId: entityId,
                description: "EO camera streaming live video",
                aliases: {
                    name: "Camera 01",
                    alternateIds: [
                        { id: cameraName, type: "ALT_ID_TYPE_SERIAL_NUMBER" }
                    ]
                },
                isLive: true,
                createdTime: creationTime,
                // ExpiryTime is set in the future and paired with a shorter publish loop,
                // so the entity stays alive as long as the publisher runs.
                expiryTime: new Date(latestTimestamp.getTime() + entityExpiryTtlSecs * 1000).toISOString(),
                ontology: {
                    template: "TEMPLATE_ASSET",
                    platformType: "CAMERA"
                },
                milView: {
                    disposition: "DISPOSITION_FRIENDLY",
                    environment: "ENVIRONMENT_LAND",
                    nationality: "NATIONALITY_UNITED_STATES_OF_AMERICA"
                },
                // Overall classification of the entity's data.
                dataClassification: {
                    default: {
                        level: "CLASSIFICATION_LEVELS_UNCLASSIFIED"
                    }
                },
                location: {
                    position: {
                        latitudeDegrees: 36.66633492530301,
                        longitudeDegrees: -116.9117674522639,
                        altitudeHaeMeters: 780.0
                    }
                },
                provenance: {
                    integrationName: integrationName,
                    dataType: dataType,
                    sourceUpdateTime: latestTimestamp.toISOString()
                },
                health: {
                    connectionStatus: "CONNECTION_STATUS_ONLINE",
                    healthStatus: "HEALTH_STATUS_HEALTHY",
                    activeAlerts: [],
                    updateTime: latestTimestamp.toISOString(),
                    // Per-subsystem health rolls up into the entity's overall
                    // healthStatus; report at least the camera sensor itself.
                    components: [
                        {
                            id: "camera-01-eo",
                            name: "EO camera sensor",
                            health: "HEALTH_STATUS_HEALTHY",
                            updateTime: latestTimestamp.toISOString()
                        }
                    ]
                },
                // Declare the camera as a sensor so Lattice can render the sensor cone.
                sensors: {
                    sensors: [
                        {
                            sensorId: "camera-01-eo",
                            sensorType: "SENSOR_TYPE_CAMERA",
                            operationalState: "OPERATIONAL_STATE_OPERATIONAL",
                            sensorDescription: "EO camera, H.264",
                            fieldsOfView: [
                                {
                                    centerRayPose: {
                                        orientation: { x: 0.0, y: 0.0, z: 0.0, w: 1.0 }
                                    },
                                    horizontalFov: 1.204, // ~69 deg (typical EO horizontal FoV)
                                    verticalFov: 0.954, // ~54.6 deg (typical EO vertical FoV)
                                    range: 150.0,
                                    mode: "SENSOR_MODE_SEARCH"
                                }
                            ]
                        }
                    ]
                },
                taskCatalog: {
                    taskDefinitions: [
                        { taskSpecificationUrl: "type.googleapis.com/<org>.<package>.<version>.<task>" }
                    ]
                }
            }
        );
        console.log(`Publishing asset with entity ID: ${entityId}`);
    } catch (error) {
        console.log(`Encountered the following error while publishing entity: ${error}`);
    }
}

(async function runIndefinitely() {
    while (true) {
        await App();
        await new Promise(resolve => setTimeout(resolve, publishIntervalMs));
    }
})();

```

```go title={"Go (gRPC)"}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/types/known/timestamppb"
	"google.golang.org/protobuf/types/known/wrapperspb"
	ontologyv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/ontology/v1"
	tasksv2 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/tasks/v2"
	_type "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/type"
)

const (
	cameraName      = "camera-01"
	integrationName = "lattice_video_integration"
	dataType        = "camera_video_feed"
	publishInterval = 5 * time.Second
	entityExpiryTTL = 60 * time.Second
	publishTimeout  = 10 * time.Second
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}

	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}
	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	client := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	// Stable entity_id derived from the camera's hardware identifier so the
	// same physical camera maps to the same Lattice entity across restarts.
	entityId := uuid.NewSHA1(uuid.NameSpaceURL, []byte(cameraName)).String()

	creationTimestamp := timestamppb.New(time.Now().UTC())

	for {
		latestTimestamp := time.Now().UTC()
		ctx, cancel := context.WithTimeout(context.Background(), publishTimeout)

		entity := &entitymanagerv1.Entity{
			EntityId:    entityId,
			Description: "EO camera streaming live video",
			Aliases: &entitymanagerv1.Aliases{
				Name: "Camera 01",
				AlternateIds: []*entitymanagerv1.AlternateId{
					{
						Id:   cameraName,
						Type: entitymanagerv1.AltIdType_ALT_ID_TYPE_SERIAL_NUMBER,
					},
				},
			},
			IsLive:      true,
			CreatedTime: creationTimestamp,
			ExpiryTime:  timestamppb.New(latestTimestamp.Add(entityExpiryTTL)),
			Ontology: &entitymanagerv1.Ontology{
				Template:     entitymanagerv1.Template_TEMPLATE_ASSET,
				PlatformType: "CAMERA",
			},
			MilView: &entitymanagerv1.MilView{
				Disposition: ontologyv1.Disposition_DISPOSITION_FRIENDLY,
				Environment: ontologyv1.Environment_ENVIRONMENT_LAND,
				Nationality: ontologyv1.Nationality_NATIONALITY_UNITED_STATES_OF_AMERICA,
			},
			// Overall classification of the entity's data.
			DataClassification: &entitymanagerv1.Classification{
				Default: &entitymanagerv1.ClassificationInformation{
					Level: entitymanagerv1.ClassificationLevels_CLASSIFICATION_LEVELS_UNCLASSIFIED,
				},
			},
			Location: &entitymanagerv1.Location{
				Position: &entitymanagerv1.Position{
					LatitudeDegrees:   36.66633492530301,
					LongitudeDegrees:  -116.9117674522639,
					AltitudeHaeMeters: wrapperspb.Double(780.0),
				},
			},
			Provenance: &entitymanagerv1.Provenance{
				IntegrationName:  integrationName,
				DataType:         dataType,
				SourceUpdateTime: timestamppb.New(latestTimestamp),
			},
			Health: &entitymanagerv1.Health{
				ConnectionStatus: entitymanagerv1.ConnectionStatus_CONNECTION_STATUS_ONLINE,
				HealthStatus:     entitymanagerv1.HealthStatus_HEALTH_STATUS_HEALTHY,
				ActiveAlerts:     []*entitymanagerv1.Alert{},
				UpdateTime:       timestamppb.New(latestTimestamp),
				// Per-subsystem health rolls up into the entity's overall
				// health_status; report at least the camera sensor itself.
				Components: []*entitymanagerv1.ComponentHealth{
					{
						Id:         "camera-01-eo",
						Name:       "EO camera sensor",
						Health:     entitymanagerv1.HealthStatus_HEALTH_STATUS_HEALTHY,
						UpdateTime: timestamppb.New(latestTimestamp),
					},
				},
			},
			Sensors: &entitymanagerv1.Sensors{
				Sensors: []*entitymanagerv1.Sensor{
					{
						SensorId:          "camera-01-eo",
						SensorType:        entitymanagerv1.SensorType_SENSOR_TYPE_CAMERA,
						OperationalState:  entitymanagerv1.OperationalState_OPERATIONAL_STATE_OPERATIONAL,
						SensorDescription: "EO camera, H.264",
						FieldsOfView: []*entitymanagerv1.FieldOfView{
							{
								CenterRayPose: &entitymanagerv1.Pose{
									Orientation: &_type.Quaternion{
										X: 0.0,
										Y: 0.0,
										Z: 0.0,
										W: 1.0,
									},
								},
								HorizontalFov: 1.204, // ~69 deg (typical EO horizontal FoV)
								VerticalFov:   0.954, // ~54.6 deg (typical EO vertical FoV)
								Range:         wrapperspb.Float(150.0),
								Mode:          entitymanagerv1.SensorMode_SENSOR_MODE_SEARCH,
							},
						},
					},
				},
			},
			TaskCatalog: &tasksv2.TaskCatalog{
				TaskDefinitions: []*tasksv2.TaskDefinition{
					{TaskSpecificationUrl: "type.googleapis.com/<org>.<package>.<version>.<task>"},
				},
			},
		}

		_, err := client.PublishEntity(ctx, &entitymanagerv1.PublishEntityRequest{Entity: entity})
		cancel()

		if err != nil {
			fmt.Printf("Error publishing entity: %v\n", err)
		} else {
			fmt.Printf("Publishing asset with entity ID: %s\n", entityId)
		}

		time.Sleep(publishInterval)
	}
}

```

```py title={"Python (gRPC)"}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
import time
import uuid
from datetime import datetime, timedelta, timezone
from bearer_auth import create_client
from protobuf.wkt import DoubleValue, FloatValue, Timestamp

from anduril.entitymanager.v1.classification_pub_pb import (
    Classification,
    ClassificationInformation,
    ClassificationLevels,
)
from anduril.entitymanager.v1.entity_pub_pb import (
    Aliases,
    AlternateId,
    Entity,
    Provenance,
)
from anduril.entitymanager.v1.entity_manager_api_pub_pb import PublishEntityRequest
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync
from anduril.entitymanager.v1.health_status_pub_pb import (
    ComponentHealth,
    ConnectionStatus,
    Health,
    HealthStatus,
)
from anduril.entitymanager.v1.location_pub_pb import Location, Pose, Position
from anduril.entitymanager.v1.ontology_pub_pb import MilView, Ontology
from anduril.entitymanager.v1.sensors_pub_pb import (
    FieldOfView,
    OperationalState,
    Sensor,
    SensorMode,
    Sensors,
    SensorType,
)
from anduril.entitymanager.v1.types_pub_pb import AltIdType, Template
from anduril.ontology.v1.type_pub_pb import Disposition, Environment, Nationality
from anduril.tasks.v2.catalog_pub_pb import TaskCatalog, TaskDefinition
from anduril.type.coords_pub_pb import Quaternion


CAMERA_NAME = "camera-01"
INTEGRATION_NAME = "lattice_video_integration"
DATA_TYPE = "camera_video_feed"
PUBLISH_INTERVAL_SEC = 5
ENTITY_EXPIRY_TTL_SEC = 60
PUBLISH_TIMEOUT_SEC = 10


def to_timestamp(dt: datetime) -> Timestamp:
    return Timestamp.from_datetime(dt)


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(EntityManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Stable entity_id derived from the camera's hardware identifier so the
    # same physical camera maps to the same Lattice entity across restarts.
    entity_id = str(uuid.uuid5(uuid.NAMESPACE_URL, CAMERA_NAME))

    creation_timestamp = to_timestamp(datetime.now(timezone.utc))

    while True:
        latest = datetime.now(timezone.utc)
        latest_ts = to_timestamp(latest)
        expiry_ts = to_timestamp(latest + timedelta(seconds=ENTITY_EXPIRY_TTL_SEC))

        entity = Entity(
            entity_id=entity_id,
            description="EO camera streaming live video",
            aliases=Aliases(
                name="Camera 01",
                alternate_ids=[
                    AlternateId(id=CAMERA_NAME, type=AltIdType.SERIAL_NUMBER),
                ],
            ),
            is_live=True,
            created_time=creation_timestamp,
            expiry_time=expiry_ts,
            ontology=Ontology(template=Template.ASSET, platform_type="CAMERA"),
            mil_view=MilView(
                disposition=Disposition.FRIENDLY,
                environment=Environment.LAND,
                nationality=Nationality.UNITED_STATES_OF_AMERICA,
            ),
            # Overall classification of the entity's data.
            data_classification=Classification(
                default=ClassificationInformation(
                    level=ClassificationLevels.UNCLASSIFIED,
                ),
            ),
            location=Location(
                position=Position(
                    latitude_degrees=36.66633492530301,
                    longitude_degrees=-116.9117674522639,
                    altitude_hae_meters=DoubleValue(value=780.0),
                ),
            ),
            provenance=Provenance(
                integration_name=INTEGRATION_NAME,
                data_type=DATA_TYPE,
                source_update_time=latest_ts,
            ),
            health=Health(
                connection_status=ConnectionStatus.ONLINE,
                health_status=HealthStatus.HEALTHY,
                active_alerts=[],
                update_time=latest_ts,
                # Per-subsystem health rolls up into the entity's overall
                # health_status; report at least the camera sensor itself.
                components=[
                    ComponentHealth(
                        id="camera-01-eo",
                        name="EO camera sensor",
                        health=HealthStatus.HEALTHY,
                        update_time=latest_ts,
                    ),
                ],
            ),
            sensors=Sensors(
                sensors=[
                    Sensor(
                        sensor_id="camera-01-eo",
                        sensor_type=SensorType.CAMERA,
                        operational_state=OperationalState.OPERATIONAL,
                        sensor_description="EO camera, H.264",
                        fields_of_view=[
                            FieldOfView(
                                center_ray_pose=Pose(
                                    orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0),
                                ),
                                horizontal_fov=1.204,  # ~69 deg (typical EO horizontal FoV)
                                vertical_fov=0.954,  # ~54.6 deg (typical EO vertical FoV)
                                range=FloatValue(value=150.0),
                                mode=SensorMode.SEARCH,
                            ),
                        ],
                    ),
                ],
            ),
            task_catalog=TaskCatalog(
                task_definitions=[
                    TaskDefinition(task_specification_url="type.googleapis.com/<org>.<package>.<version>.<task>"),
                ],
            ),
        )

        try:
            client.publish_entity(
                PublishEntityRequest(entity=entity),
                timeout_ms=PUBLISH_TIMEOUT_SEC * 1000,
            )
            print(f"Publishing asset with entity ID: {entity_id}")
        except Exception as error:
            print(f"Error publishing entity: {error}", file=sys.stderr)

        time.sleep(PUBLISH_INTERVAL_SEC)


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::{
    Aliases, AltIdType, AlternateId, Classification, ClassificationInformation,
    ClassificationLevels, ComponentHealth, ConnectionStatus, Entity, FieldOfView, Health,
    HealthStatus, Location, MilView, Ontology, OperationalState, Pose, Position,
    Provenance, PublishEntityRequest, Sensor, SensorMode, SensorType, Sensors, Template,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::ontology::v1::{
    Disposition, Environment, Nationality,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::r#type::Quaternion;
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::tasks::v2::{
    TaskCatalog, TaskDefinition,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use prost_types::Timestamp;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;
use uuid::Uuid;

use std::env;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

const CAMERA_NAME: &str = "camera-01";
const INTEGRATION_NAME: &str = "lattice_video_integration";
const DATA_TYPE: &str = "camera_video_feed";
const PUBLISH_INTERVAL_SECS: u64 = 5;
const ENTITY_EXPIRY_TTL_SECS: i64 = 60;

fn now_timestamp() -> Timestamp {
    let d = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
    Timestamp {
        seconds: d.as_secs() as i64,
        nanos: d.subsec_nanos() as i32,
    }
}

fn add_seconds(ts: &Timestamp, seconds: i64) -> Timestamp {
    Timestamp {
        seconds: ts.seconds + seconds,
        nanos: ts.nanos,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Stable entity_id derived from the camera's hardware identifier so the
    // same physical camera maps to the same Lattice entity across restarts.
    let entity_id = Uuid::new_v5(&Uuid::NAMESPACE_URL, CAMERA_NAME.as_bytes()).to_string();
    let creation_timestamp = now_timestamp();

    // Refresh the entity on a heartbeat so it does not expire while the publisher runs.
    loop {
        let latest_timestamp = now_timestamp();

        let entity = Entity {
            entity_id: entity_id.clone(),
            description: "EO camera streaming live video".to_string(),
            aliases: Some(Aliases {
                name: "Camera 01".to_string(),
                alternate_ids: vec![AlternateId {
                    id: CAMERA_NAME.to_string(),
                    r#type: AltIdType::SerialNumber as i32,
                }],
            }),
            is_live: true,
            created_time: Some(creation_timestamp.clone()),
            // ExpiryTime is set in the future and paired with a shorter publish loop,
            // so the entity stays alive as long as the publisher runs.
            expiry_time: Some(add_seconds(&latest_timestamp, ENTITY_EXPIRY_TTL_SECS)),
            ontology: Some(Ontology {
                template: Template::Asset as i32,
                platform_type: "CAMERA".to_string(),
                ..Default::default()
            }),
            mil_view: Some(MilView {
                disposition: Disposition::Friendly as i32,
                environment: Environment::Land as i32,
                nationality: Nationality::UnitedStatesOfAmerica as i32,
            }),
            // Overall classification of the entity's data.
            data_classification: Some(Classification {
                default: Some(ClassificationInformation {
                    level: ClassificationLevels::Unclassified as i32,
                    ..Default::default()
                }),
                ..Default::default()
            }),
            location: Some(Location {
                position: Some(Position {
                    latitude_degrees: 36.66633492530301,
                    longitude_degrees: -116.9117674522639,
                    altitude_hae_meters: Some(780.0),
                    ..Default::default()
                }),
                ..Default::default()
            }),
            provenance: Some(Provenance {
                integration_name: INTEGRATION_NAME.to_string(),
                data_type: DATA_TYPE.to_string(),
                source_update_time: Some(latest_timestamp.clone()),
                ..Default::default()
            }),
            health: Some(Health {
                connection_status: ConnectionStatus::Online as i32,
                health_status: HealthStatus::Healthy as i32,
                active_alerts: vec![],
                update_time: Some(latest_timestamp.clone()),
                // Per-subsystem health rolls up into the entity's overall
                // health_status; report at least the camera sensor itself.
                components: vec![ComponentHealth {
                    id: "camera-01-eo".to_string(),
                    name: "EO camera sensor".to_string(),
                    health: HealthStatus::Healthy as i32,
                    update_time: Some(latest_timestamp.clone()),
                    ..Default::default()
                }],
            }),
            // Declare the camera as a sensor so Lattice can render the sensor cone.
            sensors: Some(Sensors {
                sensors: vec![Sensor {
                    sensor_id: "camera-01-eo".to_string(),
                    sensor_type: SensorType::Camera as i32,
                    operational_state: OperationalState::Operational as i32,
                    sensor_description: "EO camera, H.264".to_string(),
                    fields_of_view: vec![FieldOfView {
                        center_ray_pose: Some(Pose {
                            orientation: Some(Quaternion {
                                x: 0.0,
                                y: 0.0,
                                z: 0.0,
                                w: 1.0,
                            }),
                            ..Default::default()
                        }),
                        horizontal_fov: 1.204, // ~69 deg (typical EO horizontal FoV)
                        vertical_fov: 0.954,   // ~54.6 deg (typical EO vertical FoV)
                        range: Some(150.0),
                        mode: SensorMode::Search as i32,
                        ..Default::default()
                    }],
                    ..Default::default()
                }],
            }),
            task_catalog: Some(TaskCatalog {
                task_definitions: vec![TaskDefinition {
                    task_specification_url: "type.googleapis.com/<org>.<package>.<version>.<task>"
                        .to_string(),
                }],
            }),
            ..Default::default()
        };

        let request = Request::new(PublishEntityRequest {
            entity: Some(entity),
        });
        match client.publish_entity(request).await {
            Ok(_) => println!("Publishing asset with entity ID: {}", entity_id),
            Err(e) => eprintln!("Error publishing entity: {}", e),
        }

        tokio::time::sleep(Duration::from_secs(PUBLISH_INTERVAL_SECS)).await;
    }
}

```

> **Info**
>
> The heartbeat loop is the simplest pattern for keeping a transient asset alive. For long-running deployments,
> consider publishing from a supervised service that resets `ExpiryTime` based on hardware health rather than a fixed interval.

#### Allocate a stream endpoint

If your source can't accept inbound RTSP connections, for example an encoder behind a NAT gateway or a push-only device,
you can request Lattice to create an endpoint where you send transport stream packets over UDP.

Send an empty `MpegTsSettings` in the request. The response contains `mpeg_ts.url`, the endpoint your
video encoder pushes to. The example also calls `OverrideEntity` to attach the new video to the camera asset:

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to start an MPEG-TS ingress stream and
// attach it to a camera asset entity.
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"time"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
	"github.com/google/uuid"
)

// Entity ID to associate the stream with. Replace before running.
const entityID = "<ENTITY-ID>"

func main() {
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")

	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	ctx := context.Background()

	// Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
	ingressID := uuid.New().String()
	fmt.Printf("Starting MPEG-TS ingress stream %s\n", ingressID)

	// An empty MpegTsSettings tells Lattice to allocate a UDP push endpoint; the URL comes back
	// on the response's mpeg_ts details. MPEG-TS ingress may be disabled per deployment (for
	// example, in the cloud); such a request is rejected, so be ready to fall back to RTSP or SRT.
	createResp, err := LatticeClient.Video.CreateIngressStream(ctx, &Lattice.CreateIngressStreamRequest{
		IngressID: &ingressID,
		Title:     Lattice.String("cam-01"),
		MpegTs:    &Lattice.MpegTsSettings{},
	})
	if err != nil {
		fmt.Printf("Failed to create ingress stream: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Successfully started stream with ID: %s\n", *createResp.GetIngressID())

	if createResp.GetMpegTs() == nil || createResp.GetMpegTs().GetURL() == nil {
		fmt.Println("Server did not return an MPEG-TS ingress URL")
		os.Exit(1)
	}
	fmt.Printf("Push MPEG-TS to: %s\n", *createResp.GetMpegTs().GetURL())

	// Attach the new video to the asset entity. A field_path of media.media replaces
	// only the media list, leaving every other component on the entity untouched.
	_, err = LatticeClient.Entities.OverrideEntity(ctx, &Lattice.EntityOverride{
		EntityID:  entityID,
		FieldPath: "media.media",
		Entity: &Lattice.Entity{
			EntityID: Lattice.String(entityID),
			Media: &Lattice.Media{
				Media: []*Lattice.MediaItem{
					{
						ItemIdentifier: &ingressID,
						Type:           Lattice.MediaItemTypeMediaTypeVideo.Ptr(),
					},
				},
			},
		},
		Provenance: &Lattice.Provenance{
			IntegrationName:  Lattice.String("lattice-video-integration"),
			DataType:         Lattice.String("video-association"),
			SourceUpdateTime: Lattice.Time(time.Now().UTC()),
		},
	})
	if err != nil {
		fmt.Printf("Failed to override entity media: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Associated video %s with entity %s\n", ingressID, entityID)
}

```

```java title={"Java (REST)"}
package org.example;

import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.Collections;
import java.util.UUID;

import com.anduril.Lattice;
import com.anduril.resources.entities.requests.EntityOverride;
import com.anduril.resources.video.requests.CreateIngressStreamRequest;
import com.anduril.types.CreateIngressStreamResponse;
import com.anduril.types.Entity;
import com.anduril.types.Media;
import com.anduril.types.MediaItem;
import com.anduril.types.MediaItemType;
import com.anduril.types.MpegTsIngress;
import com.anduril.types.Provenance;

/**
 * Starts an MPEG-TS ingress stream with the Lattice SDK and attaches it to a camera asset entity.
 */
public class CreateIngressStreamMpegTs {
    // Entity ID to associate the stream with. Replace before running.
    private static final String ENTITY_ID = "<ENTITY-ID>";

    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        // Check required variables and ensure the endpoint has its scheme
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
            String ingressId = UUID.randomUUID().toString();
            System.out.println("Starting MPEG-TS ingress stream " + ingressId);

            // An empty MpegTsSettings tells Lattice to allocate a UDP push endpoint; the URL comes
            // back on the response's mpeg_ts details. MpegTsSettings is an empty message, so send an
            // empty map. MPEG-TS ingress may be disabled per deployment (for example, in the cloud);
            // such a request is rejected, so be ready to fall back to RTSP or SRT.
            CreateIngressStreamResponse response = client.video().createIngressStream(
                CreateIngressStreamRequest.builder()
                    .ingressId(ingressId)
                    .title("cam-01")
                    .mpegTs(Collections.emptyMap())
                    .build());
            System.out.println("Successfully started stream with ID: " + response.getIngressId().orElse(ingressId));

            MpegTsIngress mpegTs = response.getMpegTs().orElse(null);
            if (mpegTs == null || mpegTs.getUrl().isEmpty()) {
                System.err.println("Server did not return an MPEG-TS ingress URL.");
                System.exit(1);
            }
            System.out.println("Push MPEG-TS to: " + mpegTs.getUrl().orElse(""));

            // Attach the new video to the asset entity. A field_path of media.media replaces
            // only the media list, leaving every other component on the entity untouched.
            MediaItem mediaItem = MediaItem.builder()
                .itemIdentifier(ingressId)
                .type(MediaItemType.MEDIA_TYPE_VIDEO)
                .build();
            EntityOverride entityOverride = EntityOverride.builder()
                .entity(Entity.builder()
                    .entityId(ENTITY_ID)
                    .media(Media.builder()
                        .media(Collections.singletonList(mediaItem))
                        .build())
                    .build())
                .provenance(Provenance.builder()
                    .integrationName("lattice-video-integration")
                    .dataType("video-association")
                    .sourceUpdateTime(OffsetDateTime.now(ZoneOffset.UTC))
                    .build())
                .build();
            client.entities().overrideEntity(ENTITY_ID, "media.media", entityOverride);

            System.out.println("Associated video " + ingressId + " with entity " + ENTITY_ID);
        } catch (Exception e) {
            System.err.println("Exception: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to start an MPEG-TS ingress
# stream and attach it to a camera asset entity.

import os
import sys
import uuid
from datetime import datetime, timezone

from anduril import Entity, Lattice, Media, MediaItem, Provenance

# Entity ID to associate the stream with. Replace before running.
ENTITY_ID = "<ENTITY-ID>"


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")

    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")
    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables.", file=sys.stderr)
        sys.exit(1)

    client = Lattice(
        base_url=f"https://{lattice_endpoint}",
        token=lambda: str(environment_token),
        # Remove the following header if you are not developing on Sandboxes.
        headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"},
    )

    # Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    ingress_id = str(uuid.uuid4())
    print(f"Starting MPEG-TS ingress stream {ingress_id}")

    # An empty MpegTsSettings tells Lattice to allocate a UDP push endpoint; the URL comes back
    # on the response's mpeg_ts details. MpegTsSettings is an empty message, so send {}.
    # MPEG-TS ingress may be disabled per deployment (for example, in the cloud); such a request
    # is rejected, so be ready to fall back to RTSP or SRT.
    try:
        create_resp = client.video.create_ingress_stream(
            ingress_id=ingress_id,
            title="cam-01",
            mpeg_ts={},
        )
    except Exception as error:
        print(f"Failed to create ingress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully started stream with ID: {create_resp.ingress_id}")

    if not create_resp.mpeg_ts or not create_resp.mpeg_ts.url:
        print("Server did not return an MPEG-TS ingress URL.", file=sys.stderr)
        sys.exit(1)
    print(f"Push MPEG-TS to: {create_resp.mpeg_ts.url}")

    # Attach the new video to the asset entity. A field_path of media.media replaces
    # only the media list, leaving every other component on the entity untouched.
    try:
        client.entities.override_entity(
            entity_id=ENTITY_ID,
            field_path="media.media",
            entity=Entity(
                entity_id=ENTITY_ID,
                media=Media(
                    media=[
                        MediaItem(
                            item_identifier=ingress_id,
                            type="MEDIA_TYPE_VIDEO",
                        ),
                    ],
                ),
            ),
            provenance=Provenance(
                integration_name="lattice-video-integration",
                data_type="video-association",
                source_update_time=datetime.now(timezone.utc),
            ),
        )
    except Exception as error:
        print(f"Failed to override entity media: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Associated video {ingress_id} with entity {ENTITY_ID}")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
import { LatticeClient } from "@anduril-industries/lattice-sdk";
import { v4 as uuidv4 } from "uuid";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;

// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Entity ID to associate the stream with. Replace before running.
const entityId = "<ENTITY-ID>";

async function App() {
    // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    const ingressId = uuidv4();
    console.log(`Starting MPEG-TS ingress stream ${ingressId}`);

    try {
        // An empty MpegTsSettings tells Lattice to allocate a UDP push endpoint; the URL comes back
        // on the response's mpegTs details. MPEG-TS ingress may be disabled per deployment (for
        // example, in the cloud); such a request is rejected, so be ready to fall back to RTSP or SRT.
        const createResp = await client.video.createIngressStream({
            ingressId: ingressId,
            title: "cam-01",
            mpegTs: {}
        });
        console.log(`Successfully started stream with ID: ${createResp.ingressId}`);

        if (!createResp.mpegTs?.url) {
            console.log("Server did not return an MPEG-TS ingress URL.");
            return;
        }
        console.log(`Push MPEG-TS to: ${createResp.mpegTs.url}`);

        // Attach the new video to the asset entity. A field_path of media.media replaces
        // only the media list, leaving every other component on the entity untouched.
        await client.entities.overrideEntity({
            entityId,
            fieldPath: "media.media",
            entity: {
                entityId: entityId,
                media: {
                    media: [
                        { itemIdentifier: ingressId, type: "MEDIA_TYPE_VIDEO" }
                    ]
                }
            },
            provenance: {
                integrationName: "lattice-video-integration",
                dataType: "video-association",
                sourceUpdateTime: new Date().toISOString()
            }
        });
        console.log(`Associated video ${ingressId} with entity ${entityId}`);
    } catch (error) {
        console.log(`Encountered the following error while starting the MPEG-TS ingress stream: ${error}`);
    }
}

App();

```

```go title={"Go (gRPC)"}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// Entity ID to associate the stream with. Replace before running.
const entityID = "<ENTITY-ID>"

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}

	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}
	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	videoClient := videomanagerv1grpc.NewVideoManagerAPIClient(conn)
	emClient := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
	ingressID := uuid.NewString()
	fmt.Printf("Starting MPEG-TS ingress stream %s\n", ingressID)

	// An empty MpegTsSettings tells Lattice to allocate a UDP push endpoint; the URL comes back
	// on the response's mpeg_ts details. MPEG-TS ingress may be disabled per deployment (for
	// example, in the cloud); such a request is rejected, so be ready to fall back to RTSP or SRT.
	createResp, err := videoClient.CreateIngressStream(ctx, &videomanagerv1.CreateIngressStreamRequest{
		IngressId: &ingressID,
		Title:     "cam-01",
		Ingress: &videomanagerv1.CreateIngressStreamRequest_MpegTs{
			MpegTs: &videomanagerv1.MpegTsSettings{},
		},
	})
	if err != nil {
		log.Fatalf("Failed to create ingress stream: %v", err)
	}
	fmt.Printf("Successfully started stream with ID: %s\n", createResp.GetIngressId())

	mpegTs := createResp.GetMpegTs()
	if mpegTs == nil {
		log.Fatalf("Server did not return an MPEG-TS ingress URL")
	}
	fmt.Printf("Push MPEG-TS to: %s\n", mpegTs.GetUrl())

	// Attach the new video to the asset entity. A field_path of media.media replaces
	// only the media list, leaving every other component on the entity untouched.
	_, err = emClient.OverrideEntity(ctx, &entitymanagerv1.OverrideEntityRequest{
		Entity: &entitymanagerv1.Entity{
			EntityId: entityID,
			Media: &entitymanagerv1.Media{
				Media: []*entitymanagerv1.MediaItem{{
					ItemIdentifier: ingressID,
					Type:           entitymanagerv1.MediaType_MEDIA_TYPE_VIDEO,
				}},
			},
		},
		FieldPath: []string{"media.media"},
		Provenance: &entitymanagerv1.Provenance{
			IntegrationName:  "lattice-video-integration",
			DataType:         "video-association",
			SourceUpdateTime: timestamppb.Now(),
		},
	})
	if err != nil {
		log.Fatalf("Failed to override entity media: %v", err)
	}
	fmt.Printf("Associated video %s with entity %s\n", ingressID, entityID)
}

```

```py title={"Python (gRPC)"}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
import uuid
from datetime import datetime, timezone
from bearer_auth import create_client
from protobuf import Oneof
from protobuf.wkt import Timestamp

from anduril.entitymanager.v1.entity_pub_pb import Entity, Provenance
from anduril.entitymanager.v1.entity_manager_api_pub_pb import OverrideEntityRequest
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync
from anduril.entitymanager.v1.media_pub_pb import Media, MediaItem, MediaType

from anduril.videomanager.v1.create_ingress_stream_request_pub_pb import (
    CreateIngressStreamRequest,
)
from anduril.videomanager.v1.mpeg_ts_settings_pub_pb import MpegTsSettings
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Entity ID to associate the stream with. Replace before running.
ENTITY_ID = "<ENTITY-ID>"


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API clients over Connect gRPC with authentication
    video_client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)
    em_client = create_client(EntityManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    ingress_id = str(uuid.uuid4())
    print(f"Starting MPEG-TS ingress stream {ingress_id}")

    # An empty MpegTsSettings tells Lattice to allocate a UDP push endpoint; the URL comes back
    # on the response's mpeg_ts details.
    # MPEG-TS ingress may be disabled per deployment (for example, in the cloud); such a request
    # is rejected, so be ready to fall back to RTSP or SRT.
    try:
        create_resp = video_client.create_ingress_stream(
            CreateIngressStreamRequest(
                ingress_id=ingress_id,
                title="cam-01",
                ingress=Oneof("mpeg_ts", MpegTsSettings()),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to create ingress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully started stream with ID: {create_resp.ingress_id}")

    if not create_resp.has_field("mpeg_ts"):
        print("Server did not return an MPEG-TS ingress URL", file=sys.stderr)
        sys.exit(1)
    print(f"Push MPEG-TS to: {create_resp.ingress.value.url}")

    # Attach the new video to the asset entity. A field_path of media.media replaces
    # only the media list, leaving every other component on the entity untouched.
    now_ts = Timestamp.from_datetime(datetime.now(timezone.utc))

    try:
        em_client.override_entity(
            OverrideEntityRequest(
                entity=Entity(
                    entity_id=ENTITY_ID,
                    media=Media(
                        media=[
                            MediaItem(
                                item_identifier=ingress_id,
                                type=MediaType.VIDEO,
                            ),
                        ],
                    ),
                ),
                field_path=["media.media"],
                provenance=Provenance(
                    integration_name="lattice-video-integration",
                    data_type="video-association",
                    source_update_time=now_ts,
                ),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to override entity media: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Associated video {ingress_id} with entity {ENTITY_ID}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::{
    Entity, Media, MediaItem, MediaType, OverrideEntityRequest, Provenance,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::{
    create_ingress_stream_request::Ingress, create_ingress_stream_response,
    CreateIngressStreamRequest, MpegTsSettings,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use prost_types::Timestamp;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;
use uuid::Uuid;

use std::env;
use std::time::{SystemTime, UNIX_EPOCH};

// Entity ID to associate the stream with. Replace before running.
const ENTITY_ID: &str = "<ENTITY-ID>";

fn now_timestamp() -> Timestamp {
    let d = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
    Timestamp {
        seconds: d.as_secs() as i64,
        nanos: d.subsec_nanos() as i32,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    // Both clients share the same channel and inject the same auth metadata.
    let video_auth = auth_header.clone();
    let video_sandbox = sandbox_header.clone();
    let mut video_client = VideoManagerApiClient::with_interceptor(
        channel.clone(),
        move |mut req: Request<()>| {
            req.metadata_mut().insert("authorization", video_auth.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", video_sandbox.clone());
            Ok(req)
        },
    );
    let mut em_client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    let ingress_id = Uuid::new_v4().to_string();
    println!("Starting MPEG-TS ingress stream {}", ingress_id);

    // An empty MpegTsSettings tells Lattice to allocate a UDP push endpoint; the URL comes back
    // on the response's mpeg_ts details. MPEG-TS ingress may be disabled per deployment (for
    // example, in the cloud); such a request is rejected, so be ready to fall back to RTSP or SRT.
    let create_req = Request::new(CreateIngressStreamRequest {
        ingress_id: Some(ingress_id.clone()),
        title: "cam-01".to_string(),
        ingress: Some(Ingress::MpegTs(MpegTsSettings {})),
    });
    let create_resp = video_client.create_ingress_stream(create_req).await?.into_inner();
    println!("Successfully started stream with ID: {}", create_resp.ingress_id);

    match create_resp.ingress {
        Some(create_ingress_stream_response::Ingress::MpegTs(mpeg_ts)) => {
            println!("Push MPEG-TS to: {}", mpeg_ts.url);
        }
        _ => {
            eprintln!("Server did not return an MPEG-TS ingress URL");
            return Ok(());
        }
    }

    // Attach the new video to the asset entity. A field_path of media.media replaces
    // only the media list, leaving every other component on the entity untouched.
    let override_req = Request::new(OverrideEntityRequest {
        entity: Some(Entity {
            entity_id: ENTITY_ID.to_string(),
            media: Some(Media {
                media: vec![MediaItem {
                    item_identifier: ingress_id.clone(),
                    r#type: MediaType::Video as i32,
                    relative_path: String::new(),
                }],
            }),
            ..Default::default()
        }),
        field_path: vec!["media.media".to_string()],
        provenance: Some(Provenance {
            integration_name: "lattice-video-integration".to_string(),
            data_type: "video-association".to_string(),
            source_update_time: Some(now_timestamp()),
            ..Default::default()
        }),
    });
    match em_client.override_entity(override_req).await {
        Ok(_) => println!("Associated video {} with entity {}", ingress_id, ENTITY_ID),
        Err(e) => eprintln!("Failed to override entity media: {}", e),
    }

    Ok(())
}

```

#### Push the stream

Encode and push the camera with `ffmpeg`.

The following captures from a macOS FaceTime camera using `H.264`. Replace `<LATTICE-MPEG-TS-URL>:<PORT>`
with the endpoint and port returned in `mpeg_ts.url`:

```bash
ffmpeg -hide_banner -loglevel warning \
  -f avfoundation -framerate 30 -video_size 1280x720 -pixel_format uyvy422 -i "0" \
  -c:v h264_videotoolbox -b:v 5M -profile:v main -realtime 1 \
  -f mpegts -flush_packets 0 \
  "<LATTICE-MPEG-TS-URL>:<PORT>?pkt_size=1316"
```

The `pkt_size=1316` query batches seven 188-byte transport stream packets per UDP packet, keeping each one under a
standard 1500-byte Ethernet MTU and avoiding IP fragmentation.

#### SRT

Use SRT when your source pushes video and you want low-latency delivery with optional AES encryption.
Lattice allocates the listener and returns the endpoint your producer pushes to.

Before attaching a video to an asset, the asset must exist as a Lattice entity. Publish a camera entity with the
`TEMPLATE_ASSET` ontology and a `PlatformType` of `CAMERA`, then refresh it on a heartbeat so it does not expire.

#### Publish a camera

Construct an `Entity` with the components Lattice needs to render a friendly land-based camera asset on the common
operational picture (COP). `ExpiryTime` is set five minutes out and paired with a five-second publish loop, so the entity stays alive as long as the publisher runs.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to publish a camera asset entity.
// Attaching a video to an asset requires the asset to already exist as a Lattice entity, so
// publish the camera on a heartbeat before starting an ingress stream.
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"time"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
	"github.com/google/uuid"
)

const (
	cameraName          = "camera-01"
	integrationName     = "lattice_video_integration"
	dataType            = "camera_video_feed"
	publishInterval     = 5 * time.Second
	entityExpiryTtlSecs = 60
)

func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")

	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	// Stable entity ID derived from the camera's hardware identifier so the same
	// physical camera maps to the same Lattice entity across restarts.
	entityId := uuid.NewSHA1(uuid.NameSpaceURL, []byte(cameraName)).String()

	creationTime := time.Now().UTC()

	// Continuously publish the entity so it does not expire.
	for {
		latestTimestamp := time.Now().UTC()
		ctx := context.Background()

		entity := Lattice.Entity{
			EntityID:    &entityId,
			Description: Lattice.String("EO camera streaming live video"),
			Aliases: &Lattice.Aliases{
				Name: Lattice.String("Camera 01"),
				AlternateIDs: []*Lattice.AlternateID{
					{
						ID:   Lattice.String(cameraName),
						Type: Lattice.AlternateIDTypeAltIDTypeSerialNumber.Ptr(),
					},
				},
			},
			IsLive:      Lattice.Bool(true),
			CreatedTime: Lattice.Time(creationTime),
			// ExpiryTime is set in the future and paired with a shorter publish loop,
			// so the entity stays alive as long as the publisher runs.
			ExpiryTime: Lattice.Time(latestTimestamp.Add(entityExpiryTtlSecs * time.Second)),
			Ontology: &Lattice.Ontology{
				Template:     Lattice.OntologyTemplateTemplateAsset.Ptr(),
				PlatformType: Lattice.String("CAMERA"),
			},
			MilView: &Lattice.MilView{
				Disposition: Lattice.MilViewDispositionDispositionFriendly.Ptr(),
				Environment: Lattice.MilViewEnvironmentEnvironmentLand.Ptr(),
				Nationality: Lattice.MilViewNationalityNationalityUnitedStatesOfAmerica.Ptr(),
			},
			// Overall classification of the entity's data.
			DataClassification: &Lattice.Classification{
				Default: &Lattice.ClassificationInformation{
					Level: Lattice.ClassificationInformationLevelClassificationLevelsUnclassified.Ptr(),
				},
			},
			Location: &Lattice.Location{
				Position: &Lattice.Position{
					LatitudeDegrees:   Lattice.Float64(36.66633492530301),
					LongitudeDegrees:  Lattice.Float64(-116.9117674522639),
					AltitudeHaeMeters: Lattice.Float64(780.0),
				},
			},
			Provenance: &Lattice.Provenance{
				IntegrationName:  Lattice.String(integrationName),
				DataType:         Lattice.String(dataType),
				SourceUpdateTime: Lattice.Time(latestTimestamp),
			},
			Health: &Lattice.Health{
				ConnectionStatus: Lattice.HealthConnectionStatusConnectionStatusOnline.Ptr(),
				HealthStatus:     Lattice.HealthHealthStatusHealthStatusHealthy.Ptr(),
				ActiveAlerts:     []*Lattice.Alert{},
				UpdateTime:       Lattice.Time(latestTimestamp),
				// Per-subsystem health rolls up into the entity's overall
				// health_status; report at least the camera sensor itself.
				Components: []*Lattice.ComponentHealth{
					{
						ID:         Lattice.String("camera-01-eo"),
						Name:       Lattice.String("EO camera sensor"),
						Health:     Lattice.ComponentHealthHealthHealthStatusHealthy.Ptr(),
						UpdateTime: Lattice.Time(latestTimestamp),
					},
				},
			},
			// Declare the camera as a sensor so Lattice can render the sensor cone.
			Sensors: &Lattice.Sensors{
				Sensors: []*Lattice.Sensor{
					{
						SensorID:          Lattice.String("camera-01-eo"),
						SensorType:        Lattice.SensorSensorTypeSensorTypeCamera.Ptr(),
						OperationalState:  Lattice.SensorOperationalStateOperationalStateOperational.Ptr(),
						SensorDescription: Lattice.String("EO camera, H.264"),
						FieldsOfView: []*Lattice.FieldOfView{
							{
								CenterRayPose: &Lattice.EntityManagerPose{
									Orientation: &Lattice.Quaternion{
										X: Lattice.Float64(0.0),
										Y: Lattice.Float64(0.0),
										Z: Lattice.Float64(0.0),
										W: Lattice.Float64(1.0),
									},
								},
								HorizontalFov: Lattice.Float64(1.204), // ~69 deg (typical EO horizontal FoV)
								VerticalFov:   Lattice.Float64(0.954), // ~54.6 deg (typical EO vertical FoV)
								Range:         Lattice.Float64(150.0),
								Mode:          Lattice.FieldOfViewModeSensorModeSearch.Ptr(),
							},
						},
					},
				},
			},
			TaskCatalog: &Lattice.TaskCatalog{
				TaskDefinitions: []*Lattice.TaskDefinition{
					{TaskSpecificationURL: Lattice.String("type.googleapis.com/<org>.<package>.<version>.<task>")},
				},
			},
		}

		// Publish the entity
		_, err := LatticeClient.Entities.PublishEntity(ctx, &entity)
		if err != nil {
			fmt.Printf("Error publishing entity: %v\n", err)
		} else {
			fmt.Printf("Publishing asset with entity ID: %s\n", entityId)
		}

		time.Sleep(publishInterval)
	}
}

```

```java title={"Java (REST)"}
package org.example;

import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.List;
import java.util.UUID;

import com.anduril.Lattice;
import com.anduril.types.Alert;
import com.anduril.types.AlternateId;
import com.anduril.types.AlternateIdType;
import com.anduril.types.Aliases;
import com.anduril.types.Classification;
import com.anduril.types.ClassificationInformation;
import com.anduril.types.ClassificationInformationLevel;
import com.anduril.types.ComponentHealth;
import com.anduril.types.ComponentHealthHealth;
import com.anduril.types.Entity;
import com.anduril.types.EntityManagerPose;
import com.anduril.types.FieldOfView;
import com.anduril.types.FieldOfViewMode;
import com.anduril.types.Health;
import com.anduril.types.HealthConnectionStatus;
import com.anduril.types.HealthHealthStatus;
import com.anduril.types.Location;
import com.anduril.types.MilView;
import com.anduril.types.MilViewDisposition;
import com.anduril.types.MilViewEnvironment;
import com.anduril.types.MilViewNationality;
import com.anduril.types.Ontology;
import com.anduril.types.OntologyTemplate;
import com.anduril.types.Position;
import com.anduril.types.Provenance;
import com.anduril.types.Quaternion;
import com.anduril.types.Sensor;
import com.anduril.types.SensorOperationalState;
import com.anduril.types.SensorSensorType;
import com.anduril.types.Sensors;
import com.anduril.types.TaskCatalog;
import com.anduril.types.TaskDefinition;

/**
 * Publishes a camera asset entity with the Lattice SDK. Attaching a video to an asset requires the
 * asset to already exist as a Lattice entity, so publish the camera on a heartbeat before starting
 * an ingress stream.
 */
public class PublishCamera {
    private static final String CAMERA_NAME = "camera-01";
    private static final String INTEGRATION_NAME = "lattice_video_integration";
    private static final String DATA_TYPE = "camera_video_feed";
    private static final long PUBLISH_INTERVAL_MS = 5000;
    private static final long ENTITY_EXPIRY_TTL_SEC = 60;

    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        // Check required variables and ensure URL has https://
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Stable entity ID derived from the camera's hardware identifier so the same
            // physical camera maps to the same Lattice entity across restarts.
            String entityId = UUID.nameUUIDFromBytes(CAMERA_NAME.getBytes()).toString();
            Instant creationTime = Instant.now();

            // Continuously publish the camera so it does not expire while the publisher runs.
            while (true) {
                try {
                    Instant latestTimestamp = Instant.now();

                    // Construct an Entity with the components Lattice needs to render a friendly
                    // land-based camera asset on the common operational picture (COP).
                    Entity entity = Entity.builder()
                        .entityId(entityId)
                        .description("EO camera streaming live video")
                        .aliases(Aliases.builder()
                            .name("Camera 01")
                            .alternateIds(List.of(AlternateId.builder()
                                .id(CAMERA_NAME)
                                .type(AlternateIdType.ALT_ID_TYPE_SERIAL_NUMBER)
                                .build()))
                            .build())
                        .isLive(true)
                        .createdTime(OffsetDateTime.ofInstant(creationTime, ZoneOffset.UTC))
                        // ExpiryTime is set in the future and paired with a shorter publish loop,
                        // so the entity stays alive as long as the publisher runs.
                        .expiryTime(OffsetDateTime.ofInstant(
                            latestTimestamp.plus(ENTITY_EXPIRY_TTL_SEC, ChronoUnit.SECONDS), ZoneOffset.UTC))
                        .ontology(Ontology.builder()
                            .template(OntologyTemplate.TEMPLATE_ASSET)
                            .platformType("CAMERA")
                            .build())
                        .milView(MilView.builder()
                            .disposition(MilViewDisposition.DISPOSITION_FRIENDLY)
                            .environment(MilViewEnvironment.ENVIRONMENT_LAND)
                            .nationality(MilViewNationality.NATIONALITY_UNITED_STATES_OF_AMERICA)
                            .build())
                        // Overall classification of the entity's data.
                        .dataClassification(Classification.builder()
                            .default_(ClassificationInformation.builder()
                                .level(ClassificationInformationLevel.CLASSIFICATION_LEVELS_UNCLASSIFIED)
                                .build())
                            .build())
                        .location(Location.builder()
                            .position(Position.builder()
                                .latitudeDegrees(36.66633492530301)
                                .longitudeDegrees(-116.9117674522639)
                                .altitudeHaeMeters(780.0)
                                .build())
                            .build())
                        .provenance(Provenance.builder()
                            .integrationName(INTEGRATION_NAME)
                            .dataType(DATA_TYPE)
                            .sourceUpdateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                            .build())
                        .health(Health.builder()
                            .connectionStatus(HealthConnectionStatus.CONNECTION_STATUS_ONLINE)
                            .healthStatus(HealthHealthStatus.HEALTH_STATUS_HEALTHY)
                            .activeAlerts(Collections.<Alert>emptyList())
                            .updateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                            // Per-subsystem health rolls up into the entity's overall
                            // health_status; report at least the camera sensor itself.
                            .components(List.of(ComponentHealth.builder()
                                .id("camera-01-eo")
                                .name("EO camera sensor")
                                .health(ComponentHealthHealth.HEALTH_STATUS_HEALTHY)
                                .updateTime(OffsetDateTime.ofInstant(latestTimestamp, ZoneOffset.UTC))
                                .build()))
                            .build())
                        // Declare the camera as a sensor so Lattice can render the sensor cone.
                        .sensors(Sensors.builder()
                            .sensors(List.of(Sensor.builder()
                                .sensorId("camera-01-eo")
                                .sensorType(SensorSensorType.SENSOR_TYPE_CAMERA)
                                .operationalState(SensorOperationalState.OPERATIONAL_STATE_OPERATIONAL)
                                .sensorDescription("EO camera, H.264")
                                .fieldsOfView(List.of(FieldOfView.builder()
                                    .centerRayPose(EntityManagerPose.builder()
                                        .orientation(Quaternion.builder()
                                            .x(0.0)
                                            .y(0.0)
                                            .z(0.0)
                                            .w(1.0)
                                            .build())
                                        .build())
                                    .horizontalFov(1.204f) // ~69 deg (typical EO horizontal FoV)
                                    .verticalFov(0.954f) // ~54.6 deg (typical EO vertical FoV)
                                    .range(150.0f)
                                    .mode(FieldOfViewMode.SENSOR_MODE_SEARCH)
                                    .build()))
                                .build()))
                            .build())
                        .taskCatalog(TaskCatalog.builder()
                            .taskDefinitions(List.of(TaskDefinition.builder()
                                .taskSpecificationUrl("type.googleapis.com/<org>.<package>.<version>.<task>")
                                .build()))
                            .build())
                        .build();

                    // Publish the entity
                    client.entities().publishEntity(entity);
                    System.out.println("Publishing asset with entity ID: " + entityId);

                    Thread.sleep(PUBLISH_INTERVAL_MS);
                } catch (Exception e) {
                    System.err.println("Error publishing entity: " + e.getMessage());
                    Thread.sleep(PUBLISH_INTERVAL_MS);
                }
            }
        } catch (Exception e) {
            System.err.println("Failed to initialize: " + e.getMessage());
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the public Lattice REST SDK (anduril-lattice-sdk) to publish a
# camera asset entity. Attaching a video to an asset requires the asset to already exist as a
# Lattice entity, so publish the camera on a heartbeat before starting an ingress stream.

import os
import sys
import time
import uuid
from datetime import datetime, timedelta, timezone

from anduril import (
    Aliases,
    AlternateId,
    Classification,
    ClassificationInformation,
    ComponentHealth,
    FieldOfView,
    Health,
    Lattice,
    Location,
    MilView,
    Ontology,
    Position,
    Provenance,
    Quaternion,
    Sensor,
    Sensors,
    EntityManagerPose,
    TaskCatalog,
    TaskDefinition,
)

CAMERA_NAME = "camera-01"
INTEGRATION_NAME = "lattice_video_integration"
DATA_TYPE = "camera_video_feed"
PUBLISH_INTERVAL_SEC = 5
ENTITY_EXPIRY_TTL_SEC = 60


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")

    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")
    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables.", file=sys.stderr)
        sys.exit(1)

    client = Lattice(
        base_url=f"https://{lattice_endpoint}",
        token=lambda: str(environment_token),
        # Remove the following header if you are not developing on Sandboxes.
        headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"},
    )

    # Stable entity_id derived from the camera's hardware identifier so the
    # same physical camera maps to the same Lattice entity across restarts.
    entity_id = str(uuid.uuid5(uuid.NAMESPACE_URL, CAMERA_NAME))

    creation_time = datetime.now(timezone.utc)

    try:
        while True:
            latest_timestamp = datetime.now(timezone.utc)

            client.entities.publish_entity(
                entity_id=entity_id,
                description="EO camera streaming live video",
                aliases=Aliases(
                    name="Camera 01",
                    alternate_ids=[
                        AlternateId(id=CAMERA_NAME, type="ALT_ID_TYPE_SERIAL_NUMBER"),
                    ],
                ),
                is_live=True,
                created_time=creation_time,
                # ExpiryTime is set in the future and paired with a shorter publish loop,
                # so the entity stays alive as long as the publisher runs.
                expiry_time=latest_timestamp + timedelta(seconds=ENTITY_EXPIRY_TTL_SEC),
                ontology=Ontology(template="TEMPLATE_ASSET", platform_type="CAMERA"),
                mil_view=MilView(
                    disposition="DISPOSITION_FRIENDLY",
                    environment="ENVIRONMENT_LAND",
                    nationality="NATIONALITY_UNITED_STATES_OF_AMERICA",
                ),
                # Overall classification of the entity's data.
                data_classification=Classification(
                    default=ClassificationInformation(
                        level="CLASSIFICATION_LEVELS_UNCLASSIFIED",
                    ),
                ),
                location=Location(
                    position=Position(
                        latitude_degrees=36.66633492530301,
                        longitude_degrees=-116.9117674522639,
                        altitude_hae_meters=780.0,
                    ),
                ),
                provenance=Provenance(
                    integration_name=INTEGRATION_NAME,
                    data_type=DATA_TYPE,
                    source_update_time=latest_timestamp,
                ),
                health=Health(
                    connection_status="CONNECTION_STATUS_ONLINE",
                    health_status="HEALTH_STATUS_HEALTHY",
                    active_alerts=[],
                    update_time=latest_timestamp,
                    # Per-subsystem health rolls up into the entity's overall
                    # health_status; report at least the camera sensor itself.
                    components=[
                        ComponentHealth(
                            id="camera-01-eo",
                            name="EO camera sensor",
                            health="HEALTH_STATUS_HEALTHY",
                            update_time=latest_timestamp,
                        ),
                    ],
                ),
                # Declare the camera as a sensor so Lattice can render the sensor cone.
                sensors=Sensors(
                    sensors=[
                        Sensor(
                            sensor_id="camera-01-eo",
                            sensor_type="SENSOR_TYPE_CAMERA",
                            operational_state="OPERATIONAL_STATE_OPERATIONAL",
                            sensor_description="EO camera, H.264",
                            fields_of_view=[
                                FieldOfView(
                                    center_ray_pose=EntityManagerPose(
                                        orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0),
                                    ),
                                    horizontal_fov=1.204,  # ~69 deg (typical EO horizontal FoV)
                                    vertical_fov=0.954,  # ~54.6 deg (typical EO vertical FoV)
                                    range=150.0,
                                    mode="SENSOR_MODE_SEARCH",
                                ),
                            ],
                        ),
                    ],
                ),
                task_catalog=TaskCatalog(
                    task_definitions=[
                        TaskDefinition(task_specification_url="type.googleapis.com/<org>.<package>.<version>.<task>"),
                    ],
                ),
            )
            print(f"Publishing asset with entity ID: {entity_id}")
            time.sleep(PUBLISH_INTERVAL_SEC)

    except (KeyboardInterrupt, SystemExit):
        print(">>> Exiting...")
    except Exception as error:
        print(f"Error publishing entity: {error}", file=sys.stderr)


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
import { LatticeClient } from "@anduril-industries/lattice-sdk";
import { v5 as uuidv5 } from "uuid";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;

// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

const cameraName = "camera-01";
const integrationName = "lattice_video_integration";
const dataType = "camera_video_feed";
const publishIntervalMs = 5000;
const entityExpiryTtlSecs = 60;

// Stable entity ID derived from the camera's hardware identifier so the same
// physical camera maps to the same Lattice entity across restarts.
const entityId = uuidv5(cameraName, uuidv5.URL);
const creationTime = new Date().toISOString();

async function App() {
    const latestTimestamp = new Date();
    try {
        // Attaching a video to an asset requires the asset to already exist as a Lattice
        // entity, so publish the camera on a heartbeat before starting an ingress stream.
        await client.entities.publishEntity(
            {
                entityId: entityId,
                description: "EO camera streaming live video",
                aliases: {
                    name: "Camera 01",
                    alternateIds: [
                        { id: cameraName, type: "ALT_ID_TYPE_SERIAL_NUMBER" }
                    ]
                },
                isLive: true,
                createdTime: creationTime,
                // ExpiryTime is set in the future and paired with a shorter publish loop,
                // so the entity stays alive as long as the publisher runs.
                expiryTime: new Date(latestTimestamp.getTime() + entityExpiryTtlSecs * 1000).toISOString(),
                ontology: {
                    template: "TEMPLATE_ASSET",
                    platformType: "CAMERA"
                },
                milView: {
                    disposition: "DISPOSITION_FRIENDLY",
                    environment: "ENVIRONMENT_LAND",
                    nationality: "NATIONALITY_UNITED_STATES_OF_AMERICA"
                },
                // Overall classification of the entity's data.
                dataClassification: {
                    default: {
                        level: "CLASSIFICATION_LEVELS_UNCLASSIFIED"
                    }
                },
                location: {
                    position: {
                        latitudeDegrees: 36.66633492530301,
                        longitudeDegrees: -116.9117674522639,
                        altitudeHaeMeters: 780.0
                    }
                },
                provenance: {
                    integrationName: integrationName,
                    dataType: dataType,
                    sourceUpdateTime: latestTimestamp.toISOString()
                },
                health: {
                    connectionStatus: "CONNECTION_STATUS_ONLINE",
                    healthStatus: "HEALTH_STATUS_HEALTHY",
                    activeAlerts: [],
                    updateTime: latestTimestamp.toISOString(),
                    // Per-subsystem health rolls up into the entity's overall
                    // healthStatus; report at least the camera sensor itself.
                    components: [
                        {
                            id: "camera-01-eo",
                            name: "EO camera sensor",
                            health: "HEALTH_STATUS_HEALTHY",
                            updateTime: latestTimestamp.toISOString()
                        }
                    ]
                },
                // Declare the camera as a sensor so Lattice can render the sensor cone.
                sensors: {
                    sensors: [
                        {
                            sensorId: "camera-01-eo",
                            sensorType: "SENSOR_TYPE_CAMERA",
                            operationalState: "OPERATIONAL_STATE_OPERATIONAL",
                            sensorDescription: "EO camera, H.264",
                            fieldsOfView: [
                                {
                                    centerRayPose: {
                                        orientation: { x: 0.0, y: 0.0, z: 0.0, w: 1.0 }
                                    },
                                    horizontalFov: 1.204, // ~69 deg (typical EO horizontal FoV)
                                    verticalFov: 0.954, // ~54.6 deg (typical EO vertical FoV)
                                    range: 150.0,
                                    mode: "SENSOR_MODE_SEARCH"
                                }
                            ]
                        }
                    ]
                },
                taskCatalog: {
                    taskDefinitions: [
                        { taskSpecificationUrl: "type.googleapis.com/<org>.<package>.<version>.<task>" }
                    ]
                }
            }
        );
        console.log(`Publishing asset with entity ID: ${entityId}`);
    } catch (error) {
        console.log(`Encountered the following error while publishing entity: ${error}`);
    }
}

(async function runIndefinitely() {
    while (true) {
        await App();
        await new Promise(resolve => setTimeout(resolve, publishIntervalMs));
    }
})();

```

```go title={"Go (gRPC)"}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/types/known/timestamppb"
	"google.golang.org/protobuf/types/known/wrapperspb"
	ontologyv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/ontology/v1"
	tasksv2 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/tasks/v2"
	_type "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/type"
)

const (
	cameraName      = "camera-01"
	integrationName = "lattice_video_integration"
	dataType        = "camera_video_feed"
	publishInterval = 5 * time.Second
	entityExpiryTTL = 60 * time.Second
	publishTimeout  = 10 * time.Second
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}

	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}
	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	client := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	// Stable entity_id derived from the camera's hardware identifier so the
	// same physical camera maps to the same Lattice entity across restarts.
	entityId := uuid.NewSHA1(uuid.NameSpaceURL, []byte(cameraName)).String()

	creationTimestamp := timestamppb.New(time.Now().UTC())

	for {
		latestTimestamp := time.Now().UTC()
		ctx, cancel := context.WithTimeout(context.Background(), publishTimeout)

		entity := &entitymanagerv1.Entity{
			EntityId:    entityId,
			Description: "EO camera streaming live video",
			Aliases: &entitymanagerv1.Aliases{
				Name: "Camera 01",
				AlternateIds: []*entitymanagerv1.AlternateId{
					{
						Id:   cameraName,
						Type: entitymanagerv1.AltIdType_ALT_ID_TYPE_SERIAL_NUMBER,
					},
				},
			},
			IsLive:      true,
			CreatedTime: creationTimestamp,
			ExpiryTime:  timestamppb.New(latestTimestamp.Add(entityExpiryTTL)),
			Ontology: &entitymanagerv1.Ontology{
				Template:     entitymanagerv1.Template_TEMPLATE_ASSET,
				PlatformType: "CAMERA",
			},
			MilView: &entitymanagerv1.MilView{
				Disposition: ontologyv1.Disposition_DISPOSITION_FRIENDLY,
				Environment: ontologyv1.Environment_ENVIRONMENT_LAND,
				Nationality: ontologyv1.Nationality_NATIONALITY_UNITED_STATES_OF_AMERICA,
			},
			// Overall classification of the entity's data.
			DataClassification: &entitymanagerv1.Classification{
				Default: &entitymanagerv1.ClassificationInformation{
					Level: entitymanagerv1.ClassificationLevels_CLASSIFICATION_LEVELS_UNCLASSIFIED,
				},
			},
			Location: &entitymanagerv1.Location{
				Position: &entitymanagerv1.Position{
					LatitudeDegrees:   36.66633492530301,
					LongitudeDegrees:  -116.9117674522639,
					AltitudeHaeMeters: wrapperspb.Double(780.0),
				},
			},
			Provenance: &entitymanagerv1.Provenance{
				IntegrationName:  integrationName,
				DataType:         dataType,
				SourceUpdateTime: timestamppb.New(latestTimestamp),
			},
			Health: &entitymanagerv1.Health{
				ConnectionStatus: entitymanagerv1.ConnectionStatus_CONNECTION_STATUS_ONLINE,
				HealthStatus:     entitymanagerv1.HealthStatus_HEALTH_STATUS_HEALTHY,
				ActiveAlerts:     []*entitymanagerv1.Alert{},
				UpdateTime:       timestamppb.New(latestTimestamp),
				// Per-subsystem health rolls up into the entity's overall
				// health_status; report at least the camera sensor itself.
				Components: []*entitymanagerv1.ComponentHealth{
					{
						Id:         "camera-01-eo",
						Name:       "EO camera sensor",
						Health:     entitymanagerv1.HealthStatus_HEALTH_STATUS_HEALTHY,
						UpdateTime: timestamppb.New(latestTimestamp),
					},
				},
			},
			Sensors: &entitymanagerv1.Sensors{
				Sensors: []*entitymanagerv1.Sensor{
					{
						SensorId:          "camera-01-eo",
						SensorType:        entitymanagerv1.SensorType_SENSOR_TYPE_CAMERA,
						OperationalState:  entitymanagerv1.OperationalState_OPERATIONAL_STATE_OPERATIONAL,
						SensorDescription: "EO camera, H.264",
						FieldsOfView: []*entitymanagerv1.FieldOfView{
							{
								CenterRayPose: &entitymanagerv1.Pose{
									Orientation: &_type.Quaternion{
										X: 0.0,
										Y: 0.0,
										Z: 0.0,
										W: 1.0,
									},
								},
								HorizontalFov: 1.204, // ~69 deg (typical EO horizontal FoV)
								VerticalFov:   0.954, // ~54.6 deg (typical EO vertical FoV)
								Range:         wrapperspb.Float(150.0),
								Mode:          entitymanagerv1.SensorMode_SENSOR_MODE_SEARCH,
							},
						},
					},
				},
			},
			TaskCatalog: &tasksv2.TaskCatalog{
				TaskDefinitions: []*tasksv2.TaskDefinition{
					{TaskSpecificationUrl: "type.googleapis.com/<org>.<package>.<version>.<task>"},
				},
			},
		}

		_, err := client.PublishEntity(ctx, &entitymanagerv1.PublishEntityRequest{Entity: entity})
		cancel()

		if err != nil {
			fmt.Printf("Error publishing entity: %v\n", err)
		} else {
			fmt.Printf("Publishing asset with entity ID: %s\n", entityId)
		}

		time.Sleep(publishInterval)
	}
}

```

```py title={"Python (gRPC)"}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
import time
import uuid
from datetime import datetime, timedelta, timezone
from bearer_auth import create_client
from protobuf.wkt import DoubleValue, FloatValue, Timestamp

from anduril.entitymanager.v1.classification_pub_pb import (
    Classification,
    ClassificationInformation,
    ClassificationLevels,
)
from anduril.entitymanager.v1.entity_pub_pb import (
    Aliases,
    AlternateId,
    Entity,
    Provenance,
)
from anduril.entitymanager.v1.entity_manager_api_pub_pb import PublishEntityRequest
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync
from anduril.entitymanager.v1.health_status_pub_pb import (
    ComponentHealth,
    ConnectionStatus,
    Health,
    HealthStatus,
)
from anduril.entitymanager.v1.location_pub_pb import Location, Pose, Position
from anduril.entitymanager.v1.ontology_pub_pb import MilView, Ontology
from anduril.entitymanager.v1.sensors_pub_pb import (
    FieldOfView,
    OperationalState,
    Sensor,
    SensorMode,
    Sensors,
    SensorType,
)
from anduril.entitymanager.v1.types_pub_pb import AltIdType, Template
from anduril.ontology.v1.type_pub_pb import Disposition, Environment, Nationality
from anduril.tasks.v2.catalog_pub_pb import TaskCatalog, TaskDefinition
from anduril.type.coords_pub_pb import Quaternion


CAMERA_NAME = "camera-01"
INTEGRATION_NAME = "lattice_video_integration"
DATA_TYPE = "camera_video_feed"
PUBLISH_INTERVAL_SEC = 5
ENTITY_EXPIRY_TTL_SEC = 60
PUBLISH_TIMEOUT_SEC = 10


def to_timestamp(dt: datetime) -> Timestamp:
    return Timestamp.from_datetime(dt)


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(EntityManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Stable entity_id derived from the camera's hardware identifier so the
    # same physical camera maps to the same Lattice entity across restarts.
    entity_id = str(uuid.uuid5(uuid.NAMESPACE_URL, CAMERA_NAME))

    creation_timestamp = to_timestamp(datetime.now(timezone.utc))

    while True:
        latest = datetime.now(timezone.utc)
        latest_ts = to_timestamp(latest)
        expiry_ts = to_timestamp(latest + timedelta(seconds=ENTITY_EXPIRY_TTL_SEC))

        entity = Entity(
            entity_id=entity_id,
            description="EO camera streaming live video",
            aliases=Aliases(
                name="Camera 01",
                alternate_ids=[
                    AlternateId(id=CAMERA_NAME, type=AltIdType.SERIAL_NUMBER),
                ],
            ),
            is_live=True,
            created_time=creation_timestamp,
            expiry_time=expiry_ts,
            ontology=Ontology(template=Template.ASSET, platform_type="CAMERA"),
            mil_view=MilView(
                disposition=Disposition.FRIENDLY,
                environment=Environment.LAND,
                nationality=Nationality.UNITED_STATES_OF_AMERICA,
            ),
            # Overall classification of the entity's data.
            data_classification=Classification(
                default=ClassificationInformation(
                    level=ClassificationLevels.UNCLASSIFIED,
                ),
            ),
            location=Location(
                position=Position(
                    latitude_degrees=36.66633492530301,
                    longitude_degrees=-116.9117674522639,
                    altitude_hae_meters=DoubleValue(value=780.0),
                ),
            ),
            provenance=Provenance(
                integration_name=INTEGRATION_NAME,
                data_type=DATA_TYPE,
                source_update_time=latest_ts,
            ),
            health=Health(
                connection_status=ConnectionStatus.ONLINE,
                health_status=HealthStatus.HEALTHY,
                active_alerts=[],
                update_time=latest_ts,
                # Per-subsystem health rolls up into the entity's overall
                # health_status; report at least the camera sensor itself.
                components=[
                    ComponentHealth(
                        id="camera-01-eo",
                        name="EO camera sensor",
                        health=HealthStatus.HEALTHY,
                        update_time=latest_ts,
                    ),
                ],
            ),
            sensors=Sensors(
                sensors=[
                    Sensor(
                        sensor_id="camera-01-eo",
                        sensor_type=SensorType.CAMERA,
                        operational_state=OperationalState.OPERATIONAL,
                        sensor_description="EO camera, H.264",
                        fields_of_view=[
                            FieldOfView(
                                center_ray_pose=Pose(
                                    orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0),
                                ),
                                horizontal_fov=1.204,  # ~69 deg (typical EO horizontal FoV)
                                vertical_fov=0.954,  # ~54.6 deg (typical EO vertical FoV)
                                range=FloatValue(value=150.0),
                                mode=SensorMode.SEARCH,
                            ),
                        ],
                    ),
                ],
            ),
            task_catalog=TaskCatalog(
                task_definitions=[
                    TaskDefinition(task_specification_url="type.googleapis.com/<org>.<package>.<version>.<task>"),
                ],
            ),
        )

        try:
            client.publish_entity(
                PublishEntityRequest(entity=entity),
                timeout_ms=PUBLISH_TIMEOUT_SEC * 1000,
            )
            print(f"Publishing asset with entity ID: {entity_id}")
        except Exception as error:
            print(f"Error publishing entity: {error}", file=sys.stderr)

        time.sleep(PUBLISH_INTERVAL_SEC)


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::{
    Aliases, AltIdType, AlternateId, Classification, ClassificationInformation,
    ClassificationLevels, ComponentHealth, ConnectionStatus, Entity, FieldOfView, Health,
    HealthStatus, Location, MilView, Ontology, OperationalState, Pose, Position,
    Provenance, PublishEntityRequest, Sensor, SensorMode, SensorType, Sensors, Template,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::ontology::v1::{
    Disposition, Environment, Nationality,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::r#type::Quaternion;
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::tasks::v2::{
    TaskCatalog, TaskDefinition,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use prost_types::Timestamp;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;
use uuid::Uuid;

use std::env;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

const CAMERA_NAME: &str = "camera-01";
const INTEGRATION_NAME: &str = "lattice_video_integration";
const DATA_TYPE: &str = "camera_video_feed";
const PUBLISH_INTERVAL_SECS: u64 = 5;
const ENTITY_EXPIRY_TTL_SECS: i64 = 60;

fn now_timestamp() -> Timestamp {
    let d = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
    Timestamp {
        seconds: d.as_secs() as i64,
        nanos: d.subsec_nanos() as i32,
    }
}

fn add_seconds(ts: &Timestamp, seconds: i64) -> Timestamp {
    Timestamp {
        seconds: ts.seconds + seconds,
        nanos: ts.nanos,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Stable entity_id derived from the camera's hardware identifier so the
    // same physical camera maps to the same Lattice entity across restarts.
    let entity_id = Uuid::new_v5(&Uuid::NAMESPACE_URL, CAMERA_NAME.as_bytes()).to_string();
    let creation_timestamp = now_timestamp();

    // Refresh the entity on a heartbeat so it does not expire while the publisher runs.
    loop {
        let latest_timestamp = now_timestamp();

        let entity = Entity {
            entity_id: entity_id.clone(),
            description: "EO camera streaming live video".to_string(),
            aliases: Some(Aliases {
                name: "Camera 01".to_string(),
                alternate_ids: vec![AlternateId {
                    id: CAMERA_NAME.to_string(),
                    r#type: AltIdType::SerialNumber as i32,
                }],
            }),
            is_live: true,
            created_time: Some(creation_timestamp.clone()),
            // ExpiryTime is set in the future and paired with a shorter publish loop,
            // so the entity stays alive as long as the publisher runs.
            expiry_time: Some(add_seconds(&latest_timestamp, ENTITY_EXPIRY_TTL_SECS)),
            ontology: Some(Ontology {
                template: Template::Asset as i32,
                platform_type: "CAMERA".to_string(),
                ..Default::default()
            }),
            mil_view: Some(MilView {
                disposition: Disposition::Friendly as i32,
                environment: Environment::Land as i32,
                nationality: Nationality::UnitedStatesOfAmerica as i32,
            }),
            // Overall classification of the entity's data.
            data_classification: Some(Classification {
                default: Some(ClassificationInformation {
                    level: ClassificationLevels::Unclassified as i32,
                    ..Default::default()
                }),
                ..Default::default()
            }),
            location: Some(Location {
                position: Some(Position {
                    latitude_degrees: 36.66633492530301,
                    longitude_degrees: -116.9117674522639,
                    altitude_hae_meters: Some(780.0),
                    ..Default::default()
                }),
                ..Default::default()
            }),
            provenance: Some(Provenance {
                integration_name: INTEGRATION_NAME.to_string(),
                data_type: DATA_TYPE.to_string(),
                source_update_time: Some(latest_timestamp.clone()),
                ..Default::default()
            }),
            health: Some(Health {
                connection_status: ConnectionStatus::Online as i32,
                health_status: HealthStatus::Healthy as i32,
                active_alerts: vec![],
                update_time: Some(latest_timestamp.clone()),
                // Per-subsystem health rolls up into the entity's overall
                // health_status; report at least the camera sensor itself.
                components: vec![ComponentHealth {
                    id: "camera-01-eo".to_string(),
                    name: "EO camera sensor".to_string(),
                    health: HealthStatus::Healthy as i32,
                    update_time: Some(latest_timestamp.clone()),
                    ..Default::default()
                }],
            }),
            // Declare the camera as a sensor so Lattice can render the sensor cone.
            sensors: Some(Sensors {
                sensors: vec![Sensor {
                    sensor_id: "camera-01-eo".to_string(),
                    sensor_type: SensorType::Camera as i32,
                    operational_state: OperationalState::Operational as i32,
                    sensor_description: "EO camera, H.264".to_string(),
                    fields_of_view: vec![FieldOfView {
                        center_ray_pose: Some(Pose {
                            orientation: Some(Quaternion {
                                x: 0.0,
                                y: 0.0,
                                z: 0.0,
                                w: 1.0,
                            }),
                            ..Default::default()
                        }),
                        horizontal_fov: 1.204, // ~69 deg (typical EO horizontal FoV)
                        vertical_fov: 0.954,   // ~54.6 deg (typical EO vertical FoV)
                        range: Some(150.0),
                        mode: SensorMode::Search as i32,
                        ..Default::default()
                    }],
                    ..Default::default()
                }],
            }),
            task_catalog: Some(TaskCatalog {
                task_definitions: vec![TaskDefinition {
                    task_specification_url: "type.googleapis.com/<org>.<package>.<version>.<task>"
                        .to_string(),
                }],
            }),
            ..Default::default()
        };

        let request = Request::new(PublishEntityRequest {
            entity: Some(entity),
        });
        match client.publish_entity(request).await {
            Ok(_) => println!("Publishing asset with entity ID: {}", entity_id),
            Err(e) => eprintln!("Error publishing entity: {}", e),
        }

        tokio::time::sleep(Duration::from_secs(PUBLISH_INTERVAL_SECS)).await;
    }
}

```

> **Info**
>
> The heartbeat loop is the simplest pattern for keeping a transient asset alive. For long-running deployments,
> consider publishing from a supervised service that resets `ExpiryTime` based on hardware health rather than a fixed interval.

#### Allocate a stream endpoint

Send an `SrtSettings` in the request. To encrypt the stream, set an optional `passphrase`; leave it empty for an
unencrypted stream. The response contains an `SrtIngress` with a `url` and a `session_id` that the producer must
use on the connection. The example also calls `OverrideEntity` to attach the new video to the camera asset:

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to start an SRT ingress stream and
// attach it to a camera asset entity.
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"strings"
	"time"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
	"github.com/google/uuid"
)

const (
	// Entity ID to associate the stream with. Replace before running.
	entityID = "<ENTITY-ID>"
	// Optional SRT passphrase; applies AES encryption to the stream when non-empty.
	passphrase = ""
)

func main() {
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")

	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	ctx := context.Background()

	// Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
	ingressID := uuid.New().String()
	fmt.Printf("Starting SRT ingress stream %s\n", ingressID)

	// SRT is a push protocol: Lattice allocates a listener and returns a URL and
	// session ID the producer must use. An optional passphrase applies AES encryption.
	srtSettings := &Lattice.SrtSettings{}
	if passphrase != "" {
		srtSettings.Passphrase = Lattice.String(passphrase)
	}

	createResp, err := LatticeClient.Video.CreateIngressStream(ctx, &Lattice.CreateIngressStreamRequest{
		IngressID: &ingressID,
		Title:     Lattice.String("cam-01"),
		Srt:       srtSettings,
	})
	if err != nil {
		fmt.Printf("Failed to create ingress stream: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Successfully started stream with ID: %s\n", *createResp.GetIngressID())

	if createResp.GetSrt() == nil || createResp.GetSrt().GetURL() == nil {
		fmt.Println("Server did not return an SRT ingress URL")
		os.Exit(1)
	}
	// The streamid begins with '#', which a URL parser (e.g. ffmpeg's) treats
	// as a fragment and strips. Percent-encode it so the full streamid survives.
	pushURL := strings.ReplaceAll(*createResp.GetSrt().GetURL(), "#", "%23")
	fmt.Printf("Push SRT to: %s\n", pushURL)
	fmt.Printf("Session ID: %s\n", *createResp.GetSrt().GetSessionID())
	if passphrase != "" {
		fmt.Println("Encryption: AES (supply the passphrase on the connection)")
	}

	// Attach the new video to the asset entity. A field_path of media.media replaces
	// only the media list, leaving every other component on the entity untouched.
	_, err = LatticeClient.Entities.OverrideEntity(ctx, &Lattice.EntityOverride{
		EntityID:  entityID,
		FieldPath: "media.media",
		Entity: &Lattice.Entity{
			EntityID: Lattice.String(entityID),
			Media: &Lattice.Media{
				Media: []*Lattice.MediaItem{
					{
						ItemIdentifier: &ingressID,
						Type:           Lattice.MediaItemTypeMediaTypeVideo.Ptr(),
					},
				},
			},
		},
		Provenance: &Lattice.Provenance{
			IntegrationName:  Lattice.String("lattice-video-integration"),
			DataType:         Lattice.String("video-association"),
			SourceUpdateTime: Lattice.Time(time.Now().UTC()),
		},
	})
	if err != nil {
		fmt.Printf("Failed to override entity media: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Associated video %s with entity %s\n", ingressID, entityID)
}

```

```java title={"Java (REST)"}
package org.example;

import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.Collections;
import java.util.UUID;

import com.anduril.Lattice;
import com.anduril.resources.entities.requests.EntityOverride;
import com.anduril.resources.video.requests.CreateIngressStreamRequest;
import com.anduril.types.CreateIngressStreamResponse;
import com.anduril.types.Entity;
import com.anduril.types.Media;
import com.anduril.types.MediaItem;
import com.anduril.types.MediaItemType;
import com.anduril.types.Provenance;
import com.anduril.types.SrtIngress;
import com.anduril.types.SrtSettings;

/**
 * Starts an SRT ingress stream with the Lattice SDK and attaches it to a camera asset entity.
 */
public class CreateIngressStreamSrt {
    // Entity ID to associate the stream with. Replace before running.
    private static final String ENTITY_ID = "<ENTITY-ID>";
    // Optional SRT passphrase; applies AES encryption to the stream when non-empty.
    private static final String PASSPHRASE = "";

    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        // Check required variables and ensure URL has https://
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
            String ingressId = UUID.randomUUID().toString();
            System.out.println("Starting SRT ingress stream " + ingressId);

            // SRT is a push protocol: Lattice allocates a listener and returns a URL and
            // session ID the producer must use. An optional passphrase applies AES encryption.
            SrtSettings.Builder srtSettings = SrtSettings.builder();
            if (!PASSPHRASE.isEmpty()) {
                srtSettings.passphrase(PASSPHRASE);
            }
            CreateIngressStreamResponse response = client.video().createIngressStream(
                CreateIngressStreamRequest.builder()
                    .ingressId(ingressId)
                    .title("cam-01")
                    .srt(srtSettings.build())
                    .build());
            System.out.println("Successfully started stream with ID: " + response.getIngressId().orElse(ingressId));

            SrtIngress srt = response.getSrt().orElse(null);
            if (srt == null || srt.getUrl().isEmpty()) {
                System.err.println("Server did not return an SRT ingress URL.");
                System.exit(1);
            }
            // The streamid begins with '#', which a URL parser (e.g. ffmpeg's) treats
            // as a fragment and strips. Percent-encode it so the full streamid survives.
            String pushUrl = srt.getUrl().orElse("").replace("#", "%23");
            System.out.println("Push SRT to: " + pushUrl);
            System.out.println("Session ID: " + srt.getSessionId().orElse(""));
            if (!PASSPHRASE.isEmpty()) {
                System.out.println("Encryption: AES (supply the passphrase on the connection)");
            }

            // Attach the new video to the asset entity. A field_path of media.media replaces
            // only the media list, leaving every other component on the entity untouched.
            MediaItem mediaItem = MediaItem.builder()
                .itemIdentifier(ingressId)
                .type(MediaItemType.MEDIA_TYPE_VIDEO)
                .build();
            EntityOverride entityOverride = EntityOverride.builder()
                .entity(Entity.builder()
                    .entityId(ENTITY_ID)
                    .media(Media.builder()
                        .media(Collections.singletonList(mediaItem))
                        .build())
                    .build())
                .provenance(Provenance.builder()
                    .integrationName("lattice-video-integration")
                    .dataType("video-association")
                    .sourceUpdateTime(OffsetDateTime.now(ZoneOffset.UTC))
                    .build())
                .build();
            client.entities().overrideEntity(ENTITY_ID, "media.media", entityOverride);

            System.out.println("Associated video " + ingressId + " with entity " + ENTITY_ID);
        } catch (Exception e) {
            System.err.println("Exception: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to start an SRT ingress
# stream and attach it to a camera asset entity.

import os
import sys
import uuid
from datetime import datetime, timezone

from anduril import Entity, Lattice, Media, MediaItem, Provenance, SrtSettings

# Entity ID to associate the stream with. Replace before running.
ENTITY_ID = "<ENTITY-ID>"
# Optional SRT passphrase; applies AES encryption to the stream when non-empty.
PASSPHRASE = ""


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")

    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")
    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables.", file=sys.stderr)
        sys.exit(1)

    client = Lattice(
        base_url=f"https://{lattice_endpoint}",
        token=lambda: str(environment_token),
        # Remove the following header if you are not developing on Sandboxes.
        headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"},
    )

    # Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    ingress_id = str(uuid.uuid4())
    print(f"Starting SRT ingress stream {ingress_id}")

    # SRT is a push protocol: Lattice allocates a listener and returns a URL and
    # session ID the producer must use. An optional passphrase applies AES encryption.
    srt_settings = SrtSettings(passphrase=PASSPHRASE) if PASSPHRASE else SrtSettings()

    try:
        create_resp = client.video.create_ingress_stream(
            ingress_id=ingress_id,
            title="cam-01",
            srt=srt_settings,
        )
    except Exception as error:
        print(f"Failed to create ingress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully started stream with ID: {create_resp.ingress_id}")

    if not create_resp.srt or not create_resp.srt.url:
        print("Server did not return an SRT ingress URL.", file=sys.stderr)
        sys.exit(1)
    # The streamid begins with '#', which a URL parser (e.g. ffmpeg's) treats
    # as a fragment and strips. Percent-encode it so the full streamid survives.
    push_url = create_resp.srt.url.replace("#", "%23")
    print(f"Push SRT to: {push_url}")
    print(f"Session ID: {create_resp.srt.session_id}")
    if PASSPHRASE:
        print("Encryption: AES (supply the passphrase on the connection)")

    # Attach the new video to the asset entity. A field_path of media.media replaces
    # only the media list, leaving every other component on the entity untouched.
    try:
        client.entities.override_entity(
            entity_id=ENTITY_ID,
            field_path="media.media",
            entity=Entity(
                entity_id=ENTITY_ID,
                media=Media(
                    media=[
                        MediaItem(
                            item_identifier=ingress_id,
                            type="MEDIA_TYPE_VIDEO",
                        ),
                    ],
                ),
            ),
            provenance=Provenance(
                integration_name="lattice-video-integration",
                data_type="video-association",
                source_update_time=datetime.now(timezone.utc),
            ),
        )
    except Exception as error:
        print(f"Failed to override entity media: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Associated video {ingress_id} with entity {ENTITY_ID}")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
import { LatticeClient } from "@anduril-industries/lattice-sdk";
import { v4 as uuidv4 } from "uuid";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;

// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Entity ID to associate the stream with. Replace before running.
const entityId = "<ENTITY-ID>";
// Optional SRT passphrase; applies AES encryption to the stream when non-empty.
const passphrase = "";

async function App() {
    // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    const ingressId = uuidv4();
    console.log(`Starting SRT ingress stream ${ingressId}`);

    try {
        // SRT is a push protocol: Lattice allocates a listener and returns a URL and session ID
        // the producer must use. An optional passphrase applies AES encryption.
        const createResp = await client.video.createIngressStream({
            ingressId: ingressId,
            title: "cam-01",
            srt: passphrase ? { passphrase } : {}
        });
        console.log(`Successfully started stream with ID: ${createResp.ingressId}`);

        if (!createResp.srt?.url) {
            console.log("Server did not return an SRT ingress URL.");
            return;
        }
        // The streamid begins with '#', which a URL parser (e.g. ffmpeg's) treats
        // as a fragment and strips. Percent-encode it so the full streamid survives.
        const pushUrl = createResp.srt.url.replace("#", "%23");
        console.log(`Push SRT to: ${pushUrl}`);
        console.log(`Session ID: ${createResp.srt.sessionId}`);
        if (passphrase) {
            console.log("Encryption: AES (supply the passphrase on the connection)");
        }

        // Attach the new video to the asset entity. A field_path of media.media replaces
        // only the media list, leaving every other component on the entity untouched.
        await client.entities.overrideEntity({
            entityId,
            fieldPath: "media.media",
            entity: {
                entityId: entityId,
                media: {
                    media: [
                        { itemIdentifier: ingressId, type: "MEDIA_TYPE_VIDEO" }
                    ]
                }
            },
            provenance: {
                integrationName: "lattice-video-integration",
                dataType: "video-association",
                sourceUpdateTime: new Date().toISOString()
            }
        });
        console.log(`Associated video ${ingressId} with entity ${entityId}`);
    } catch (error) {
        console.log(`Encountered the following error while starting the SRT ingress stream: ${error}`);
    }
}

App();

```

```go title={"Go (gRPC)"}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const (
	// Entity ID to associate the stream with. Replace before running.
	entityID = "<ENTITY-ID>"
	// Optional SRT passphrase; applies AES encryption to the stream when non-empty.
	passphrase = ""
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}

	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}
	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	videoClient := videomanagerv1grpc.NewVideoManagerAPIClient(conn)
	emClient := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
	ingressID := uuid.NewString()
	fmt.Printf("Starting SRT ingress stream %s\n", ingressID)

	// SRT is a push protocol: Lattice allocates a listener and returns a URL and
	// session ID the producer must use. An optional passphrase applies AES encryption.
	srtSettings := &videomanagerv1.SrtSettings{}
	if passphrase != "" {
		srtSettings.Passphrase = proto.String(passphrase)
	}

	createResp, err := videoClient.CreateIngressStream(ctx, &videomanagerv1.CreateIngressStreamRequest{
		IngressId: &ingressID,
		Title:     "cam-01",
		Ingress: &videomanagerv1.CreateIngressStreamRequest_Srt{
			Srt: srtSettings,
		},
	})
	if err != nil {
		log.Fatalf("Failed to create ingress stream: %v", err)
	}
	fmt.Printf("Successfully started stream with ID: %s\n", createResp.GetIngressId())

	srt := createResp.GetSrt()
	if srt == nil {
		log.Fatalf("Server did not return an SRT ingress URL")
	}
	// The streamid begins with '#', which a URL parser (e.g. ffmpeg's) treats
	// as a fragment and strips. Percent-encode it so the full streamid survives.
	pushURL := strings.ReplaceAll(srt.GetUrl(), "#", "%23")
	fmt.Printf("Push SRT to: %s\n", pushURL)
	fmt.Printf("Session ID: %s\n", srt.GetSessionId())
	if passphrase != "" {
		fmt.Println("Encryption: AES (supply the passphrase on the connection)")
	}

	// Attach the new video to the asset entity. A field_path of media.media replaces
	// only the media list, leaving every other component on the entity untouched.
	_, err = emClient.OverrideEntity(ctx, &entitymanagerv1.OverrideEntityRequest{
		Entity: &entitymanagerv1.Entity{
			EntityId: entityID,
			Media: &entitymanagerv1.Media{
				Media: []*entitymanagerv1.MediaItem{{
					ItemIdentifier: ingressID,
					Type:           entitymanagerv1.MediaType_MEDIA_TYPE_VIDEO,
				}},
			},
		},
		FieldPath: []string{"media.media"},
		Provenance: &entitymanagerv1.Provenance{
			IntegrationName:  "lattice-video-integration",
			DataType:         "video-association",
			SourceUpdateTime: timestamppb.Now(),
		},
	})
	if err != nil {
		log.Fatalf("Failed to override entity media: %v", err)
	}
	fmt.Printf("Associated video %s with entity %s\n", ingressID, entityID)
}

```

```py title={"Python (gRPC)"}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
import uuid
from datetime import datetime, timezone
from bearer_auth import create_client
from protobuf import Oneof
from protobuf.wkt import Timestamp

from anduril.entitymanager.v1.entity_pub_pb import Entity, Provenance
from anduril.entitymanager.v1.entity_manager_api_pub_pb import OverrideEntityRequest
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync
from anduril.entitymanager.v1.media_pub_pb import Media, MediaItem, MediaType

from anduril.videomanager.v1.create_ingress_stream_request_pub_pb import (
    CreateIngressStreamRequest,
)
from anduril.videomanager.v1.srt_settings_pub_pb import SrtSettings
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Entity ID to associate the stream with. Replace before running.
ENTITY_ID = "<ENTITY-ID>"
# Optional SRT passphrase; applies AES encryption to the stream when non-empty.
PASSPHRASE = ""


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API clients over Connect gRPC with authentication
    video_client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)
    em_client = create_client(EntityManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    ingress_id = str(uuid.uuid4())
    print(f"Starting SRT ingress stream {ingress_id}")

    # SRT is a push protocol: Lattice allocates a listener and returns a URL and
    # session ID the producer must use. An optional passphrase applies AES encryption.
    srt_settings = SrtSettings(passphrase=PASSPHRASE) if PASSPHRASE else SrtSettings()

    try:
        create_resp = video_client.create_ingress_stream(
            CreateIngressStreamRequest(
                ingress_id=ingress_id,
                title="cam-01",
                ingress=Oneof("srt", srt_settings),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to create ingress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully started stream with ID: {create_resp.ingress_id}")

    if not create_resp.has_field("srt"):
        print("Server did not return an SRT ingress URL", file=sys.stderr)
        sys.exit(1)
    # The streamid begins with '#', which a URL parser (e.g. ffmpeg's) treats
    # as a fragment and strips. Percent-encode it so the full streamid survives.
    push_url = create_resp.ingress.value.url.replace("#", "%23")
    print(f"Push SRT to: {push_url}")
    print(f"Session ID: {create_resp.ingress.value.session_id}")
    if PASSPHRASE:
        print("Encryption: AES (supply the passphrase on the connection)")

    # Attach the new video to the asset entity. A field_path of media.media replaces
    # only the media list, leaving every other component on the entity untouched.
    now_ts = Timestamp.from_datetime(datetime.now(timezone.utc))

    try:
        em_client.override_entity(
            OverrideEntityRequest(
                entity=Entity(
                    entity_id=ENTITY_ID,
                    media=Media(
                        media=[
                            MediaItem(
                                item_identifier=ingress_id,
                                type=MediaType.VIDEO,
                            ),
                        ],
                    ),
                ),
                field_path=["media.media"],
                provenance=Provenance(
                    integration_name="lattice-video-integration",
                    data_type="video-association",
                    source_update_time=now_ts,
                ),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to override entity media: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Associated video {ingress_id} with entity {ENTITY_ID}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::{
    Entity, Media, MediaItem, MediaType, OverrideEntityRequest, Provenance,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::{
    create_ingress_stream_request::Ingress as IngressSettings,
    create_ingress_stream_response::Ingress as IngressResult, CreateIngressStreamRequest,
    SrtSettings,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use prost_types::Timestamp;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;
use uuid::Uuid;

use std::env;
use std::time::{SystemTime, UNIX_EPOCH};

// Entity ID to associate the stream with. Replace before running.
const ENTITY_ID: &str = "<ENTITY-ID>";
// Optional SRT passphrase; applies AES encryption to the stream when non-empty.
const PASSPHRASE: &str = "";

fn now_timestamp() -> Timestamp {
    let d = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
    Timestamp {
        seconds: d.as_secs() as i64,
        nanos: d.subsec_nanos() as i32,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    // Both clients share the same channel and inject the same auth metadata.
    let video_auth = auth_header.clone();
    let video_sandbox = sandbox_header.clone();
    let mut video_client = VideoManagerApiClient::with_interceptor(
        channel.clone(),
        move |mut req: Request<()>| {
            req.metadata_mut().insert("authorization", video_auth.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", video_sandbox.clone());
            Ok(req)
        },
    );
    let mut em_client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Generate a client-side UUID so the stream is correlatable and the call is retry-safe.
    let ingress_id = Uuid::new_v4().to_string();
    println!("Starting SRT ingress stream {}", ingress_id);

    // SRT is a push protocol: Lattice allocates a listener and returns a URL and
    // session ID the producer must use. An optional passphrase applies AES encryption.
    let srt_settings = SrtSettings {
        passphrase: if PASSPHRASE.is_empty() {
            None
        } else {
            Some(PASSPHRASE.to_string())
        },
    };
    let create_req = Request::new(CreateIngressStreamRequest {
        ingress_id: Some(ingress_id.clone()),
        title: "cam-01".to_string(),
        ingress: Some(IngressSettings::Srt(srt_settings)),
    });
    let create_resp = video_client.create_ingress_stream(create_req).await?.into_inner();
    println!("Successfully started stream with ID: {}", create_resp.ingress_id);

    match create_resp.ingress {
        Some(IngressResult::Srt(srt)) => {
            // The streamid begins with '#', which a URL parser (e.g. ffmpeg's) treats
            // as a fragment and strips. Percent-encode it so the full streamid survives.
            let push_url = srt.url.replace('#', "%23");
            println!("Push SRT to: {}", push_url);
            println!("Session ID: {}", srt.session_id);
            if !PASSPHRASE.is_empty() {
                println!("Encryption: AES (supply the passphrase on the connection)");
            }
        }
        _ => {
            eprintln!("Server did not return an SRT ingress URL");
            std::process::exit(1);
        }
    }

    // Attach the new video to the asset entity. A field_path of media.media replaces
    // only the media list, leaving every other component on the entity untouched.
    let override_req = Request::new(OverrideEntityRequest {
        entity: Some(Entity {
            entity_id: ENTITY_ID.to_string(),
            media: Some(Media {
                media: vec![MediaItem {
                    item_identifier: ingress_id.clone(),
                    r#type: MediaType::Video as i32,
                    relative_path: String::new(),
                }],
            }),
            ..Default::default()
        }),
        field_path: vec!["media.media".to_string()],
        provenance: Some(Provenance {
            integration_name: "lattice-video-integration".to_string(),
            data_type: "video-association".to_string(),
            source_update_time: Some(now_timestamp()),
            ..Default::default()
        }),
    });
    match em_client.override_entity(override_req).await {
        Ok(_) => println!("Associated video {} with entity {}", ingress_id, ENTITY_ID),
        Err(e) => eprintln!("Failed to override entity media: {}", e),
    }

    Ok(())
}

```

#### Push the stream

Encode and push video with `ffmpeg`. The command below streams an animated test pattern, so you
can verify the stream:

```bash
ffmpeg -re -f lavfi -i "testrc2=size=1280x720:rate=24" \
  -c:v libx264 -preset veryfast -tune zerolatency -pix_fmt yuv420p \
  -f mpegts "srt://<host>:<port>?streamid=%23!::s=<session-id>"
```

If you set a `passphrase`, append the SRT encryption parameters to the URL: `pbkeylen=16` selects AES-128:

```bash
ffmpeg -re -f lavfi -i "mandelbrot=size=1280x720:rate=30" \
  -c:v libx264 -preset veryfast -tune zerolatency -pix_fmt yuv420p \
  -f mpegts "srt://<host>:<port>?streamid=%23!::s=<session-id>&passphrase=<your-passphrase>&pbkeylen=16"
```

Once packets start arriving, operators select the camera asset in Lattice to view the live feed.

![Shows the live camera feed playing for the selected asset in Lattice.](/_fern-files/developer.anduril.com/anduril.docs.buildwithfern.com/2edb4fcc76cbbf5d6fd0e34191732654fc189aea27ff16cbecf388b1da322c55/assets/images/screenshots/lattice-ui-live-feed.gif)

## Re-publish a stream to a downstream consumer

An egress stream re-publishes a live stream out of Lattice so an external consumer, such as another C2 system, a recorder, or
an analytics pipeline, can pull the feed. A single ingress stream can feed many egress consumers, each with its own transport
and egress stream.

Creating an egress stream requires the `ingress_id` of a live stream. Use the ID returned when you
[started the stream](#start-a-stream), or discover it with [`ListIngressStreams`](/guides/video/manage-streams#list-active-streams).
The source must be live: Lattice rejects a stream whose status is `STREAM_STATUS_UNAVAILABLE` as not-live.

`CreateEgressStream` returns an `egress_id`. Persist it to [retrieve](/guides/video/manage-streams#get-a-single-egress-stream) or
[stop](/guides/video/manage-streams#stop-an-egress-stream) the egress stream later. Egress over both RTSP and SRT is pull-based:
Lattice hosts the endpoint and the downstream consumer connects to pull the feed.

#### RTSP

Lattice hosts an RTSP server for the egress stream and returns the URL to distribute to consumers. Send empty RTSP settings
to select the transport; no source URL is needed, because Lattice sources the frames from the ingress stream. The response
contains an `RtspEgress` with a `url` that downstream consumers pull from:

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to re-publish a live ingress
// stream to a downstream consumer as an RTSP egress stream.
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)

// Ingress ID of the live stream to re-publish. Replace before running.
const ingressID = "<INGRESS-ID>"

func main() {
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")

	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	ctx := context.Background()

	fmt.Printf("Creating RTSP egress stream for ingress %s\n", ingressID)

	// RTSP egress is pull-based: Lattice hosts an RTSP server and returns the URL the
	// downstream consumer pulls from. Send empty RtspSettings to select the RTSP variant;
	// no source URL is needed because Lattice supplies the frames from the ingress stream.
	createResp, err := LatticeClient.Video.CreateEgressStream(ctx, &Lattice.CreateEgressStreamRequest{
		IngressID: Lattice.String(ingressID),
		Rtsp:      &Lattice.RtspSettings{},
	})
	if err != nil {
		fmt.Printf("Failed to create egress stream: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Successfully created egress stream with ID: %s\n", *createResp.GetEgressID())

	if createResp.GetRtsp() == nil || createResp.GetRtsp().GetURL() == nil {
		fmt.Println("Server did not return an RTSP egress URL")
		os.Exit(1)
	}
	fmt.Printf("Downstream consumers pull from: %s\n", *createResp.GetRtsp().GetURL())
}

```

```java title={"Java (REST)"}
package org.example;

import com.anduril.Lattice;
import com.anduril.resources.video.requests.CreateEgressStreamRequest;
import com.anduril.types.CreateEgressStreamResponse;
import com.anduril.types.RtspEgress;
import com.anduril.types.RtspSettings;

/**
 * Re-publishes a live ingress stream to a downstream consumer as an RTSP egress stream with the
 * Lattice SDK.
 */
public class CreateEgressStreamRtsp {
    // Ingress ID of the live stream to re-publish. Replace before running.
    private static final String INGRESS_ID = "<INGRESS-ID>";

    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        // Check required variables and ensure the endpoint has its scheme
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            System.out.println("Creating RTSP egress stream for ingress " + INGRESS_ID);

            // RTSP egress is pull-based: Lattice hosts an RTSP server and returns the URL the
            // downstream consumer pulls from. Send empty RtspSettings to select the RTSP variant;
            // no source URL is needed because Lattice supplies the frames from the ingress stream.
            CreateEgressStreamResponse response = client.video().createEgressStream(
                CreateEgressStreamRequest.builder()
                    .ingressId(INGRESS_ID)
                    .rtsp(RtspSettings.builder().build())
                    .build());
            System.out.println("Successfully created egress stream with ID: "
                + response.getEgressId().orElse(""));

            RtspEgress rtsp = response.getRtsp().orElse(null);
            if (rtsp == null || rtsp.getUrl().isEmpty()) {
                System.err.println("Server did not return an RTSP egress URL.");
                System.exit(1);
            }
            System.out.println("Downstream consumers pull from: " + rtsp.getUrl().orElse(""));
        } catch (Exception e) {
            System.err.println("Failed to create egress stream: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to re-publish a live ingress
# stream to a downstream consumer as an RTSP egress stream.

import os
import sys

from anduril import Lattice, RtspSettings

# Ingress ID of the live stream to re-publish. Replace before running.
INGRESS_ID = "<INGRESS-ID>"


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")

    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")
    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables.", file=sys.stderr)
        sys.exit(1)

    client = Lattice(
        base_url=f"https://{lattice_endpoint}",
        token=lambda: str(environment_token),
        # Remove the following header if you are not developing on Sandboxes.
        headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"},
    )

    print(f"Creating RTSP egress stream for ingress {INGRESS_ID}")

    # RTSP egress is pull-based: Lattice hosts an RTSP server and returns the URL the
    # downstream consumer pulls from. Send empty RtspSettings to select the RTSP variant;
    # no source URL is needed because Lattice supplies the frames from the ingress stream.
    try:
        create_resp = client.video.create_egress_stream(
            ingress_id=INGRESS_ID,
            rtsp=RtspSettings(),
        )
    except Exception as error:
        print(f"Failed to create egress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully created egress stream with ID: {create_resp.egress_id}")

    if not create_resp.rtsp or not create_resp.rtsp.url:
        print("Server did not return an RTSP egress URL.", file=sys.stderr)
        sys.exit(1)
    print(f"Downstream consumers pull from: {create_resp.rtsp.url}")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
import { LatticeClient } from "@anduril-industries/lattice-sdk";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;

// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Ingress ID of the live stream to re-publish. Replace before running.
const ingressId = "<INGRESS-ID>";

async function App() {
    console.log(`Creating RTSP egress stream for ingress ${ingressId}`);

    try {
        // RTSP egress is pull-based: Lattice hosts an RTSP server and returns the URL the
        // downstream consumer pulls from. Send empty RTSP settings to select the RTSP variant;
        // no source URL is needed because Lattice supplies the frames from the ingress stream.
        const createResp = await client.video.createEgressStream({
            ingressId: ingressId,
            rtsp: {}
        });
        console.log(`Successfully created egress stream with ID: ${createResp.egressId}`);

        if (!createResp.rtsp?.url) {
            console.log("Server did not return an RTSP egress URL.");
            return;
        }
        console.log(`Downstream consumers pull from: ${createResp.rtsp.url}`);
    } catch (error) {
        console.log(`Encountered the following error while creating the RTSP egress stream: ${error}`);
    }
}

App();

```

```go title={"Go (gRPC)"}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// Ingress ID of the live stream to re-publish. Replace before running.
const ingressID = "<INGRESS-ID>"

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}

	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}
	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	client := videomanagerv1grpc.NewVideoManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	fmt.Printf("Creating RTSP egress stream for ingress %s\n", ingressID)

	// RTSP egress is pull-based: Lattice hosts an RTSP server and returns the URL the
	// downstream consumer pulls from. Send an empty RtspSettings to select the RTSP
	// variant; no source URL is needed because Lattice supplies the frames from the
	// ingress stream.
	createResp, err := client.CreateEgressStream(ctx, &videomanagerv1.CreateEgressStreamRequest{
		IngressId: ingressID,
		Egress: &videomanagerv1.CreateEgressStreamRequest_Rtsp{
			Rtsp: &videomanagerv1.RtspSettings{},
		},
	})
	if err != nil {
		log.Fatalf("Failed to create egress stream: %v", err)
	}
	fmt.Printf("Successfully created egress stream with ID: %s\n", createResp.GetEgressId())

	rtsp := createResp.GetRtsp()
	if rtsp == nil {
		log.Fatalf("Server did not return an RTSP egress URL")
	}
	fmt.Printf("Downstream consumers pull from: %s\n", rtsp.GetUrl())
}

```

```py title={"Python (gRPC)"}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
from bearer_auth import create_client
from protobuf import Oneof

from anduril.videomanager.v1.create_egress_stream_request_pub_pb import (
    CreateEgressStreamRequest,
)
from anduril.videomanager.v1.rtsp_settings_pub_pb import RtspSettings
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Ingress ID of the live stream to re-publish. Replace before running.
INGRESS_ID = "<INGRESS-ID>"


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    print(f"Creating RTSP egress stream for ingress {INGRESS_ID}")

    # RTSP egress is pull-based: Lattice hosts an RTSP server and returns the URL the
    # downstream consumer pulls from. Send empty RtspSettings to select the RTSP variant;
    # no source URL is needed because Lattice supplies the frames from the ingress stream.
    try:
        create_resp = client.create_egress_stream(
            CreateEgressStreamRequest(
                ingress_id=INGRESS_ID,
                egress=Oneof("rtsp", RtspSettings()),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to create egress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully created egress stream with ID: {create_resp.egress_id}")

    if not create_resp.has_field("rtsp"):
        print("Server did not return an RTSP egress URL", file=sys.stderr)
        sys.exit(1)
    print(f"Downstream consumers pull from: {create_resp.egress.value.url}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::{
    create_egress_stream_request::Egress as EgressSettings,
    create_egress_stream_response::Egress as EgressResult, CreateEgressStreamRequest, RtspSettings,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

// Ingress ID of the live stream to re-publish. Replace before running.
const INGRESS_ID: &str = "<INGRESS-ID>";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = VideoManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    println!("Creating RTSP egress stream for ingress {}", INGRESS_ID);

    // RTSP egress is pull-based: Lattice hosts an RTSP server and returns the URL the
    // downstream consumer pulls from. Send an empty RtspSettings to select the RTSP
    // variant; no source URL is needed because Lattice supplies the frames from the
    // ingress stream. An egress stream only re-publishes an ingress stream; it never
    // attaches to an entity.
    let create_req = Request::new(CreateEgressStreamRequest {
        ingress_id: INGRESS_ID.to_string(),
        egress: Some(EgressSettings::Rtsp(RtspSettings { url: None })),
    });
    let create_resp = client.create_egress_stream(create_req).await?.into_inner();
    println!("Successfully created egress stream with ID: {}", create_resp.egress_id);

    match create_resp.egress {
        Some(EgressResult::Rtsp(rtsp)) if !rtsp.url.is_empty() => {
            println!("Downstream consumers pull from: {}", rtsp.url);
        }
        _ => {
            eprintln!("Server did not return an RTSP egress URL");
            std::process::exit(1);
        }
    }

    Ok(())
}

```

#### SRT

Lattice allocates an SRT listener for the egress stream and returns the endpoint consumers pull from. To encrypt the stream,
set an optional `passphrase`; leave it empty for an unencrypted stream. The response contains an `SrtEgress` with a `url` and
a `session_id` that the consumer must supply on the connection, along with the passphrase if you set one:

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to re-publish a live ingress
// stream to a downstream consumer as an SRT egress stream.
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)

const (
	// Ingress ID of the live stream to re-publish. Replace before running.
	ingressID = "<INGRESS-ID>"
	// Optional SRT passphrase; applies AES encryption to the stream when non-empty.
	passphrase = ""
)

func main() {
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")

	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	ctx := context.Background()

	fmt.Printf("Creating SRT egress stream for ingress %s\n", ingressID)

	// SRT egress is pull-based: Lattice allocates a listener and returns a URL and session
	// ID the downstream consumer must use to pull the stream. An optional passphrase
	// applies AES encryption.
	srtSettings := &Lattice.SrtSettings{}
	if passphrase != "" {
		srtSettings.Passphrase = Lattice.String(passphrase)
	}

	createResp, err := LatticeClient.Video.CreateEgressStream(ctx, &Lattice.CreateEgressStreamRequest{
		IngressID: Lattice.String(ingressID),
		Srt:       srtSettings,
	})
	if err != nil {
		fmt.Printf("Failed to create egress stream: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Successfully created egress stream with ID: %s\n", *createResp.GetEgressID())

	if createResp.GetSrt() == nil || createResp.GetSrt().GetURL() == nil {
		fmt.Println("Server did not return an SRT egress URL")
		os.Exit(1)
	}
	fmt.Printf("Downstream consumers pull from: %s\n", *createResp.GetSrt().GetURL())
	fmt.Printf("Session ID: %s\n", *createResp.GetSrt().GetSessionID())
	if passphrase != "" {
		fmt.Println("Encryption: AES (supply the passphrase on the connection)")
	}
}

```

```java title={"Java (REST)"}
package org.example;

import com.anduril.Lattice;
import com.anduril.resources.video.requests.CreateEgressStreamRequest;
import com.anduril.types.CreateEgressStreamResponse;
import com.anduril.types.SrtEgress;
import com.anduril.types.SrtSettings;

/**
 * Re-publishes a live ingress stream to a downstream consumer as an SRT egress stream with the
 * Lattice SDK.
 */
public class CreateEgressStreamSrt {
    // Ingress ID of the live stream to re-publish. Replace before running.
    private static final String INGRESS_ID = "<INGRESS-ID>";
    // Optional SRT passphrase; applies AES encryption to the stream when non-empty.
    private static final String PASSPHRASE = "";

    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        // Check required variables and ensure the endpoint has its scheme
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            System.out.println("Creating SRT egress stream for ingress " + INGRESS_ID);

            // SRT egress is pull-based: Lattice allocates a listener and returns a URL and session
            // ID the downstream consumer must use to pull the stream. An optional passphrase
            // applies AES encryption.
            SrtSettings.Builder srtSettings = SrtSettings.builder();
            if (!PASSPHRASE.isEmpty()) {
                srtSettings.passphrase(PASSPHRASE);
            }
            CreateEgressStreamResponse response = client.video().createEgressStream(
                CreateEgressStreamRequest.builder()
                    .ingressId(INGRESS_ID)
                    .srt(srtSettings.build())
                    .build());
            System.out.println("Successfully created egress stream with ID: "
                + response.getEgressId().orElse(""));

            SrtEgress srt = response.getSrt().orElse(null);
            if (srt == null || srt.getUrl().isEmpty()) {
                System.err.println("Server did not return an SRT egress URL.");
                System.exit(1);
            }
            System.out.println("Downstream consumers pull from: " + srt.getUrl().orElse(""));
            System.out.println("Session ID: " + srt.getSessionId().orElse(""));
            if (!PASSPHRASE.isEmpty()) {
                System.out.println("Encryption: AES (supply the passphrase on the connection)");
            }
        } catch (Exception e) {
            System.err.println("Failed to create egress stream: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to re-publish a live ingress
# stream to a downstream consumer as an SRT egress stream.

import os
import sys

from anduril import Lattice, SrtSettings

# Ingress ID of the live stream to re-publish. Replace before running.
INGRESS_ID = "<INGRESS-ID>"
# Optional SRT passphrase; applies AES encryption to the stream when non-empty.
PASSPHRASE = ""


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")

    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")
    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables.", file=sys.stderr)
        sys.exit(1)

    client = Lattice(
        base_url=f"https://{lattice_endpoint}",
        token=lambda: str(environment_token),
        # Remove the following header if you are not developing on Sandboxes.
        headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"},
    )

    print(f"Creating SRT egress stream for ingress {INGRESS_ID}")

    # SRT egress is pull-based: Lattice allocates a listener and returns a URL and session
    # ID the downstream consumer must use to pull the stream. An optional passphrase
    # applies AES encryption.
    srt_settings = SrtSettings(passphrase=PASSPHRASE) if PASSPHRASE else SrtSettings()

    try:
        create_resp = client.video.create_egress_stream(
            ingress_id=INGRESS_ID,
            srt=srt_settings,
        )
    except Exception as error:
        print(f"Failed to create egress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully created egress stream with ID: {create_resp.egress_id}")

    if not create_resp.srt or not create_resp.srt.url:
        print("Server did not return an SRT egress URL.", file=sys.stderr)
        sys.exit(1)
    print(f"Downstream consumers pull from: {create_resp.srt.url}")
    print(f"Session ID: {create_resp.srt.session_id}")
    if PASSPHRASE:
        print("Encryption: AES (supply the passphrase on the connection)")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
import { LatticeClient } from "@anduril-industries/lattice-sdk";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;

// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Ingress ID of the live stream to re-publish. Replace before running.
const ingressId = "<INGRESS-ID>";
// Optional SRT passphrase; applies AES encryption to the stream when non-empty.
const passphrase = "";

async function App() {
    console.log(`Creating SRT egress stream for ingress ${ingressId}`);

    try {
        // SRT egress is pull-based: Lattice allocates a listener and returns a URL and session ID
        // the downstream consumer must use to pull the stream. An optional passphrase applies
        // AES encryption.
        const createResp = await client.video.createEgressStream({
            ingressId: ingressId,
            srt: passphrase ? { passphrase } : {}
        });
        console.log(`Successfully created egress stream with ID: ${createResp.egressId}`);

        if (!createResp.srt?.url) {
            console.log("Server did not return an SRT egress URL.");
            return;
        }
        console.log(`Downstream consumers pull from: ${createResp.srt.url}`);
        console.log(`Session ID: ${createResp.srt.sessionId}`);
        if (passphrase) {
            console.log("Encryption: AES (supply the passphrase on the connection)");
        }
    } catch (error) {
        console.log(`Encountered the following error while creating the SRT egress stream: ${error}`);
    }
}

App();

```

```go title={"Go (gRPC)"}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/proto"
)

const (
	// Ingress ID of the live stream to re-publish. Replace before running.
	ingressID = "<INGRESS-ID>"
	// Optional SRT passphrase; applies AES encryption to the stream when non-empty.
	passphrase = ""
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}

	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}
	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	client := videomanagerv1grpc.NewVideoManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	fmt.Printf("Creating SRT egress stream for ingress %s\n", ingressID)

	// SRT egress is pull-based: Lattice allocates a listener and returns a URL and
	// session ID the downstream consumer must use to pull the stream. An optional
	// passphrase applies AES encryption.
	srtSettings := &videomanagerv1.SrtSettings{}
	if passphrase != "" {
		srtSettings.Passphrase = proto.String(passphrase)
	}

	createResp, err := client.CreateEgressStream(ctx, &videomanagerv1.CreateEgressStreamRequest{
		IngressId: ingressID,
		Egress: &videomanagerv1.CreateEgressStreamRequest_Srt{
			Srt: srtSettings,
		},
	})
	if err != nil {
		log.Fatalf("Failed to create egress stream: %v", err)
	}
	fmt.Printf("Successfully created egress stream with ID: %s\n", createResp.GetEgressId())

	srt := createResp.GetSrt()
	if srt == nil {
		log.Fatalf("Server did not return an SRT egress URL")
	}
	fmt.Printf("Downstream consumers pull from: %s\n", srt.GetUrl())
	fmt.Printf("Session ID: %s\n", srt.GetSessionId())
	if passphrase != "" {
		fmt.Println("Encryption: AES (supply the passphrase on the connection)")
	}
}

```

```py title={"Python (gRPC)"}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
from bearer_auth import create_client
from protobuf import Oneof

from anduril.videomanager.v1.create_egress_stream_request_pub_pb import (
    CreateEgressStreamRequest,
)
from anduril.videomanager.v1.srt_settings_pub_pb import SrtSettings
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Ingress ID of the live stream to re-publish. Replace before running.
INGRESS_ID = "<INGRESS-ID>"
# Optional SRT passphrase; applies AES encryption to the stream when non-empty.
PASSPHRASE = ""


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    print(f"Creating SRT egress stream for ingress {INGRESS_ID}")

    # SRT egress is pull-based: Lattice allocates a listener and returns a URL and session
    # ID the downstream consumer must use to pull the stream. An optional passphrase
    # applies AES encryption.
    srt_settings = SrtSettings(passphrase=PASSPHRASE) if PASSPHRASE else SrtSettings()

    try:
        create_resp = client.create_egress_stream(
            CreateEgressStreamRequest(
                ingress_id=INGRESS_ID,
                egress=Oneof("srt", srt_settings),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to create egress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Successfully created egress stream with ID: {create_resp.egress_id}")

    if not create_resp.has_field("srt"):
        print("Server did not return an SRT egress URL", file=sys.stderr)
        sys.exit(1)
    print(f"Downstream consumers pull from: {create_resp.egress.value.url}")
    print(f"Session ID: {create_resp.egress.value.session_id}")
    if PASSPHRASE:
        print("Encryption: AES (supply the passphrase on the connection)")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::{
    create_egress_stream_request::Egress as EgressSettings,
    create_egress_stream_response::Egress as EgressResult, CreateEgressStreamRequest, SrtSettings,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

// Ingress ID of the live stream to re-publish. Replace before running.
const INGRESS_ID: &str = "<INGRESS-ID>";
// Optional SRT passphrase; applies AES encryption to the stream when non-empty.
const PASSPHRASE: &str = "";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = VideoManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    println!("Creating SRT egress stream for ingress {}", INGRESS_ID);

    // SRT egress is pull-based: Lattice allocates a listener and returns a URL and session
    // ID the downstream consumer must use to pull the stream. An optional passphrase
    // applies AES encryption. An egress stream only re-publishes an ingress stream; it
    // never attaches to an entity.
    let srt_settings = SrtSettings {
        passphrase: if PASSPHRASE.is_empty() {
            None
        } else {
            Some(PASSPHRASE.to_string())
        },
    };
    let create_req = Request::new(CreateEgressStreamRequest {
        ingress_id: INGRESS_ID.to_string(),
        egress: Some(EgressSettings::Srt(srt_settings)),
    });
    let create_resp = client.create_egress_stream(create_req).await?.into_inner();
    println!("Successfully created egress stream with ID: {}", create_resp.egress_id);

    match create_resp.egress {
        Some(EgressResult::Srt(srt)) if !srt.url.is_empty() => {
            println!("Downstream consumers pull from: {}", srt.url);
            println!("Session ID: {}", srt.session_id);
            if !PASSPHRASE.is_empty() {
                println!("Encryption: AES (supply the passphrase on the connection)");
            }
        }
        _ => {
            eprintln!("Server did not return an SRT egress URL");
            std::process::exit(1);
        }
    }

    Ok(())
}

```

## What's next

* Learn how to [manage existing streams](/guides/video/manage-streams).