Coverage for src / lstautorta / config / configuration.py: 100%
49 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-10 11:56 +0000
1from functools import cached_property
2from typing import Annotated
4from annotated_types import Gt
5from pydantic import BaseModel, Field, computed_field, model_validator
7from lstautorta.utils.slurm import parse_slurm_nodes
10class DataStreamConnectionConfiguration(BaseModel):
11 """Parameters to connect to data streamers"""
13 hostname: str = Field(
14 title="hostname",
15 description="Hostname on the network of the data streamer",
16 examples=["localhost", "tcs05-ib0"],
17 )
18 port: int = Field(
19 gt=0, le=65535, title="Port", description="Port on the network of the data streamer", examples=[25000, 3391]
20 )
23class AutoRTAConfiguration(BaseModel):
24 """Parameters of the autoRTA script."""
26 copy_env: bool = Field(
27 title="RAM env copy",
28 description="If true: copies the conda environment and model files to worker nodes RAM at the start of the night",
29 examples=[True],
30 )
31 env_archive: str = Field(
32 title="Environment archive",
33 description="Environment archive to copy to `copy_destination_dir` if `copy_env` is true, not used otherwise.",
34 examples=["/fefs/onsite/pipeline/rta/sag_reco_auto_rta/RTA_Dev_EVBv6_CDB_shm_2024_05_06.tar.gz"],
35 )
36 env_archive_extraction_path: str = Field(
37 title="Environment archive extraction path",
38 description="The directory where the environment archive will be extracted if `copy_env` is true, not used otherwise. "
39 "The extraction is a simple tar -xzf archive.tar.gz -C env_archive_extraction_path command, so the directory must exist.",
40 examples=["/dev/shm/env_folder"],
41 )
42 hiperta_CDB_config_file: str = Field(
43 title="HiPeRTA configuration file",
44 description="Path to hiperta_stream_start CDB configuration (static configuration)",
45 examples=["/fefs/onsite/pipeline/rta/sag_reco_auto_rta/configuration/CDB_configuration_2024_01_31.json"],
46 )
47 ignore_old_observation: bool = Field(
48 title="Ignore old observation",
49 description="If the queried observation has a tstart that is more than 4 hours before current time, do not start r0-dl1. "
50 "This should typically be True in production, and False for day tests.",
51 examples=[True],
52 )
53 log_level: str = Field(
54 title="LST_AUTO_RTA logging level",
55 description="Logging level for the lst auto RTA script. Accept values from python's logging module",
56 examples=["DEBUG", "INFO", "WARNING"],
57 )
58 models_archive_copy_path: str = Field(
59 title="Reconstruction models copy destination",
60 description="The directory where the reco models will be copied if `copy_env` is True, not used otherwise. "
61 'Note: the copy is a simple "cp" command, so a trailing "/" will copy the environment in a subfolder of this path.',
62 examples=["/dev/shm/models_archive/2024_05_17"],
63 )
64 models_archive_path: str = Field(
65 title="Reconstruction models archive path",
66 description='Path to the reconstruction models "archive" (folder containing model with special structure), '
67 "it will be copied to `copy_destination_dir` if `copy_env` is true.",
68 examples=["/fefs/onsite/pipeline/rta/sag_reco_auto_rta/model_archives/2024_05_06"],
69 )
70 stop_time_UTC_hours: int = Field(
71 ge=0,
72 lt=24,
73 title="Stop time hour for auto RTA, in UTC time zone.",
74 description="Auto RTA will shut down when `stop_time_UTC_hours:stop_time_UTC_minutes` is reached",
75 examples=[7],
76 )
77 stop_time_UTC_minutes: int = Field(
78 ge=0,
79 lt=60,
80 title="Stop time minute for auto RTA, in UTC time zone",
81 description="Auto RTA will shut down when stop_time_UTC_hours:stop_time_UTC_minutes is reached",
82 examples=[30],
83 )
84 check_node_connection: bool = Field(
85 title="Nodes connection check",
86 description="If true: do NOT start the RTA if the worker nodes are not connected to infinyband network.",
87 examples=[True],
88 )
89 db_hostname: str = Field(
90 title="LST DB hostname", description="Hostname of the LST DB of observation service data", examples=["lst101"]
91 )
92 data_dir: str = Field(
93 title="Data directory",
94 description="Base directory for RTA output files."
95 " Observations files and logs will be written in an appropriate subfolder",
96 examples=["/fefs/onsite/pipeline/rta/data/"],
97 )
98 slurm_account: str = Field(
99 title="Slurm account", description="Slurm user of the auto_rta slurm commands", examples=["lstrta"]
100 )
101 slurm_reservations: list[str] = Field(
102 title="Slurm reservations",
103 description="List of slurm reservation to use to start RTA jobs (all nodes of all reservations will be used)",
104 examples=[["rta_one_node", "rta_3_nodes_nightly"]],
105 )
106 tel_ids_to_data_servers: dict[Annotated[int, Gt(0)], list[DataStreamConnectionConfiguration]] = Field(
107 title="Streamers per tel ID",
108 description="Map from telescope ID to data servers connections",
109 examples=[{1: [{"hostname": "tcs05-ib0", "port": 25000}, {"hostname": "tcs05-ib0", "port": 25001}]}],
110 )
112 @computed_field
113 @cached_property
114 def slurm_nodes(self) -> dict[str, list[str]]:
115 """List of slurm nodes per reservation.
117 Returns
118 -------
119 Dict[str, List[str]]
120 List of slurm nodes per reservation.
121 """
122 return parse_slurm_nodes(self.slurm_reservations, self.slurm_account)
124 @model_validator(mode="after")
125 def check_enough_slurm_nodes(self) -> "AutoRTAConfiguration":
126 """Check if there are enough slurm nodes (nb_nodes) to start all r0->dl1 jobs (nb_jobs). IE nb_nodes >= nb_jobs
128 Returns
129 -------
130 AutoRTAConfiguration
131 valid AutoRTAConfiguration
133 Raises
134 ------
135 ValueError
136 If the number of slurm nodes is less than the number of data server connections.
137 """
138 nb_slurm_nodes = sum([len(nodes) for nodes in self.slurm_nodes.values()])
139 nb_connections = sum([len(connections) for connections in self.tel_ids_to_data_servers.values()])
140 if nb_slurm_nodes < nb_connections:
141 raise ValueError(
142 f"Number of slurm nodes {nb_slurm_nodes} less than number of r0->dl1 jobs {nb_connections} !\n"
143 "Reminder: nodes in INACTIVE reservation are discarded !"
144 )
145 return self
148class ObservationParameters(BaseModel):
149 """Observation run-time parameters to pass to hiperta_stream_start (dynamic configuration)"""
151 sb_id: int = Field(
152 ge=0,
153 title="Scheduling Block ID",
154 description="ID of the observation's scheduling block.",
155 examples=[12345],
156 )
157 obs_id: int = Field(
158 ge=0,
159 title="Observation ID",
160 description="Id of the observation",
161 examples=[12345],
162 )
163 tel_id: int = Field(
164 ge=0,
165 title="Telescope ID",
166 description="Telescope ID",
167 examples=[1],
168 )
169 RA_pointing: float = Field(
170 ge=0.0,
171 le=360.0,
172 title="Rate Ascension",
173 description="Pointing Rate Ascension during the observation.",
174 examples=[20.0],
175 )
176 DEC_pointing: float = Field(
177 ge=-90.0,
178 le=90.0,
179 title="Declination",
180 description="Pointing declination during the observation",
181 examples=[60.0],
182 )
183 dl1_dir: str = Field(
184 title="DL1 directory",
185 description="Path to the directory where to write the DL1 files.",
186 examples=["/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/DL1"],
187 )
188 dl2_dir: str = Field(
189 title="DL2 directory",
190 description="Path to the directory where to write the DL2 files.",
191 examples=["/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/DL2"],
192 )
193 dl3_dir: str = Field(
194 title="DL3 directory",
195 description="Path to the directory where to write the DL3 files.",
196 examples=["/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/DL3"],
197 )
198 log_dir: str = Field(
199 title="Log directory",
200 description="Path to the directory where to write the log files.",
201 examples=["/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/logs"],
202 )
203 reco_manager_log_file: str = Field(
204 title="hiperta_stream_start log file",
205 description="Log file for reco manager entrypoint (hiperta_stream_start). "
206 "This path must NOT contain any @{} string substitution"
207 "(opening the log file is the first thing reco-manager will do, before substituting strings)",
208 examples=[
209 "/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/hiperta_stream.log",
210 ],
211 )
212 data_stream_connections: list[DataStreamConnectionConfiguration] = Field(
213 title="Stream Connections",
214 description="Parameters of the connections to data streamers",
215 examples=[[{"hostname": "tcs06", "port": 25000}]],
216 )
217 slurm_nodelists: dict[str, list[str]] = Field(
218 title="Slurm Node List",
219 description="List of the slurm nodes to use, per slurm reservation",
220 examples=[{"rta_3_nodes_nightly": ["cp15", "cp16"], "rta-one-node": ["cp19"]}],
221 )