Skip to content

Shared Reference

Ciarc.

constants

ACCELERATION = 0.02 module-attribute

ACHIEVEMENTS_ENDPOINT = f'{BASE_URL}achievements' module-attribute

ANNOUNCEMENTS_ENDPOINT = f'{BASE_URL}announcements' module-attribute

BACKUP_ENDPOINT = f'{BASE_URL}backup' module-attribute

BASE_URL = 'http://10.100.10.11:33000/' module-attribute

BEACON_ENDPOINT = f'{BASE_URL}beacon' module-attribute

CONSOLE_DOWNLOAD_PATH = 'logs/rift_console/images/download/' module-attribute

CONSOLE_EBT_PATH = 'logs/rift_console/images/ebt/' module-attribute

CONSOLE_FROM_MELVONAUT_PATH = 'logs/rift_console/from_melvonaut/' module-attribute

CONSOLE_IMAGE_VIEWER_LIMIT = 1000 module-attribute

CONSOLE_LIVE_PATH = 'logs/rift_console/images/live/' module-attribute

CONSOLE_LOG_PATH = 'logs/rift_console/' module-attribute

CONSOLE_STICHED_PATH = 'logs/rift_console/images/stitched/' module-attribute

CONTROL_ENDPOINT = f'{BASE_URL}control' module-attribute

DAILYMAP_ENDPOINT = f'{BASE_URL}dailyMap' module-attribute

EVENT_LOCATION_CSV = 'logs/melvonaut/event_melvonaut.csv' module-attribute

IMAGE_ANGLE_POSITION = 2 module-attribute

IMAGE_ENDPOINT = f'{BASE_URL}image' module-attribute

IMAGE_LOCATION = IMAGE_PATH + 'image_{melv_id}_{angle}_{time}_x_{cor_x}_y_{cor_y}.png' module-attribute

IMAGE_NAME_UNDERSCORE_COUNT = 8 module-attribute

IMAGE_NOISE_FORGIVENESS = 20 module-attribute

IMAGE_PATH = 'logs/melvonaut/images/' module-attribute

IMAGE_PATH_BASE = 'logs/melvonaut/images/' module-attribute

MEL_LOG_FORMAT = 'log_melvonaut_{time:YYYY-MM-DD_HH}.log' module-attribute

MEL_LOG_LOCATION = MEL_LOG_PATH + MEL_LOG_FORMAT module-attribute

MEL_LOG_PATH = 'logs/melvonaut/' module-attribute

MEL_PERSISTENT_SETTINGS = 'logs/melvonaut/persistent_settings.json' module-attribute

NUMBER_OF_WORKER_THREADS = cpu_count() or 4 - 2 module-attribute

OBJECTIVE_ENDPOINT = f'{BASE_URL}objective' module-attribute

OBSERVATION_ENDPOINT = f'{BASE_URL}observation' module-attribute

PANORAMA_PATH = 'media/' module-attribute

RESET_ENDPOINT = f'{BASE_URL}reset' module-attribute

RIFT_LOG_LEVEL = 'INFO' module-attribute

RIFT_LOG_LOCATION = 'logs/rift_console/log_rift-console_{time:YYYY-MM-DD_HH}.log' module-attribute

SAVE_PANORAMA_STEP = 1000 module-attribute

SEARCH_GRID_SIDE_LENGTH = 15 module-attribute

SIMULATION_ENDPOINT = f'{BASE_URL}simulation' module-attribute

SLOTS_ENDPOINT = f'{BASE_URL}slots' module-attribute

SORT_IMAGE_BY_POSITION = True module-attribute

STATE_TRANSITION_FROM_SAFE_TIME = 20 * 60 module-attribute

STATE_TRANSITION_TIME = 3 * 60 module-attribute

STATE_TRANSITION_TO_SAFE_TIME = 1 * 60 module-attribute

STITCHING_BORDER = 1000 module-attribute

STITCHING_COUNT_LIMIT = 5000 module-attribute

TELEMETRY_LOCATION_CSV = 'logs/melvonaut/telemetry_melvonaut.csv' module-attribute

TELEMETRY_LOCATION_JSON = 'logs/melvonaut/telemetry_melvonaut.json' module-attribute

THUMBNAIL_X = 1000 module-attribute

THUMBNAIL_Y = 500 module-attribute

TRAJ_STEP = 10 module-attribute

TRAJ_TIME = 3600 * 12 module-attribute

USE_LEGACY_IMAGE_NAMES = False module-attribute

WORLD_X = 21600 module-attribute

WORLD_Y = 10800 module-attribute

models

Achievement

Bases: BaseModel

From CIARC API.

Source code in src/shared/models.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class Achievement(BaseModel):
    """
    From CIARC API.
    """

    name: str
    done: bool
    points: int
    description: str
    goal_parameter_threshold: Union[bool, int, float, str]
    goal_parameter: Union[bool, int, float, str]

    @staticmethod
    def parse_api(data: dict) -> list["Achievement"]:  # type: ignore
        """
        Parse CIARC API into list of Achievment.
        """
        achv = []
        for a in data["achievements"]:
            achv.append(Achievement(**a))

        return achv

description instance-attribute

done instance-attribute

goal_parameter instance-attribute

goal_parameter_threshold instance-attribute

name instance-attribute

points instance-attribute

parse_api(data) staticmethod

Parse CIARC API into list of Achievment.

Source code in src/shared/models.py 
173
174
175
176
177
178
179
180
181
182
@staticmethod
def parse_api(data: dict) -> list["Achievement"]:  # type: ignore
    """
    Parse CIARC API into list of Achievment.
    """
    achv = []
    for a in data["achievements"]:
        achv.append(Achievement(**a))

    return achv

BaseTelemetry

Bases: BaseModel

Based on /observation endpoint.

Source code in src/shared/models.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
class BaseTelemetry(BaseModel):
    """Based on /observation endpoint."""

    model_config = ConfigDict(use_enum_values=True)

    class AreaCovered(BaseModel):
        narrow: float
        normal: float
        wide: float

    class DataVolume(BaseModel):
        data_volume_received: int
        data_volume_sent: int

    active_time: float
    angle: CameraAngle
    area_covered: AreaCovered
    battery: float
    data_volume: DataVolume
    distance_covered: float
    fuel: float
    width_x: int
    height_y: int
    images_taken: int
    max_battery: float
    objectives_done: int
    objectives_points: int
    simulation_speed: int
    state: State
    timestamp: datetime.datetime
    vx: float
    vy: float

    def __str__(self) -> str:
        return (
            f"Telemetry@{self.timestamp.isoformat()} state={self.state} angle={self.angle} "
            f"(x,y)=({self.width_x},{self.height_y}) (vx,vy)=({self.vx},{self.vy}) "
            f"battery={self.battery}/{self.max_battery} fuel={self.fuel} sim_speed={self.simulation_speed} "
            f"dist_cov={self.distance_covered} area_cov={self.area_covered.narrow}/{self.area_covered.normal}/{self.area_covered.wide} "
            f"active_t={self.active_time} #images={self.images_taken} obj-done/points={self.objectives_done}/{self.objectives_points} "
            f"data-s/r={self.data_volume.data_volume_sent}/{self.data_volume.data_volume_received}"
        )

active_time instance-attribute

angle instance-attribute

area_covered instance-attribute

battery instance-attribute

data_volume instance-attribute

distance_covered instance-attribute

fuel instance-attribute

height_y instance-attribute

images_taken instance-attribute

max_battery instance-attribute

model_config = ConfigDict(use_enum_values=True) class-attribute instance-attribute

objectives_done instance-attribute

objectives_points instance-attribute

simulation_speed instance-attribute

state instance-attribute

timestamp instance-attribute

vx instance-attribute

vy instance-attribute

width_x instance-attribute

AreaCovered

Bases: BaseModel

Source code in src/shared/models.py 
200
201
202
203
class AreaCovered(BaseModel):
    narrow: float
    normal: float
    wide: float
narrow instance-attribute
normal instance-attribute
wide instance-attribute

DataVolume

Bases: BaseModel

Source code in src/shared/models.py 
205
206
207
class DataVolume(BaseModel):
    data_volume_received: int
    data_volume_sent: int
data_volume_received instance-attribute
data_volume_sent instance-attribute

__str__()

Source code in src/shared/models.py 
228
229
230
231
232
233
234
235
236
def __str__(self) -> str:
    return (
        f"Telemetry@{self.timestamp.isoformat()} state={self.state} angle={self.angle} "
        f"(x,y)=({self.width_x},{self.height_y}) (vx,vy)=({self.vx},{self.vy}) "
        f"battery={self.battery}/{self.max_battery} fuel={self.fuel} sim_speed={self.simulation_speed} "
        f"dist_cov={self.distance_covered} area_cov={self.area_covered.narrow}/{self.area_covered.normal}/{self.area_covered.wide} "
        f"active_t={self.active_time} #images={self.images_taken} obj-done/points={self.objectives_done}/{self.objectives_points} "
        f"data-s/r={self.data_volume.data_volume_sent}/{self.data_volume.data_volume_received}"
    )

BeaconObjective

Bases: BaseModel

Emergency beacon objective from CIARC API.

Source code in src/shared/models.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class BeaconObjective(BaseModel):
    """
    Emergency beacon objective from CIARC API.
    """

    id: int
    name: str
    start: datetime.datetime
    end: datetime.datetime
    decrease_rate: float
    attempts_made: int
    description: str

    @staticmethod
    def parse_api(data: dict) -> list["BeaconObjective"]:  # type: ignore
        """
        Parse CIARC API to list of this class
        """
        beacon_obj = []
        for b in data["beacon_objectives"]:
            beacon_obj.append(BeaconObjective(**b))

        return sorted(beacon_obj, key=lambda event: event.start)

attempts_made instance-attribute

decrease_rate instance-attribute

description instance-attribute

end instance-attribute

id instance-attribute

name instance-attribute

start instance-attribute

parse_api(data) staticmethod

Parse CIARC API to list of this class

Source code in src/shared/models.py 
149
150
151
152
153
154
155
156
157
158
@staticmethod
def parse_api(data: dict) -> list["BeaconObjective"]:  # type: ignore
    """
    Parse CIARC API to list of this class
    """
    beacon_obj = []
    for b in data["beacon_objectives"]:
        beacon_obj.append(BeaconObjective(**b))

    return sorted(beacon_obj, key=lambda event: event.start)

CameraAngle

Bases: StrEnum

Different camera angles possible on MELVIN.

Source code in src/shared/models.py 
23
24
25
26
27
28
29
30
31
class CameraAngle(StrEnum):
    """
    Different camera angles possible on MELVIN.
    """

    Wide = "wide"
    Narrow = "narrow"
    Normal = "normal"
    Unknown = "unknown"

Narrow = 'narrow' class-attribute instance-attribute

Normal = 'normal' class-attribute instance-attribute

Unknown = 'unknown' class-attribute instance-attribute

Wide = 'wide' class-attribute instance-attribute

Event

Bases: BaseModel

Message by /announcements, includes time and position for ebt processing.

Source code in src/shared/models.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
class Event(BaseModel):
    """Message by /announcements, includes time and position for ebt processing."""

    event: str
    id: int
    timestamp: Optional[datetime.datetime] = None
    current_x: Optional[float] = None
    current_y: Optional[float] = None

    def __str__(self) -> str:
        return f"Event: {self.event} (x,y)=({self.current_x},{self.current_y}) t={time_seconds(self.timestamp or live_utc())}"

    def easy_parse(self) -> tuple[float, float, float]:
        """Custom parsing wrapper for ebt calculation."""
        pattern = r"DISTANCE_(\d+\.\d+)"
        dist = re.findall(pattern, self.event)[0]
        if dist and self.current_x and self.current_y:
            return (float(dist), self.current_x, self.current_y)
        else:
            logger.warning(f"Tried to parse incomplete event: {self}")
            return (0.0, 0.0, 0.0)

    async def to_csv(self) -> None:
        """Melvonaut saves events."""
        event_dict = self.model_dump()
        if self.timestamp:
            event_dict["timestamp"] = self.timestamp.isoformat()
        if not Path(con.EVENT_LOCATION_CSV).is_file():
            async with async_open(con.EVENT_LOCATION_CSV, "w") as afp:
                writer = csv.DictWriter(afp, fieldnames=event_dict.keys())
                await writer.writeheader()
                await writer.writerow(event_dict)
            # logger.debug(f"Writing event to {con.EVENT_LOCATION_CSV}")
        else:
            async with async_open(con.EVENT_LOCATION_CSV, "a") as afp:
                writer = csv.DictWriter(afp, fieldnames=event_dict.keys())
                await writer.writerow(event_dict)
            # logger.debug(f"Writing event to {con.EVENT_LOCATION_CSV}")

    @staticmethod
    def load_events_from_csv(path: str) -> list["Event"]:
        """Melvonaut saves events as csv, Rift-console loads them."""
        events = []
        if not Path(path).is_file():
            logger.warning(f"No event file found under {path}")
        else:
            with open(path, "r") as f:
                for row in csv.DictReader(f):
                    read_event = Event(
                        event=row["event"],
                        id=int(row["id"]),
                        timestamp=datetime.datetime.fromisoformat(row["timestamp"]),
                        current_x=float(row["current_x"]),
                        current_y=float(row["current_y"]),
                    )
                    events.append(read_event)
            logger.info(f"Loaded {len(events)} events from {path}")
        return events

current_x = None class-attribute instance-attribute

current_y = None class-attribute instance-attribute

event instance-attribute

id instance-attribute

timestamp = None class-attribute instance-attribute

__str__()

Source code in src/shared/models.py 
406
407
def __str__(self) -> str:
    return f"Event: {self.event} (x,y)=({self.current_x},{self.current_y}) t={time_seconds(self.timestamp or live_utc())}"

easy_parse()

Custom parsing wrapper for ebt calculation.

Source code in src/shared/models.py 
409
410
411
412
413
414
415
416
417
def easy_parse(self) -> tuple[float, float, float]:
    """Custom parsing wrapper for ebt calculation."""
    pattern = r"DISTANCE_(\d+\.\d+)"
    dist = re.findall(pattern, self.event)[0]
    if dist and self.current_x and self.current_y:
        return (float(dist), self.current_x, self.current_y)
    else:
        logger.warning(f"Tried to parse incomplete event: {self}")
        return (0.0, 0.0, 0.0)

load_events_from_csv(path) staticmethod

Melvonaut saves events as csv, Rift-console loads them.

Source code in src/shared/models.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
@staticmethod
def load_events_from_csv(path: str) -> list["Event"]:
    """Melvonaut saves events as csv, Rift-console loads them."""
    events = []
    if not Path(path).is_file():
        logger.warning(f"No event file found under {path}")
    else:
        with open(path, "r") as f:
            for row in csv.DictReader(f):
                read_event = Event(
                    event=row["event"],
                    id=int(row["id"]),
                    timestamp=datetime.datetime.fromisoformat(row["timestamp"]),
                    current_x=float(row["current_x"]),
                    current_y=float(row["current_y"]),
                )
                events.append(read_event)
        logger.info(f"Loaded {len(events)} events from {path}")
    return events

to_csv() async

Melvonaut saves events.

Source code in src/shared/models.py 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
async def to_csv(self) -> None:
    """Melvonaut saves events."""
    event_dict = self.model_dump()
    if self.timestamp:
        event_dict["timestamp"] = self.timestamp.isoformat()
    if not Path(con.EVENT_LOCATION_CSV).is_file():
        async with async_open(con.EVENT_LOCATION_CSV, "w") as afp:
            writer = csv.DictWriter(afp, fieldnames=event_dict.keys())
            await writer.writeheader()
            await writer.writerow(event_dict)
        # logger.debug(f"Writing event to {con.EVENT_LOCATION_CSV}")
    else:
        async with async_open(con.EVENT_LOCATION_CSV, "a") as afp:
            writer = csv.DictWriter(afp, fieldnames=event_dict.keys())
            await writer.writerow(event_dict)

HttpCode

Bases: Enum

Used HTTP codes for API.

Source code in src/shared/models.py 
185
186
187
188
189
190
191
class HttpCode(Enum):
    """Used HTTP codes for API."""

    GET = "get"
    PUT = "put"
    DELETE = "delete"
    POST = "post"

DELETE = 'delete' class-attribute instance-attribute

GET = 'get' class-attribute instance-attribute

POST = 'post' class-attribute instance-attribute

PUT = 'put' class-attribute instance-attribute

MELVINTask

Bases: StrEnum

Our custom programs/missions/states in which we can place Melvin. In evaluation phase only mapping and ebt was used. The other two were used in Phase 2, or could be used in a future update.

Source code in src/shared/models.py 
240
241
242
243
244
245
246
247
248
249
250
class MELVINTask(StrEnum):
    """
    Our custom programs/missions/states in which we can place Melvin.
    In evaluation phase only mapping and ebt was used.
    The other two were used in Phase 2, or could be used in a future update.
    """

    Mapping = "mapping"
    Next_objective = "next_objective"
    Fixed_objective = "fixed_objective"
    EBT = "ebt"

EBT = 'ebt' class-attribute instance-attribute

Fixed_objective = 'fixed_objective' class-attribute instance-attribute

Mapping = 'mapping' class-attribute instance-attribute

Next_objective = 'next_objective' class-attribute instance-attribute

MelvinImage

Bases: BaseModel

Our format for a single image taken by MELVIN.

Source code in src/shared/models.py 
371
372
373
374
375
376
377
378
379
380
class MelvinImage(BaseModel):
    """Our format for a single image taken by MELVIN."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    image: Image.Image
    angle: CameraAngle
    cor_x: int
    cor_y: int
    time: datetime.datetime

angle instance-attribute

cor_x instance-attribute

cor_y instance-attribute

image instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

time instance-attribute

Ping

Part of EBT objective, one single distance/ping.

Source code in src/shared/models.py 
383
384
385
386
387
388
389
390
391
392
393
394
class Ping:
    """Part of EBT objective, one single distance/ping."""

    def __init__(self, x: int, y: int, d: float, mind: int, maxd: int):
        self.x = x
        self.y = y
        self.d = d
        self.mind = mind
        self.maxd = maxd

    def __str__(self) -> str:
        return f"Ping: x={self.x}, y={self.y}, d={self.d}, mind={self.mind}, maxd={self.maxd}"

d = d instance-attribute

maxd = maxd instance-attribute

mind = mind instance-attribute

x = x instance-attribute

y = y instance-attribute

__init__(x, y, d, mind, maxd)

Source code in src/shared/models.py 
386
387
388
389
390
391
def __init__(self, x: int, y: int, d: float, mind: int, maxd: int):
    self.x = x
    self.y = y
    self.d = d
    self.mind = mind
    self.maxd = maxd

__str__()

Source code in src/shared/models.py 
393
394
def __str__(self) -> str:
    return f"Ping: x={self.x}, y={self.y}, d={self.d}, mind={self.mind}, maxd={self.maxd}"

Slot

Bases: BaseModel

One communication slot in which MELVIN can be contacted.

Methods:

Name Description
parse_api

dict) -> tuple[int, list["Slot"]]: Parses the given API data to extract slots and the number of slots used.
Source code in src/shared/models.py 
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
class Slot(BaseModel):
    """
    One communication slot in which MELVIN can be contacted.

    Methods:
        parse_api(data: dict) -> tuple[int, list["Slot"]]:
            Parses the given API data to extract slots and the number of slots used.
    """

    id: int
    start: datetime.datetime
    end: datetime.datetime
    enabled: bool

    @staticmethod
    def parse_api(data: dict) -> tuple[int, list["Slot"]]:  # type: ignore
        """
        Parses CIARC API response into the list of available slots.

        Args:
            data (dict): The API response from /slots.

        Returns:
            tuple[int, list["Slot"]]: Number of communication slots used and
                list of slots sorted by the earliest start time.

        """
        slots_used = data["communication_slots_used"]
        slots = []
        for s in data["slots"]:
            slots.append(Slot(**s))

        slots.sort(key=lambda slot: slot.start)
        # logger.debug(f"Deparsed Slot API used: {slots_used} - {slots}")
        return (slots_used, slots)

enabled instance-attribute

end instance-attribute

id instance-attribute

start instance-attribute

parse_api(data) staticmethod

Parses CIARC API response into the list of available slots.

Parameters:

Name Type Description Default
data dict

The API response from /slots.

 required

Returns:

Type Description
tuple[int, list[Slot]]

tuple[int, list["Slot"]]: Number of communication slots used and list of slots sorted by the earliest start time.
Source code in src/shared/models.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@staticmethod
def parse_api(data: dict) -> tuple[int, list["Slot"]]:  # type: ignore
    """
    Parses CIARC API response into the list of available slots.

    Args:
        data (dict): The API response from /slots.

    Returns:
        tuple[int, list["Slot"]]: Number of communication slots used and
            list of slots sorted by the earliest start time.

    """
    slots_used = data["communication_slots_used"]
    slots = []
    for s in data["slots"]:
        slots.append(Slot(**s))

    slots.sort(key=lambda slot: slot.start)
    # logger.debug(f"Deparsed Slot API used: {slots_used} - {slots}")
    return (slots_used, slots)

State

Bases: StrEnum

From CIARC user manual

Source code in src/shared/models.py 
34
35
36
37
38
39
40
41
42
43
class State(StrEnum):
    """From CIARC user manual"""

    Deployment = "deployment"
    Acquisition = "acquisition"
    Charge = "charge"
    Safe = "safe"
    Communication = "communication"
    Transition = "transition"
    Unknown = "none"

Acquisition = 'acquisition' class-attribute instance-attribute

Charge = 'charge' class-attribute instance-attribute

Communication = 'communication' class-attribute instance-attribute

Deployment = 'deployment' class-attribute instance-attribute

Safe = 'safe' class-attribute instance-attribute

Transition = 'transition' class-attribute instance-attribute

Unknown = 'none' class-attribute instance-attribute

Timer

Bases: BaseModel

Starts tasks after a given intervall. E.g. take the next picture X-seconds after the current one.

Source code in src/shared/models.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
class Timer(BaseModel):
    """Starts tasks after a given intervall. E.g. take the next picture X-seconds after the current one."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    _timeout: float
    _callback: Callable[[], Awaitable[Any]]
    _task: asyncio.Task[None]

    def __init__(self, timeout: float, callback: Callable[[], Awaitable[Any]]):
        super().__init__()
        self._timeout = timeout
        self._callback = callback
        self._task = asyncio.create_task(self._job())

    async def _job(self) -> None:
        await asyncio.sleep(self._timeout)
        await self._callback()

    def cancel(self) -> None:
        self._task.cancel()

    def get_task(self) -> asyncio.Task[None]:
        return self._task

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

__init__(timeout, callback)

Source code in src/shared/models.py 
354
355
356
357
358
def __init__(self, timeout: float, callback: Callable[[], Awaitable[Any]]):
    super().__init__()
    self._timeout = timeout
    self._callback = callback
    self._task = asyncio.create_task(self._job())

cancel()

Source code in src/shared/models.py 
364
365
def cancel(self) -> None:
    self._task.cancel()

get_task()

Source code in src/shared/models.py 
367
368
def get_task(self) -> asyncio.Task[None]:
    return self._task

ZonedObjective

Bases: BaseModel

One hidden or visible objective, completed by taking pictures of its position.

Source code in src/shared/models.py 
 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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class ZonedObjective(BaseModel):
    """
    One hidden or visible objective, completed by taking pictures of its position.
    """

    id: int  # could be null acording to Dto
    name: str
    start: datetime.datetime
    end: datetime.datetime
    decrease_rate: float
    zone: Optional[tuple[int, int, int, int]]  # could be a str acording to dto
    optic_required: CameraAngle  # cast from str
    coverage_required: float
    description: str  # cast from str
    secret: bool
    # sprite is ignored as said in email

    @staticmethod
    def parse_api(data: dict) -> list["ZonedObjective"]:  # type: ignore
        """
        Extracts and parses objectives from its matching api endpoint
        """
        z_obj_list = []
        # parse objective list
        for obj in data["zoned_objectives"]:
            if type(obj["zone"]) is str:
                zone = None
            else:
                zone = (
                    int(obj["zone"][0]),
                    int(obj["zone"][1]),
                    int(obj["zone"][2]),
                    int(obj["zone"][3]),
                )

            z_obj_list.append(
                ZonedObjective(
                    id=obj["id"],
                    name=obj["name"],
                    start=datetime.datetime.fromisoformat(obj["start"]),
                    end=datetime.datetime.fromisoformat(obj["end"]),
                    decrease_rate=obj["decrease_rate"],
                    zone=zone,
                    optic_required=CameraAngle(obj["optic_required"]),
                    coverage_required=obj["coverage_required"],
                    description=obj["description"],
                    secret=obj["secret"],
                )
            )

        return sorted(z_obj_list, key=lambda event: event.start)

coverage_required instance-attribute

decrease_rate instance-attribute

description instance-attribute

end instance-attribute

id instance-attribute

name instance-attribute

optic_required instance-attribute

secret instance-attribute

start instance-attribute

zone instance-attribute

parse_api(data) staticmethod

Extracts and parses objectives from its matching api endpoint

Source code in src/shared/models.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@staticmethod
def parse_api(data: dict) -> list["ZonedObjective"]:  # type: ignore
    """
    Extracts and parses objectives from its matching api endpoint
    """
    z_obj_list = []
    # parse objective list
    for obj in data["zoned_objectives"]:
        if type(obj["zone"]) is str:
            zone = None
        else:
            zone = (
                int(obj["zone"][0]),
                int(obj["zone"][1]),
                int(obj["zone"][2]),
                int(obj["zone"][3]),
            )

        z_obj_list.append(
            ZonedObjective(
                id=obj["id"],
                name=obj["name"],
                start=datetime.datetime.fromisoformat(obj["start"]),
                end=datetime.datetime.fromisoformat(obj["end"]),
                decrease_rate=obj["decrease_rate"],
                zone=zone,
                optic_required=CameraAngle(obj["optic_required"]),
                coverage_required=obj["coverage_required"],
                description=obj["description"],
                secret=obj["secret"],
            )
        )

    return sorted(z_obj_list, key=lambda event: event.start)

lens_size_by_angle(angle)

Returns covered area by a single picture.

Source code in src/shared/models.py 
295
296
297
298
299
300
301
302
303
304
305
306
def lens_size_by_angle(angle: CameraAngle) -> int:
    """
    Returns covered area by a single picture.
    """
    match angle:
        case CameraAngle.Narrow:
            lens_size = 600
        case CameraAngle.Normal:
            lens_size = 800
        case CameraAngle.Wide:
            lens_size = 1000
    return lens_size

limited_log(message)

Log limit for info

Source code in src/shared/models.py 
332
333
334
335
@log_rate_limiter(3)  # type: ignore
def limited_log(message: str) -> None:
    """Log limit for info"""
    logger.info(message)

limited_log_debug(message)

Log limit for debug

Source code in src/shared/models.py 
339
340
341
342
@log_rate_limiter(1)  # type: ignore
def limited_log_debug(message: str) -> None:
    """Log limit for debug"""
    logger.debug(message)

live_utc()

Returns live datetime object, including timezone utc

Source code in src/shared/models.py 
469
470
471
def live_utc() -> datetime.datetime:
    """Returns live datetime object, including timezone utc"""
    return datetime.datetime.now(datetime.timezone.utc)

log_rate_limiter(interval_seconds)

Limits how often a single event can trigger a lot entry. Prevents cluttering of the same message. Probaly not a "good" final solution.

Source code in src/shared/models.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def log_rate_limiter(interval_seconds: int):  # type: ignore
    """
    Limits how often a single event can trigger a lot entry. Prevents cluttering of the same message.
    Probaly not a "good" final solution.
    """

    # habe luhki nach loguru log rate limiter gefragt, gibt anscheinend keine besser inbuild lösung
    def decorator(func):  # type: ignore
        last_log_time = [0]  # Use a list to allow modification of non-local state

        def wrapper(*args, **kwargs):  # type: ignore
            nonlocal last_log_time
            current_time = time.time()
            if current_time - last_log_time[0] >= interval_seconds:
                func(*args, **kwargs)
                last_log_time[0] = current_time  # type: ignore

        return wrapper

    return decorator

time_seconds(date)

Source code in src/shared/models.py 
474
475
def time_seconds(date: datetime.datetime) -> str:
    return date.strftime("%Y-%m-%dT%H:%M:%S")