forked from datatorch-actions/dextr
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathentry.py
116 lines (91 loc) · 3.49 KB
/
entry.py
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
from datatorch import get_input, agent, set_output
from datatorch.api.api import ApiClient
from datatorch.api.entity.sources.image import Segmentations
from datatorch.api.scripts.utils.simplify import simplify_points
import requests
import docker
import time
import os
from typing import List, Tuple
from docker.models.resource import Model
from urllib.parse import urlparse
Point = Tuple[float, float]
directory = os.path.dirname(os.path.abspath(__file__))
agent_dir = agent.directories().root
points = get_input("points")
image_path = get_input("imagePath")
address = urlparse(get_input("url"))
image = get_input("image")
annotation_id = get_input("annotationId")
simplify = get_input("simplify")
# [[10,20],[30, 40],[50,60],[70,80]]
# points: List[Point] = [(10.0, 20.0), (30.0, 40.0), (50.0, 60.0), (70.0, 80.0)]
# image_path = "/home/desktop/.config/datatorch/agent/temp/download-file/20201025_102443 (17th copy).jpg"
CONTAINER_NAME = "datatorch-dextr-action"
def valid_image_path():
if not image_path.startswith(agent_dir):
print(f"Directory must be inside the agent folder ({agent_dir}).")
exit(1)
if not os.path.isfile(image_path):
print(f"Image path must be a file ({image_path}).")
exit(1)
def start_server(port: int):
docker_client = docker.from_env()
print(f"Creating DEXTR container on port {port}.")
print(f"Downloading {image} docker image. This may take a few mins.", flush=True)
container = docker_client.containers.run(
image,
detach=True,
ports={"8000/tcp": port},
restart_policy={"Name": "always"},
volumes={agent_dir: {"bind": "/agent", "mode": "rw"}},
)
if isinstance(container, Model):
print(f"Created DEXTR Container ({container.short_id}).")
def call_dextr(path: str, points: List[Point], address: str) -> List[List[Point]]:
agent_folder = agent.directories().root
container_path = path.replace(agent_folder, "/agent")
print(f"Sending request to '{address}' (POST)")
print(f"Image Path = {path}")
print(f"Container Path = {container_path}")
print(f"Points = {points}")
response = requests.post(address, json={"path": container_path, "points": points})
json = response.json()
print(f"Response = {json}")
return json["polygons"]
def send_request():
attempts = 0
while True:
try:
attempts += 1
print(f"Attemp {attempts}: Request to DEXTR Server")
seg = call_dextr(image_path, points, address.geturl())
output_seg = (
seg
if simplify == 0
else [
simplify_points(polygon, tolerance=simplify, highestQuality=False)
for polygon in seg
]
)
set_output("polygons", output_seg)
print(annotation_id)
if annotation_id is not None:
print(f"Creating segmentation source for annotation {annotation_id}")
s = Segmentations()
s.annotation_id = annotation_id
s.path_data = output_seg
s.create(ApiClient())
exit(0)
except Exception as ex:
if attempts > 5:
print(ex)
break
print(f"Attemp {attempts}: Could not connect to dextr.")
start_server(address.port or 80)
time.sleep(5)
print("Could not send request.")
exit(1)
if __name__ == "__main__":
valid_image_path()
send_request()