> 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: : paths: cam: 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 :` once the server is ready. #### Verify the stream From any host on the same network, confirm that the feed is playable. Replace `` with the address of the host running MediaMTX. ```bash ffplay rtsp://:/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/...")}, }, }, } // 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.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/...") .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/..."), ], ), ) 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/..." } ] } } ); 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/..."}, }, }, } _, 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/..."), ], ), ) 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> { 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/..." .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 = "" // RTSP source Lattice will pull from. rtspURL = "rtsp://:/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 = ""; // RTSP source Lattice will pull from. private static final String RTSP_URL = "rtsp://:/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 = "" # RTSP source Lattice will pull from. RTSP_URL = "rtsp://:/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 = ""; // RTSP source Lattice will pull from. const rtspUrl = "rtsp://:/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 = "" // RTSP source Lattice will pull from. rtspURL = "rtsp://:/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 = "" # RTSP source Lattice will pull from. RTSP_URL = "rtsp://:/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 = ""; // RTSP source Lattice will pull from. const RTSP_URL: &str = "rtsp://:/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> { 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/...")}, }, }, } // 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.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/...") .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/..."), ], ), ) 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/..." } ] } } ); 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/..."}, }, }, } _, 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/..."), ], ), ) 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> { 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/..." .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 = "" 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 = ""; 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 = "" 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 = ""; 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 = "" 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 = "" 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 = ""; 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> { 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 `:` 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 \ ":?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/...")}, }, }, } // 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.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/...") .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/..."), ], ), ) 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/..." } ] } } ); 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/..."}, }, }, } _, 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/..."), ], ), ) 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> { 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/..." .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 = "" // 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 = ""; // 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 = "" # 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 = ""; // 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 = "" // 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 = "" # 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 = ""; // 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> { 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://:?streamid=%23!::s=" ``` 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://:?streamid=%23!::s=&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 = "" 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 = ""; 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 = "" 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 = ""; 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 = "" 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 = "" 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 = ""; #[tokio::main] async fn main() -> Result<(), Box> { 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 = "" // 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 = ""; // 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 = "" # 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 = ""; // 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 = "" // 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 = "" # 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 = ""; // Optional SRT passphrase; applies AES encryption to the stream when non-empty. const PASSPHRASE: &str = ""; #[tokio::main] async fn main() -> Result<(), Box> { 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). > Publish a camera asset and start an RTSP, MPEG-TS or SRT ingress stream.