AirCompSim/experiment_config.py at main · Anamort/AirCompSim · GitHub

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
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import List, Optional


@dataclass
class ExperimentConfig:
    number_of_users: List[int] = field(default_factory=lambda: [20, 40, 60, 80, 100])
    number_of_servers: List[int] = field(default_factory=lambda: [4])
    number_of_uavs: List[int] = field(default_factory=lambda: [0, 5, 10, 15, 20])
    uav_waiting_policy: List[int] = field(default_factory=lambda: [100])
    uav_radius: List[int] = field(default_factory=lambda: [100])
    uav_fly_policy: List[str] = field(default_factory=lambda: ["LSI"])
    user_mobility_policy: List[str] = field(default_factory=lambda: ["Mobile"])
    repeat_count: int = 50
    simulation_time: float = 1000.0
    seed: Optional[int] = None
    is_drl: bool = False
    is_drl_training: bool = False
    output_dir: Optional[str] = None
    generate_edge_radius_plot: bool = False
    write_root_csvs: bool = False

    @classmethod
    def smoke(cls) -> ExperimentConfig:
        return cls(
            number_of_users=[20],
            number_of_uavs=[0, 5],
            repeat_count=1,
            seed=42,
            generate_edge_radius_plot=True,
        )

    @classmethod
    def paper(cls) -> ExperimentConfig:
        return cls(
            number_of_users=[20, 40, 60, 80, 100],
            number_of_servers=[4],
            number_of_uavs=[0, 5, 10, 15, 20],
            uav_waiting_policy=[100],
            uav_radius=[100],
            uav_fly_policy=["LSI"],
            user_mobility_policy=["Mobile"],
            repeat_count=50,
            simulation_time=1000.0,
            generate_edge_radius_plot=True,
        )

    def total_simulations(self) -> int:
        return (
            len(self.number_of_users)
            * len(self.number_of_servers)
            * len(self.number_of_uavs)
            * len(self.uav_fly_policy)
            * len(self.uav_waiting_policy)
            * len(self.uav_radius)
            * len(self.user_mobility_policy)
            * self.repeat_count
        )

    def ensure_output_dir(self) -> Path:
        if self.output_dir:
            path = Path(self.output_dir)
        else:
            run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
            path = Path("results") / run_id
            self.output_dir = str(path)
        path.mkdir(parents=True, exist_ok=True)
        return path