#832

Committed 12 Oct 2023 08:31AM UTC coverage: 0.0% (-68.5%) from 68.493%

Build # #832

Build Type

push

circle-ci

Committed by

xzdandy

Commit Message

Minor

Run Details

0 of 12331 relevant lines covered (0.0%)

0.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/evadb/functions/abstract/tracker_abstract_function.py

# coding=utf-8
# Copyright 2018-2023 EvaDB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import typing

import numpy
import pandas as pd

from evadb.catalog.catalog_type import NdArrayType
from evadb.functions.abstract.abstract_function import AbstractFunction
from evadb.functions.decorators.decorators import forward, setup
from evadb.functions.decorators.io_descriptors.data_types import PandasDataframe


class EvaDBTrackerAbstractFunction(AbstractFunction):
    """
    An abstract class for all EvaDB object trackers.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    @setup(cacheable=False, function_type="object_tracker", batchable=False)
    def setup(self, *args, **kwargs):
        super().setup(*args, **kwargs)

    @forward(
        input_signatures=[
            PandasDataframe(
                columns=["frame_id", "frame", "bboxes", "scores", "labels"],
                column_types=[
                    NdArrayType.INT32,
                    NdArrayType.FLOAT32,
                    NdArrayType.FLOAT32,
                    NdArrayType.FLOAT32,
                    NdArrayType.STR,
                ],
                column_shapes=[(1,), (None, None, 3), (None, 4), (None,), (None,)],
            )
        ],
        output_signatures=[
            PandasDataframe(
                columns=["track_ids", "track_labels", "track_bboxes", "track_scores"],
                column_types=[
                    NdArrayType.INT32,
                    NdArrayType.INT32,
                    NdArrayType.FLOAT32,
                    NdArrayType.FLOAT32,
                ],
                column_shapes=[(None,), (None,), (None, 4), (None,)],
            )
        ],
    )
    def forward(
        self,
        frame_id: numpy.ndarray,
        frame: numpy.ndarray,
        labels: numpy.ndarray,
        bboxes: numpy.ndarray,
        scores: numpy.ndarray,
    ) -> typing.Tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray]:
        """
        Args:
            frame_id (numpy.ndarray): the frame id of current frame
            frame (numpy.ndarray): the input frame with shape (C, H, W)
            labels (numpy.ndarray): Corresponding labels for each box
            bboxes (numpy.ndarray): Array of shape `(n, 4)` or of shape `(4,)` where
            each row contains `(xmin, ymin, width, height)`.
            scores (numpy.ndarray): Corresponding scores for each box
        Returns:
            track_ids (numpy.ndarray): Corresponding track id for each box
            track_labels (numpy.ndarray): Corresponding labels for each box
            track_bboxes (numpy.ndarray):  Array of shape `(n, 4)` of tracked objects
            track_scores (numpy.ndarray): Corresponding scores for each box
        """
        raise NotImplementedError

    def __call__(self, *args, **kwargs):
        assert isinstance(
            args[0], pd.DataFrame
        ), f"Expecting pd.DataFrame, got {type(args[0])}"

        results = []
        for _, row in args[0].iterrows():
            tuple = (
                numpy.array(row[0]),
                numpy.array(row[1]),
                numpy.stack(row[2]),
                numpy.stack(row[3]),
                numpy.stack(row[4]),
            )
            results.append(self.forward(*tuple))
        return pd.DataFrame(
            results,
            columns=["track_ids", "track_labels", "track_bboxes", "track_scores"],
        )

1	# coding=utf-8
2	# Copyright 2018-2023 EvaDB
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15	import typing	×
16
17	import numpy	×
18	import pandas as pd	×
19
20	from evadb.catalog.catalog_type import NdArrayType	×
21	from evadb.functions.abstract.abstract_function import AbstractFunction	×
22	from evadb.functions.decorators.decorators import forward, setup	×
23	from evadb.functions.decorators.io_descriptors.data_types import PandasDataframe	×
24
25
26	class EvaDBTrackerAbstractFunction(AbstractFunction):	×
27	"""
28	An abstract class for all EvaDB object trackers.
29	"""
30
31	def __init__(self, args, *kwargs):	×
32	super().__init__(args, *kwargs)	×
33
34	@setup(cacheable=False, function_type="object_tracker", batchable=False)	×
35	def setup(self, args, *kwargs):	×
36	super().setup(args, *kwargs)	×
37
38	@forward(	×
39	input_signatures=[
40	PandasDataframe(
41	columns=["frame_id", "frame", "bboxes", "scores", "labels"],
42	column_types=[
43	NdArrayType.INT32,
44	NdArrayType.FLOAT32,
45	NdArrayType.FLOAT32,
46	NdArrayType.FLOAT32,
47	NdArrayType.STR,
48	],
49	column_shapes=[(1,), (None, None, 3), (None, 4), (None,), (None,)],
50	)
51	],
52	output_signatures=[
53	PandasDataframe(
54	columns=["track_ids", "track_labels", "track_bboxes", "track_scores"],
55	column_types=[
56	NdArrayType.INT32,
57	NdArrayType.INT32,
58	NdArrayType.FLOAT32,
59	NdArrayType.FLOAT32,
60	],
61	column_shapes=[(None,), (None,), (None, 4), (None,)],
62	)
63	],
64	)
65	def forward(	×
66	self,
67	frame_id: numpy.ndarray,
68	frame: numpy.ndarray,
69	labels: numpy.ndarray,
70	bboxes: numpy.ndarray,
71	scores: numpy.ndarray,
72	) -> typing.Tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray]:
73	"""
74	Args:
75	frame_id (numpy.ndarray): the frame id of current frame
76	frame (numpy.ndarray): the input frame with shape (C, H, W)
77	labels (numpy.ndarray): Corresponding labels for each box
78	bboxes (numpy.ndarray): Array of shape `(n, 4)` or of shape `(4,)` where
79	each row contains `(xmin, ymin, width, height)`.
80	scores (numpy.ndarray): Corresponding scores for each box
81	Returns:
82	track_ids (numpy.ndarray): Corresponding track id for each box
83	track_labels (numpy.ndarray): Corresponding labels for each box
84	track_bboxes (numpy.ndarray): Array of shape `(n, 4)` of tracked objects
85	track_scores (numpy.ndarray): Corresponding scores for each box
86	"""
87	raise NotImplementedError
88
89	def __call__(self, args, *kwargs):	×
90	assert isinstance(	×
91	args[0], pd.DataFrame
92	), f"Expecting pd.DataFrame, got {type(args[0])}"
93
94	results = []	×
95	for _, row in args[0].iterrows():	×
96	tuple = (	×
97	numpy.array(row[0]),
98	numpy.array(row[1]),
99	numpy.stack(row[2]),
100	numpy.stack(row[3]),
101	numpy.stack(row[4]),
102	)
103	results.append(self.forward(*tuple))	×
104	return pd.DataFrame(	×
105	results,
106	columns=["track_ids", "track_labels", "track_bboxes", "track_scores"],
107	)

georgia-tech-db / eva / #832

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous