-
Notifications
You must be signed in to change notification settings - Fork 76
Expand file tree
/
Copy pathpyfunc_common.py
More file actions
115 lines (103 loc) · 2.65 KB
/
pyfunc_common.py
File metadata and controls
115 lines (103 loc) · 2.65 KB
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
import inspect
import json
from logging import Logger
from typing import Any, Optional
from savant.base.pyfunc import PyFunc, PyFuncNoopCallException
from savant.gstreamer import Gst
from savant.gstreamer.utils import (
gst_post_stream_failed_error,
gst_post_stream_failed_warning,
)
def init_pyfunc(
gst_element: Gst.Element,
logger: Logger,
module: str,
class_name: str,
kwargs: Optional[str],
dev_mode: bool = False,
) -> Optional[PyFunc]:
# pylint: disable=broad-exception-caught
if kwargs:
try:
kwargs = json.loads(kwargs)
except Exception as exc:
return handle_fatal_error(
gst_element,
logger,
exc,
f'Failed to parse kwargs for "{module}.{class_name}" pyfunc.',
dev_mode,
return_ok=None,
return_err=None,
)
else:
kwargs = None
try:
pyfunc = PyFunc(
module=module,
class_name=class_name,
kwargs=kwargs,
dev_mode=dev_mode,
)
except Exception as exc:
return handle_fatal_error(
gst_element,
logger,
exc,
f'Failed to initialize "{module}.{class_name}" pyfunc.',
dev_mode,
return_ok=None,
return_err=None,
)
try:
pyfunc.load_user_code()
except Exception as exc:
return handle_fatal_error(
gst_element,
logger,
exc,
f'Failed to load user code for {pyfunc}.',
dev_mode,
return_ok=None,
return_err=None,
)
return pyfunc
def handle_fatal_error(
gst_element: Gst.Element,
logger: Logger,
exc: BaseException,
msg: str,
dev_mode: bool,
return_ok: Any,
return_err: Any,
) -> Any:
if dev_mode:
if not isinstance(exc, PyFuncNoopCallException):
logger.exception(msg)
return return_ok
gst_post_stream_failed_error(
gst_element=gst_element,
frame=inspect.currentframe(),
file_path=__file__,
text=msg,
)
logger.exception(msg)
return return_err
def handle_non_fatal_error(
gst_element: Gst.Element,
logger: Logger,
exc: BaseException,
msg: str,
dev_mode: bool,
):
if dev_mode:
if not isinstance(exc, PyFuncNoopCallException):
logger.warning(msg)
return
gst_post_stream_failed_warning(
gst_element=gst_element,
frame=inspect.currentframe(),
file_path=__file__,
text=msg,
)
logger.warning(msg, exc_info=exc)