-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcommon.py
More file actions
111 lines (93 loc) · 3.13 KB
/
common.py
File metadata and controls
111 lines (93 loc) · 3.13 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
from enum import IntEnum
from pathlib import Path
from asyncio import Lock
import tempfile
from typing import Generic, TypeVar, Dict
from typing_extensions import override
DATA_FOLDER = "data"
TEMP_FOLDER = Path(tempfile.gettempdir(), "FTCMachineLearning").absolute()
__necessary_directories = (Path(DATA_FOLDER), Path(DATA_FOLDER,
"videos"), TEMP_FOLDER)
def ensure_directory(path: Path):
path.mkdir(parents=True, exist_ok=True)
def ensure_directories():
for directory in __necessary_directories:
ensure_directory(directory)
class BackendError(IntEnum):
SUCCESS = 0
VIDEO_NOT_FOUND = 1
VIDEO_PROCESSING = 2
FRAME_NOT_FOUND = 3
NO_MORE_FRAMES = 4
UNACCEPTABLE_ALGORITHM = 5
NO_FILE_UPLOADED = 6
UNSUPPORTED_FILE_TYPE = 7
INVALID_REQUEST = 8
TRACKING_FAILED = 9
FRAME_NOT_LABELED = 10
TASK_WAS_CANCELLED = 11
INVALID_ARGUMENT = 12
@property
def error_message(self):
if self == BackendError.SUCCESS:
return "Success"
elif self == BackendError.VIDEO_NOT_FOUND:
return "Video not found"
elif self == BackendError.VIDEO_PROCESSING:
return "Frame extracting for this video is still ongoing"
elif self == BackendError.FRAME_NOT_FOUND:
return "Frame not found"
elif self == BackendError.NO_MORE_FRAMES:
return "No more frames for object tracking"
elif self == BackendError.UNACCEPTABLE_ALGORITHM:
return "Unacceptable algorithm for object tracking"
elif self == BackendError.NO_FILE_UPLOADED:
return "No file uploaded"
elif self == BackendError.UNSUPPORTED_FILE_TYPE:
return "Unsupported file type"
elif self == BackendError.INVALID_REQUEST:
return "Invalid request, check your request body"
elif self == BackendError.TRACKING_FAILED:
return "Object tracking failed"
elif self == BackendError.FRAME_NOT_LABELED:
return "Frame hasn't labeled, can't start object tracking"
elif self == BackendError.TASK_WAS_CANCELLED:
return "Task was cancelled"
elif self == BackendError.INVALID_ARGUMENT:
return "Invalid argument"
else:
return "Unknown error"
class ReturnResult:
def __init__(self, status: BackendError, *data):
self.status = status
self.message = status.error_message
self._data = data
@property
def is_success(self):
return self.status == BackendError.SUCCESS
@property
def data(self):
if len(self._data) == 1:
return self._data[0]
else:
return self._data
@classmethod
def success(cls, *data):
return cls(BackendError.SUCCESS, *data)
K = TypeVar('K')
V = TypeVar('V')
class DictProxy(Generic[K, V], Dict[K, V]):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.lock = Lock()
async def __getitem__(self, key: K):
async with self.lock:
return super().__getitem__(key)
async def __setitem__(self, key: K, value: V):
async with self.lock:
return super().__setitem__(key, value)
async def put(self, key: K, value: V):
return await self.__setitem__(key, value)
async def values(self):
async with self.lock:
return super().values()