Coverage for src/rift_console/ciarc_api.py: 0%
192 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-08 09:36 +0000
1from typing import Any, Optional
2import requests
3import datetime
5from loguru import logger
7import shared.constants as con
8from shared.models import (
9 Achievement,
10 BeaconObjective,
11 CameraAngle,
12 BaseTelemetry,
13 State,
14 Slot,
15 ZonedObjective,
16 live_utc,
17 HttpCode,
18)
20def console_api(
21 method: HttpCode,
22 endpoint: str,
23 params: dict[str, Any] = {},
24 json: dict[str, Any] = {},
25 files: dict[str, Any] = {},
26) -> Any:
27 """Wrapper for ciarc api with error handling."""
28 try:
29 with requests.Session() as s:
30 match method:
31 case HttpCode.GET:
32 r = s.get(endpoint, timeout=10)
33 case HttpCode.PUT:
34 r = s.put(endpoint, params=params, json=json, timeout=5)
35 case HttpCode.DELETE:
36 r = s.delete(endpoint, params=params, timeout=5)
37 case HttpCode.POST:
38 r = s.post(endpoint, params=params, files=files)
40 except requests.exceptions.ConnectionError:
41 logger.error("Console: ConnectionError - possible no VPN?")
42 return {}
43 except requests.exceptions.ReadTimeout:
44 logger.error("Console: Timeout error - possible no VPN?")
45 return {}
47 match r.status_code:
48 case 200:
49 logger.debug(f"Console: received from API - {type(r.json())} - {r.json()}")
50 return r.json()
51 case 405:
52 # this happens with illegal request, for example GET instead of PUT
53 logger.error(
54 f"Console: API Not Allowed {r.status_code} - {type(r.json())} - {r.json()}"
55 )
56 return {}
57 case 422:
58 # this happens for an illegal control request, for example accelerating while not in acquisition
59 logger.warning(
60 f"Console: API Unprocessable Content- {r.status_code} - {type(r.json())} - {r.json()}."
61 )
62 return {}
63 case 500:
64 # this happens with an bug on the api side? Should not appear anymore
65 logger.warning(
66 f"Console: API File not found - {r.status_code} - {type(r.json())} - {r.json()}."
67 )
68 return {}
69 case _:
70 # unknow error?
71 logger.warning(
72 f"Console: could not contact satellite - {r.status_code} - {type(r.json())} - {r.json()}."
73 )
74 return {}
77def console_api_image(angle: CameraAngle) -> Optional[str]:
78 """Wrapper for CIARC API with images, since it has a slighly different syntax."""
79 try:
80 with requests.Session() as s:
81 r = s.get(con.IMAGE_ENDPOINT, timeout=10)
82 except requests.exceptions.ConnectionError:
83 logger.error("Console: ConnectionError - possible no VPN?")
84 return None
85 except requests.exceptions.ReadTimeout:
86 logger.error("Console: Timeout error - possible no VPN?")
87 return None
89 match r.status_code:
90 case 200:
91 img_timestamp = datetime.datetime.fromisoformat(
92 r.headers.get("image-timestamp") or ""
93 ).strftime("%Y-%m-%dT%H:%M:%S")
94 with open(
95 con.CONSOLE_LIVE_PATH + "live_" + angle + "_" + img_timestamp + ".png",
96 "wb",
97 ) as f:
98 f.write(r.content)
99 logger.warning(f"Console: received Image - {img_timestamp}")
100 return img_timestamp
101 case 400:
102 logger.warning(f"Console: Bad request - {type(r.json())} - {r.json()}")
103 return None
104 case _:
105 logger.warning(f"Console: Unkown error: - {type(r.json())} - {r}")
106 return None
109def reset() -> None:
110 """Reset Simulation."""
111 console_api(method=HttpCode.GET, endpoint=con.RESET_ENDPOINT)
112 return
114def save_backup() -> datetime.datetime:
115 """Save backup of simulation."""
116 console_api(method=HttpCode.GET, endpoint=con.BACKUP_ENDPOINT)
117 t = live_utc()
118 logger.info("Console: saving satellite state.")
120 return t
122def load_backup(last_backup_date: Optional[datetime.datetime]) -> None:
123 """Load backup of simulation."""
124 console_api(method=HttpCode.PUT, endpoint=con.BACKUP_ENDPOINT)
125 logger.info(f"Console: restoring satellite state from {last_backup_date}.")
127 return
129def change_simulation_env(
130 is_network_simulation: bool = False, user_speed_multiplier: int = 1
131) -> None:
132 """Change simspeed or network simulation."""
133 params = {
134 "is_network_simulation": str(is_network_simulation).lower(),
135 "user_speed_multiplier": str(user_speed_multiplier),
136 }
137 console_api(method=HttpCode.PUT, endpoint=con.SIMULATION_ENDPOINT, params=params)
138 logger.info(
139 f"Console: simulation speed set to {user_speed_multiplier} - network simulation is {is_network_simulation}."
140 )
142 return
145def update_api() -> (
146 Optional[
147 tuple[
148 int,
149 list[Slot],
150 list[ZonedObjective],
151 list[BeaconObjective],
152 list[Achievement],
153 ]
154 ]
155):
156 """Pull status like slots and objectives, that are available even outside comms window."""
157 s = console_api(method=HttpCode.GET, endpoint=con.SLOTS_ENDPOINT)
158 o = console_api(method=HttpCode.GET, endpoint=con.OBJECTIVE_ENDPOINT)
159 a = console_api(method=HttpCode.GET, endpoint=con.ACHIEVEMENTS_ENDPOINT)
160 if s and o and a:
161 (slots_used, slots) = Slot.parse_api(s)
162 zoned_objectives = ZonedObjective.parse_api(o)
163 beacon_objectives = BeaconObjective.parse_api(o)
164 achievements = Achievement.parse_api(a)
165 logger.info(
166 f"Updated slots, objectives, achievments, used {slots_used} slots so far."
167 )
168 return (slots_used, slots, zoned_objectives, beacon_objectives, achievements)
169 else:
170 logger.warning("Could not update slots, objectives, achievments.")
171 return None
174def live_telemetry() -> Optional[BaseTelemetry]:
175 """Pulls /observation."""
176 d = console_api(method=HttpCode.GET, endpoint=con.OBSERVATION_ENDPOINT)
177 if d:
178 b = BaseTelemetry(**d)
179 logger.info(f"Console: received live telemetry\n{b}.")
180 return b
181 else:
182 logger.warning("Live telemtry failed.")
183 return None
186def change_angle(angle: CameraAngle) -> Any:
187 """Change camera angle, keep veloctiy constant."""
188 obs = console_api(method=HttpCode.GET, endpoint=con.OBSERVATION_ENDPOINT)
189 if not obs:
190 logger.warning("Console: no telemetry available, could not change camera angle")
191 return {}
192 json = {
193 "vel_x": obs["vx"],
194 "vel_y": obs["vy"],
195 "camera_angle": angle,
196 "state": obs["state"],
197 }
198 d = console_api(method=HttpCode.PUT, endpoint=con.CONTROL_ENDPOINT, json=json)
200 if d and d["camera_angle"] == angle:
201 logger.info(f"Console: angle changed to {d["camera_angle"]}.")
202 else:
203 logger.warning("Console: could not change angle, not in acquisition?")
204 return {}
206 return d
209def change_state(state: State) -> Any:
210 """Change State."""
211 obs = console_api(method=HttpCode.GET, endpoint=con.OBSERVATION_ENDPOINT)
212 if not obs:
213 logger.warning("Console: no telemetry available, could not change camera angle")
214 return
215 json = {
216 "vel_x": obs["vx"],
217 "vel_y": obs["vy"],
218 "camera_angle": obs["angle"],
219 "state": state,
220 }
221 d = console_api(method=HttpCode.PUT, endpoint=con.CONTROL_ENDPOINT, json=json)
223 if d and d["state"] == state:
224 logger.info(f"Console: state changed to {d["state"]}.")
225 else:
226 logger.warning("Console: could not change state, not in acquisition?")
227 return {}
229 return d
232def change_velocity(vel_x: float, vel_y: float) -> Any:
233 """Change velocity of MELVIN."""
234 obs = console_api(method=HttpCode.GET, endpoint=con.OBSERVATION_ENDPOINT)
235 if not obs:
236 logger.warning("Console: no telemetry available, could not change camera angle")
237 return {}
238 json = {
239 "vel_x": vel_x,
240 "vel_y": vel_y,
241 "camera_angle": obs["angle"],
242 "state": obs["state"],
243 }
244 d = console_api(method=HttpCode.PUT, endpoint=con.CONTROL_ENDPOINT, json=json)
246 if d and d["vel_x"] == vel_x and d["vel_y"] == vel_y:
247 logger.info(f"Console: velocity changed to ({d["vel_x"]},{d["vel_y"]}).")
248 return d
249 else:
250 logger.warning("Console: could not change velocity, not in acquisition?")
251 return {}
254def book_slot(slot_id: int, enabled: bool) -> None:
255 """Book coms slot."""
256 params = {
257 "slot_id": slot_id,
258 "enabled": str(enabled).lower(),
259 }
260 d = console_api(method=HttpCode.PUT, endpoint=con.SLOTS_ENDPOINT, params=params)
262 if d:
263 if d["enabled"]:
264 logger.info(f"Console: booked communication slot {d["id"]}.")
265 else:
266 logger.info(f"Console: cancled communication slot {d["id"]}")
267 else:
268 logger.warning("Console: could not book slot, not in acquisition?")
271def delete_objective(id: int) -> None:
272 """Delete objective (only used while testing)."""
273 params = {
274 "id": str(id),
275 }
276 d = console_api(
277 method=HttpCode.DELETE, endpoint=con.OBJECTIVE_ENDPOINT, params=params
278 )
279 if d:
280 logger.info(f"Console: removed objective with id - {id}.")
281 else:
282 logger.warning(f"Console: could not delete objective with id - {id}")
285def add_modify_zoned_objective(
286 id: int,
287 name: str,
288 start: datetime.datetime,
289 end: datetime.datetime,
290 zone: tuple[int, int, int, int],
291 optic_required: CameraAngle,
292 coverage_required: float,
293 description: str,
294 secret: bool,
295) -> bool:
296 """Add zoned or hidden objectives for testing."""
297 json = {
298 "zoned_objectives": [
299 {
300 "id": id,
301 "name": name,
302 "start": start.replace(tzinfo=datetime.timezone.utc).isoformat(),
303 "end": end.replace(tzinfo=datetime.timezone.utc).isoformat(),
304 "decrease_rate": 0.99, # hardcoded since not in use
305 "zone": [zone[0], zone[1], zone[2], zone[3]],
306 "optic_required": optic_required,
307 "coverage_required": coverage_required,
308 "description": description,
309 "sprite": "string", # hardcoded since not in use
310 "secret": secret,
311 }
312 ],
313 "beacon_objectives": [],
314 }
316 d = console_api(method=HttpCode.PUT, endpoint=con.OBJECTIVE_ENDPOINT, json=json)
318 if d:
319 logger.info(f"Console: add/modifyed zoned objective {id}/{name}.")
320 return True
321 else:
322 logger.warning(f"Console: could not add/modifyed zoned objective {id}/{name}")
323 return False
326def add_modify_ebt_objective(
327 id: int,
328 name: str,
329 start: datetime.datetime,
330 end: datetime.datetime,
331 description: str,
332 beacon_height: int,
333 beacon_width: int,
334) -> bool:
335 """Add EBT objectives for testing."""
336 json = {
337 "zoned_objectives": [],
338 "beacon_objectives": [
339 {
340 "id": id,
341 "name": name,
342 "start": start.replace(tzinfo=datetime.timezone.utc).isoformat(),
343 "end": end.replace(tzinfo=datetime.timezone.utc).isoformat(),
344 "decrease_rate": 0.99, # hardcoded since not in use
345 "description": description,
346 "beacon_height": beacon_height,
347 "beacon_width": beacon_width,
348 "attempts_made": 0, # did not change anything in API
349 }
350 ],
351 }
353 d = console_api(method=HttpCode.PUT, endpoint=con.OBJECTIVE_ENDPOINT, json=json)
355 if d:
356 logger.info(f"Console: add/modifyed ebt objective {id}/{name}.")
357 return True
358 else:
359 logger.warning(f"Console: could not add/modifyed ebt objective {id}/{name}")
360 return False
363def send_beacon(beacon_id: int, height: int, width: int) -> Any:
364 """Guess a EBT position."""
365 params = {"beacon_id": beacon_id, "height": height, "width": width}
366 d = console_api(method=HttpCode.PUT, endpoint=con.BEACON_ENDPOINT, params=params)
367 if d:
368 logger.info(f"Console: send_beacon - {d}.")
369 return d
370 else:
371 logger.warning(f"Console: could not send_beacon - {id}")
372 return {}
375def upload_worldmap(image_path: str) -> Any:
376 """Upload a worldmap"""
377 files = {"image": (image_path, open(image_path, "rb"), "image/png")}
378 d = console_api(method=HttpCode.POST, endpoint=con.DAILYMAP_ENDPOINT, files=files)
379 if d:
380 logger.info(f"Console: Uploaded world map - {d}.")
381 # shutil.copyfile(
382 # image_path,
383 # "src/rift_console/static/media/"
384 # + live_utc().strftime("%d-%m-%Y")
385 # + "worldmap.png",
386 # )
387 return d
388 else:
389 logger.warning("Console: could not upload world map")
390 return ""
393def upload_objective(image_path: str, objective_id: int) -> Any:
394 """Upload an images of an objective."""
395 params = {
396 "objective_id": objective_id,
397 }
398 files = {"image": (image_path, open(image_path, "rb"), "image/png")}
399 d = console_api(
400 method=HttpCode.POST, endpoint=con.IMAGE_ENDPOINT, params=params, files=files
401 )
402 if d:
403 logger.info(f"Console: Uploaded objective - {d}.")
404 # shutil.copyfile(
405 # image_path,
406 # con.CONSOLE_STICHED_PATH + str(objective_id) + "objective.png",
407 # )
408 return d
409 else:
410 logger.warning("Console: could not upload objective")
411 return ""