Skip to navigation
Video

Create video streams

Preview

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

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 before running the examples on this page.

Start a stream

Lattice supports the following video streaming protocols:

If you don’t already have an RTSP source, stand one up with MediaMTX:

1

Install MediaMTX

Download a pre-built MediaMTX binary for your platform from the MediaMTX releases page and place it on the host that publishes the camera feed.

2

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 for the available source types.

mediamtx.yml
rtspAddress: :<your-port>
paths:
  cam:
    source: <your-camera-source>
3

Start MediaMTX

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

./mediamtx mediamtx.yml

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

4

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.

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:

1

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.

// 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)
	}
}

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.

2

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:

// 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)
}

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.

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, or discover it with ListIngressStreams. 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 or stop 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.

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:

// 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())
}

What’s next