scenescape/controller/src/controller/detections_builder.py at af6c1994bca5a5d18666c33d3254480cbc012f61 · open-edge-platform/scenescape · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
# SPDX-FileCopyrightText: (C) 2024 - 2026 Intel Corporation
# SPDX-License-Identifier: Apache-2.0

import numpy as np

from controller.scene import TripwireEvent
from scene_common.earth_lla import convertXYZToLLA, calculateHeading
from scene_common.geometry import DEFAULTZ, Point, Size
from scene_common.timestamp import get_iso_time


def buildDetectionsDict(objects, scene):
  result_dict = {}
  for obj in objects:
    obj_dict = prepareObjDict(scene, obj, False)
    result_dict[obj_dict['id']] = obj_dict
  return result_dict

def buildDetectionsList(objects, scene, update_visibility=False):
  result_list = []
  for obj in objects:
    obj_dict = prepareObjDict(scene, obj, update_visibility)
    result_list.append(obj_dict)
  return result_list

def prepareObjDict(scene, obj, update_visibility):
  aobj = obj
  if isinstance(obj, TripwireEvent):
    aobj = obj.object
  otype = aobj.category

  scene_loc_vector = aobj.sceneLoc.asCartesianVector

  velocity = aobj.velocity
  if velocity is None:
    velocity = Point(0, 0, 0)
  if not velocity.is3D:
    velocity = Point(velocity.x, velocity.y, DEFAULTZ)

  obj_dict = aobj.info
  obj_dict.update({
    'id': aobj.gid, # gid is the global ID - computed by SceneScape server.
    'type': otype,
    'translation': scene_loc_vector,
    'size': aobj.size,
    'velocity': velocity.asCartesianVector
  })

  rotation = aobj.rotation
  if rotation is not None:
    obj_dict['rotation'] = rotation

  if scene and scene.output_lla:
    lat_long_alt = convertXYZToLLA(scene.trs_xyz_to_lla, scene_loc_vector)
    obj_dict['lat_long_alt'] = lat_long_alt.tolist()
    heading = calculateHeading(scene.trs_xyz_to_lla, aobj.sceneLoc.asCartesianVector, velocity.asCartesianVector)
    obj_dict['heading'] = heading.tolist()

  reid = aobj.reidVector
  if reid is not None:
    if isinstance(reid, np.ndarray):
      obj_dict['reid'] = reid.tolist()
    else:
      obj_dict['reid'] = reid

  if hasattr(aobj, 'visibility'):
    obj_dict['visibility'] = aobj.visibility
    if update_visibility:
      computeCameraBounds(scene, aobj, obj_dict)

  chain_data = aobj.chain_data
  if len(chain_data.regions):
    obj_dict['regions'] = chain_data.regions
  if len(chain_data.sensors):
    obj_dict['sensors'] = chain_data.sensors
  if hasattr(aobj, 'confidence'):
    obj_dict['confidence'] = aobj.confidence
  if hasattr(aobj, 'similarity'):
    obj_dict['similarity'] = aobj.similarity
  if hasattr(aobj, 'first_seen'):
    obj_dict['first_seen'] = get_iso_time(aobj.first_seen)
  if isinstance(obj, TripwireEvent):
    obj_dict['direction'] = obj.direction
  if hasattr(aobj, 'asset_scale'):
    obj_dict['asset_scale'] = aobj.asset_scale
  if len(aobj.chain_data.persist):
    obj_dict['persistent_data'] = aobj.chain_data.persist
  # Preserve camera_bounds from deserialized objects (analytics-only mode)
  if hasattr(aobj, '_camera_bounds') and aobj._camera_bounds:
    obj_dict['camera_bounds'] = aobj._camera_bounds
  return obj_dict

def computeCameraBounds(scene, aobj, obj_dict):
  camera_bounds = {}
  for cameraID in obj_dict['visibility']:
    bounds = None
    if aobj and len(aobj.vectors) > 0 and hasattr(aobj.vectors[0].camera, 'cameraID') \
          and cameraID == aobj.vectors[0].camera.cameraID:
      bounds = getattr(aobj, 'boundingBoxPixels', None)
    elif scene:
      camera = scene.cameraWithID(cameraID)
      if camera is not None and 'bb_meters' in obj_dict:
        obj_translation = None
        obj_size = None
        if aobj:
          obj_translation = aobj.sceneLoc
          obj_size = aobj.bbMeters.size
        else:
          obj_translation = Point(obj_dict['translation'])
          obj_size = Size(obj_dict['bb_meters']['width'], obj_dict['bb_meters']['height'])
        bounds = camera.pose.projectEstimatedBoundsToCameraPixels(obj_translation,
                                                                  obj_size)
    if bounds:
      camera_bounds[cameraID] = bounds.asDict
  obj_dict['camera_bounds'] = camera_bounds
  return