lstautorta.config.configuration

Classes:

Name	Description
AutoRTAConfiguration	Parameters of the autoRTA script.
DataStreamConnectionConfiguration	Parameters to connect to data streamers
ObservationParameters	Observation run-time parameters to pass to hiperta_stream_start (dynamic configuration)

AutoRTAConfiguration

Bases: BaseModel

              flowchart TD
              lstautorta.config.configuration.AutoRTAConfiguration[AutoRTAConfiguration]

              

              click lstautorta.config.configuration.AutoRTAConfiguration href "" "lstautorta.config.configuration.AutoRTAConfiguration"
            

Parameters of the autoRTA script.

Methods:

Name	Description
check_enough_slurm_nodes	Check if there are enough slurm nodes (nb_nodes) to start all r0->dl1 jobs (nb_jobs). IE nb_nodes >= nb_jobs

Attributes:

Name	Type	Description
slurm_nodes	dict[str, list[str]]	List of slurm nodes per reservation.

Source code in src/lstautorta/config/configuration.py

class AutoRTAConfiguration(BaseModel):
    """Parameters of the autoRTA script."""

    copy_env: bool = Field(
        title="RAM env copy",
        description="If true: copies the conda environment and model files to worker nodes RAM at the start of the night",
        examples=[True],
    )
    env_archive: str = Field(
        title="Environment archive",
        description="Environment archive to copy to `copy_destination_dir` if `copy_env` is true, not used otherwise.",
        examples=["/fefs/onsite/pipeline/rta/sag_reco_auto_rta/RTA_Dev_EVBv6_CDB_shm_2024_05_06.tar.gz"],
    )
    env_archive_extraction_path: str = Field(
        title="Environment archive extraction path",
        description="The directory where the environment archive will be extracted if `copy_env` is true, not used otherwise. "
        "The extraction is a simple tar -xzf archive.tar.gz -C env_archive_extraction_path command, so the directory must exist.",
        examples=["/dev/shm/env_folder"],
    )
    hiperta_CDB_config_file: str = Field(
        title="HiPeRTA configuration file",
        description="Path to hiperta_stream_start CDB configuration (static configuration)",
        examples=["/fefs/onsite/pipeline/rta/sag_reco_auto_rta/configuration/CDB_configuration_2024_01_31.json"],
    )
    ignore_old_observation: bool = Field(
        title="Ignore old observation",
        description="If the queried observation has a tstart that is more than 4 hours before current time, do not start r0-dl1. "
        "This should typically be True in production, and False for day tests.",
        examples=[True],
    )
    log_level: str = Field(
        title="LST_AUTO_RTA logging level",
        description="Logging level for the lst auto RTA script. Accept values from python's logging module",
        examples=["DEBUG", "INFO", "WARNING"],
    )
    models_archive_copy_path: str = Field(
        title="Reconstruction models copy destination",
        description="The directory where the reco models will be copied if `copy_env` is True, not used otherwise. "
        'Note: the copy is a simple "cp" command, so a trailing "/" will copy the environment in a subfolder of this path.',
        examples=["/dev/shm/models_archive/2024_05_17"],
    )
    models_archive_path: str = Field(
        title="Reconstruction models archive path",
        description='Path to the reconstruction models "archive" (folder containing model with special structure), '
        "it will be copied to `copy_destination_dir` if `copy_env` is true.",
        examples=["/fefs/onsite/pipeline/rta/sag_reco_auto_rta/model_archives/2024_05_06"],
    )
    stop_time_UTC_hours: int = Field(
        ge=0,
        lt=24,
        title="Stop time hour for auto RTA, in UTC time zone.",
        description="Auto RTA will shut down when `stop_time_UTC_hours:stop_time_UTC_minutes` is reached",
        examples=[7],
    )
    stop_time_UTC_minutes: int = Field(
        ge=0,
        lt=60,
        title="Stop time minute for auto RTA, in UTC time zone",
        description="Auto RTA will shut down when stop_time_UTC_hours:stop_time_UTC_minutes is reached",
        examples=[30],
    )
    check_node_connection: bool = Field(
        title="Nodes connection check",
        description="If true: do NOT start the RTA if the worker nodes are not connected to infinyband network.",
        examples=[True],
    )
    db_hostname: str = Field(
        title="LST DB hostname", description="Hostname of the LST DB of observation service data", examples=["lst101"]
    )
    data_dir: str = Field(
        title="Data directory",
        description="Base directory for RTA output files."
        " Observations files and logs will be written in an appropriate subfolder",
        examples=["/fefs/onsite/pipeline/rta/data/"],
    )
    slurm_account: str = Field(
        title="Slurm account", description="Slurm user of the auto_rta slurm commands", examples=["lstrta"]
    )
    slurm_reservations: list[str] = Field(
        title="Slurm reservations",
        description="List of slurm reservation to use to start RTA jobs (all nodes of all reservations will be used)",
        examples=[["rta_one_node", "rta_3_nodes_nightly"]],
    )
    tel_ids_to_data_servers: dict[Annotated[int, Gt(0)], list[DataStreamConnectionConfiguration]] = Field(
        title="Streamers per tel ID",
        description="Map from telescope ID to data servers connections",
        examples=[{1: [{"hostname": "tcs05-ib0", "port": 25000}, {"hostname": "tcs05-ib0", "port": 25001}]}],
    )

    @computed_field
    @cached_property
    def slurm_nodes(self) -> dict[str, list[str]]:
        """List of slurm nodes per reservation.

        Returns
        -------
        Dict[str, List[str]]
            List of slurm nodes per reservation.
        """
        return parse_slurm_nodes(self.slurm_reservations, self.slurm_account)

    @model_validator(mode="after")
    def check_enough_slurm_nodes(self) -> "AutoRTAConfiguration":
        """Check if there are enough slurm nodes (nb_nodes) to start all r0->dl1 jobs (nb_jobs). IE nb_nodes >= nb_jobs

        Returns
        -------
        AutoRTAConfiguration
            valid AutoRTAConfiguration

        Raises
        ------
        ValueError
            If the number of slurm nodes is less than the number of data server connections.
        """
        nb_slurm_nodes = sum([len(nodes) for nodes in self.slurm_nodes.values()])
        nb_connections = sum([len(connections) for connections in self.tel_ids_to_data_servers.values()])
        if nb_slurm_nodes < nb_connections:
            raise ValueError(
                f"Number of slurm nodes {nb_slurm_nodes} less than number of r0->dl1 jobs {nb_connections} !\n"
                "Reminder: nodes in INACTIVE reservation are discarded !"
            )
        return self

slurm_nodes cached property

slurm_nodes

List of slurm nodes per reservation.

Returns:

Type	Description
Dict[str, List[str]]	List of slurm nodes per reservation.

check_enough_slurm_nodes

check_enough_slurm_nodes()

Check if there are enough slurm nodes (nb_nodes) to start all r0->dl1 jobs (nb_jobs). IE nb_nodes >= nb_jobs

Returns:

Type	Description
AutoRTAConfiguration	valid AutoRTAConfiguration

Raises:

Type	Description
ValueError	If the number of slurm nodes is less than the number of data server connections.

Source code in src/lstautorta/config/configuration.py

@model_validator(mode="after")
def check_enough_slurm_nodes(self) -> "AutoRTAConfiguration":
    """Check if there are enough slurm nodes (nb_nodes) to start all r0->dl1 jobs (nb_jobs). IE nb_nodes >= nb_jobs

    Returns
    -------
    AutoRTAConfiguration
        valid AutoRTAConfiguration

    Raises
    ------
    ValueError
        If the number of slurm nodes is less than the number of data server connections.
    """
    nb_slurm_nodes = sum([len(nodes) for nodes in self.slurm_nodes.values()])
    nb_connections = sum([len(connections) for connections in self.tel_ids_to_data_servers.values()])
    if nb_slurm_nodes < nb_connections:
        raise ValueError(
            f"Number of slurm nodes {nb_slurm_nodes} less than number of r0->dl1 jobs {nb_connections} !\n"
            "Reminder: nodes in INACTIVE reservation are discarded !"
        )
    return self

DataStreamConnectionConfiguration

Bases: BaseModel

              flowchart TD
              lstautorta.config.configuration.DataStreamConnectionConfiguration[DataStreamConnectionConfiguration]

              

              click lstautorta.config.configuration.DataStreamConnectionConfiguration href "" "lstautorta.config.configuration.DataStreamConnectionConfiguration"
            

Parameters to connect to data streamers

Source code in src/lstautorta/config/configuration.py

class DataStreamConnectionConfiguration(BaseModel):
    """Parameters to connect to data streamers"""

    hostname: str = Field(
        title="hostname",
        description="Hostname on the network of the data streamer",
        examples=["localhost", "tcs05-ib0"],
    )
    port: int = Field(
        gt=0, le=65535, title="Port", description="Port on the network of the data streamer", examples=[25000, 3391]
    )

ObservationParameters

Bases: BaseModel

              flowchart TD
              lstautorta.config.configuration.ObservationParameters[ObservationParameters]

              

              click lstautorta.config.configuration.ObservationParameters href "" "lstautorta.config.configuration.ObservationParameters"
            

Observation run-time parameters to pass to hiperta_stream_start (dynamic configuration)

Source code in src/lstautorta/config/configuration.py

class ObservationParameters(BaseModel):
    """Observation run-time parameters to pass to hiperta_stream_start (dynamic configuration)"""

    sb_id: int = Field(
        ge=0,
        title="Scheduling Block ID",
        description="ID of the observation's scheduling block.",
        examples=[12345],
    )
    obs_id: int = Field(
        ge=0,
        title="Observation ID",
        description="Id of the observation",
        examples=[12345],
    )
    tel_id: int = Field(
        ge=0,
        title="Telescope ID",
        description="Telescope ID",
        examples=[1],
    )
    RA_pointing: float = Field(
        ge=0.0,
        le=360.0,
        title="Rate Ascension",
        description="Pointing Rate Ascension during the observation.",
        examples=[20.0],
    )
    DEC_pointing: float = Field(
        ge=-90.0,
        le=90.0,
        title="Declination",
        description="Pointing declination during the observation",
        examples=[60.0],
    )
    dl1_dir: str = Field(
        title="DL1 directory",
        description="Path to the directory where to write the DL1 files.",
        examples=["/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/DL1"],
    )
    dl2_dir: str = Field(
        title="DL2 directory",
        description="Path to the directory where to write the DL2 files.",
        examples=["/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/DL2"],
    )
    dl3_dir: str = Field(
        title="DL3 directory",
        description="Path to the directory where to write the DL3 files.",
        examples=["/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/DL3"],
    )
    log_dir: str = Field(
        title="Log directory",
        description="Path to the directory where to write the log files.",
        examples=["/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/logs"],
    )
    reco_manager_log_file: str = Field(
        title="hiperta_stream_start log file",
        description="Log file for reco manager entrypoint (hiperta_stream_start). "
        "This path must NOT contain any @{} string substitution"
        "(opening the log file is the first thing reco-manager will do, before substituting strings)",
        examples=[
            "/fefs/onsite/pipeline/rta/data/YYYY/MM/DD/hiperta_stream.log",
        ],
    )
    data_stream_connections: list[DataStreamConnectionConfiguration] = Field(
        title="Stream Connections",
        description="Parameters of the connections to data streamers",
        examples=[[{"hostname": "tcs06", "port": 25000}]],
    )
    slurm_nodelists: dict[str, list[str]] = Field(
        title="Slurm Node List",
        description="List of the slurm nodes to use, per slurm reservation",
        examples=[{"rta_3_nodes_nightly": ["cp15", "cp16"], "rta-one-node": ["cp19"]}],
    )