Entities

Watch entities

Monitoring entities in Lattice using the Lattice SDK

This shows how to use the SDK to fetch entities from Lattice and stream real-time information about entity components.

Complete the steps to learn how to use the following APIs:

  • GetEntity — Retrieves a single entity from Lattice.

Before you begin

  • To configure your app to watch entities, set up your Lattice environment.
  • Learn about required components and various entity shapes in Lattice.

Get an entity

Get details of a specific entity using an entity_ID and the GetEntity API:

1

Find the entity URL

Open the entity details panel from the Lattice UI toolbar:

Shows the drop down menu where you can see the entity details.
2

Copy the entity ID

Choose Copy Content from the drop down menu, then choose Copy Asset URL:

Shows the drop down menu where you can copy asset URL and find the entity ID.

Extract the unique entity ID from this URL. For example, in the following, the entity ID is YOUR_ENTITY_ID: https://your_lattice_url.com/c2/entities/YOUR_ENTITY_ID

3

Get the entity object

Use the GetEntity API action to retrieve a single entity object from Lattice. Replace $ENTITY_ID in the following example with the entity ID you copied in the previous step:

1package main
2
3import (
4	"context"
5	"fmt"
6	"net/http"
7	"os"
8	"time"
9
10	"github.com/anduril/lattice-sdk-go/v2/client"
11	"github.com/anduril/lattice-sdk-go/v2/option"
12)
13
14func main() {
15	// Get environment variables
16	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
17	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
18
19	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
20	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
21
22	// Check required environment variables
23	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
24		fmt.Println("Missing required environment variables")
25		os.Exit(1)
26	}
27
28	// Initialize headers for sandbox authorization
29	headers := http.Header{}
30	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
31
32	// Create the client
33	LatticeClient := client.NewClient(
34		option.WithToken(environmentToken),
35		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
36		option.WithHTTPHeader(headers),
37	)
38
39	// Continuously get the entity
40	for {
41		// Create context for the request
42		ctx := context.Background()
43
44		// Get the entity
45		entity, err := LatticeClient.Entities.GetEntity(
46			ctx,
47			"Demo-Sim-Asset1",
48		)
49
50		// Handle errors
51		if err != nil {
52			fmt.Printf("Error: %v\n", err)
53		} else {
54			fmt.Printf("Asset name     | %s\n", *entity.Aliases.Name)
55			fmt.Printf("Asset location | %f, %f\n", *entity.Location.Position.LatitudeDegrees, *entity.Location.Position.LongitudeDegrees)
56		}
57
58		// Wait before next request
59		time.Sleep(5 * time.Second)
60	}
61}
4

Verify the response

If successful, you’ll see the entity’saliases.name and its real-time location logged in the console:

$Asset name     | Demo-Sim-Asset1
>Asset location | 37.7749, -122.4194

Stream entity components

Available only in gRPC

This API is availble only in gRPC. To use it, you must generate your own artifacts using the Lattice protobuf definitions in the Buf Schema Registry website.

For more information about generating your own artifacts, see Generate Lattice SDK gRPC.

Use the StreamEntityComponents API to stream components whenever an entity object is updated. Lattice lets you stream all components when you set includeAllComponents to true. To specify which components to stream, set the componentsToInclude and provide a list of components in snake_case: [“location_uncertainty”, “aliases”]

1

Stream all components

Use the StreamEntityComponents action to start a stream:

Python
1from anduril.entitymanager.v1 import EntityManagerApiStub, StreamEntityComponentsRequest
2from grpclib.client import Channel
3import asyncio, logging, os, sys
4
5lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
6environment_token = os.getenv('ENVIRONMENT_TOKEN')
7# Remove the following statement if you are not developing on a Sandboxes.
8sandboxes_token = os.getenv('SANDBOXES_TOKEN')
9
10# Remove sandboxes_token from the following statement if you are not developing on a Sandboxes.
11if not environment_token or not lattice_endpoint or not sandboxes_token:
12    logging.error("Missing required environment variables.")
13    sys.exit(1)
14    
15# Set the authentication header as a dictionary.
16metadata = {
17    'authorization': f"Bearer {environment_token}",
18    # Remove the following header if you are not developing on a Sandboxes.
19    'anduril-sandbox-authorization': f"Bearer {sandboxes_token}"
20}
21async def stream_entities():
22    # open secure channel
23    channel = Channel(host=lattice_endpoint, port=443, ssl=True)
24
25    # create service instance
26    entity_manager_stub = EntityManagerApiStub(channel)
27
28    # Stream all entities with all components.
29    stream = entity_manager_stub.stream_entity_components(
30        StreamEntityComponentsRequest(include_all_components=True),
31        metadata=metadata
32    )
33    
34    async for event in stream:
35        logging.info(event.entity_event.entity)
36
37if __name__ == "__main__":
38    raise SystemExit(asyncio.run(stream_entities()))
2

Stream specific components

Use the StreamEntityComponents and specify the components to stream in snake_case:

Python
1from anduril.entitymanager.v1 import EntityManagerApiStub, StreamEntityComponentsRequest
2from grpclib.client import Channel
3import asyncio, logging, os, sys
4
5lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
6environment_token = os.getenv('ENVIRONMENT_TOKEN')
7# Remove the following statement if you are not developing on a Sandboxes.
8sandboxes_token = os.getenv('SANDBOXES_TOKEN')
9
10# Remove sandboxes_token from the following statement if you are not developing on a Sandboxes.
11if not environment_token or not lattice_endpoint or not sandboxes_token:
12    logging.error("Missing required environment variables.")
13    sys.exit(1)
14    
15# Set the authentication header as a dictionary.
16metadata = {
17    'authorization': f"Bearer {environment_token}",
18    # Remove the following header if you are not developing on a Sandboxes.
19    'anduril-sandbox-authorization': f"Bearer {sandboxes_token}"
20}
21async def stream_entities():
22    channel = Channel(host=lattice_endpoint, port=443, ssl=True)
23
24    # Create an API stub for the EntityManager.
25    entity_manager_stub = EntityManagerApiStub(channel)
26
27    # Stream all entities with all components.
28    stream = entity_manager_stub.stream_entity_components(
29        StreamEntityComponentsRequest(components_to_include=["location_uncertainty", "aliases"]),
30        metadata=metadata
31    )
32    
33    async for event in stream:
34        logging.info(event.entity_event.entity)
35
36if __name__ == "__main__":
37    raise SystemExit(asyncio.run(stream_entities()))

What’s next?