Skip to content

Catena Config

Catena Config Model

Bases: ExtendedBaseModel

Source code in catena/models/config.py 
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
81
82
83
84
85
86
87
88
89
90
91
class CatenaConfig(ExtendedBaseModel):
    version: Optional[str] = 1.0
    clusters: Optional[Dict[str, ClusterDefinition]] = {}


    @classmethod
    def read(cls, path:Optional[str] = None):
        """
        Read configuration from specified yaml file
        """
        if path is None:
            path = Path.home() / ".catena/conf.yml"

        if not Path(path).is_file():
            print(f"⚠️ [red] file does not exist: {path}[/red]")
            exit(1)

        # read manifest
        with open(path, 'r') as f:
            data = safe_loader(f, Loader=Loader)

        return cls(**data)


    def cluster_profiles(self):
        """
        Returns available cluster profiles defined in catena configuration
        file
        """
        return list(self.clusters.keys())


    def get_profile(self, name:str):
        """
        Return configuration properties for the named cluster profile
        """
        if name in self.clusters:
            return self.clusters.get(name)

        print(f"⚠️ [red] profile '{name}' does not exist[/red]")
        print(f"Possiblities include: {self.cluster_profiles()}")

cluster_profiles()

Returns available cluster profiles defined in catena configuration file

Source code in catena/models/config.py 
75
76
77
78
79
80
def cluster_profiles(self):
    """
    Returns available cluster profiles defined in catena configuration
    file
    """
    return list(self.clusters.keys())

get_profile(name)

Return configuration properties for the named cluster profile

Source code in catena/models/config.py 
83
84
85
86
87
88
89
90
91
def get_profile(self, name:str):
    """
    Return configuration properties for the named cluster profile
    """
    if name in self.clusters:
        return self.clusters.get(name)

    print(f"⚠️ [red] profile '{name}' does not exist[/red]")
    print(f"Possiblities include: {self.cluster_profiles()}")

read(path=None) classmethod

Read configuration from specified yaml file

Source code in catena/models/config.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@classmethod
def read(cls, path:Optional[str] = None):
    """
    Read configuration from specified yaml file
    """
    if path is None:
        path = Path.home() / ".catena/conf.yml"

    if not Path(path).is_file():
        print(f"⚠️ [red] file does not exist: {path}[/red]")
        exit(1)

    # read manifest
    with open(path, 'r') as f:
        data = safe_loader(f, Loader=Loader)

    return cls(**data)

Cluster Definition Model

Bases: BaseModel

Source code in catena/models/config.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class ClusterDefinition(BaseModel):
    backend: Optional[str] = 'slurm'

    class Config:
        extra = Extra.allow
        arbitrary_types_allowed = True

    def __init__(self, **kwargs):

        super().__init__(**kwargs)

        cluster_args = {k: v  for k, v in kwargs.items() if k not in self.__fields__}

        if self.backend == 'slurm':
            slurm_cluster = SlurmCluster(**cluster_args)
            for attr, val in slurm_cluster.dict().items():
                self.__dict__[attr] = val

            self._cluster = slurm_cluster

SLURM Cluster Definition Model

Bases: BaseModel

Source code in catena/models/config.py 
12
13
14
15
16
17
18
19
class SlurmCluster(BaseModel):
    api_host: str
    api_proto: Optional[str] = 'http'
    api_version: Optional[str] = '0.0.35'
    api_port: Optional[str] = '6820'

    class Config:
        api_version_compat = ['0.0.35']