Skip to content

App state

This app has state for different components placed into

You may notice that all names are postpended with Data. Here, we use classes to create them, but create seperate classes with methods. This makes API documentation cleaner and the namespaces more Pydantic-like.

State#

State is used to hold certain app-wide state variables that control the app itself.

API#

Holds app-wide state variables

Attributes:

Name Type Description
mapping Reactive[DataFrame]

bitmappings dataframe, used for targeting filters.
datamodel Reactive[DataFrame]

column glossary dataframe
df Reactive[DataFrame]

loaded dataframe file, defaults to nothing but app will instantiate with one loaded.
columns Reactive[dict[str, list[str]]]

column lookup, used for guardrailing.
release str

release type
datatype str

datatype of file
Source code in src/sdss_explorer/dashboard/dataclass/state.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
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
class StateData:
    """Holds app-wide state variables

    Attributes:
        mapping (solara.Reactive[vx.DataFrame]): bitmappings dataframe, used for targeting filters.
        datamodel (solara.Reactive[pd.DataFrame]): column glossary dataframe

        df (solara.Reactive[vx.DataFrame]): loaded dataframe file, defaults to nothing but app will instantiate with one loaded.
        columns (solara.Reactive[dict[str,list[str]]]): column lookup, used for guardrailing.

        release (str): release type
        datatype (str): datatype of file
    """

    def __init__(self):
        # app settings, underscored to hide prop
        self._release = sl.reactive(cast(str, None))  # TODO: dr19
        self._datatype = sl.reactive(cast(str, None))

        # globally shared, read only files
        self.mapping = sl.reactive(
            open_file("mappings.parquet"))  # mappings for bitmasks
        self.datamodel = sl.reactive(load_datamodel(self.release))  # datamodel spec

        # adaptively rerendered on changes; set on startup in app root
        self.df = sl.reactive(cast(vx.DataFrame, None))  # main datafile
        self.columns = sl.reactive(cast(
            dict, None))  # column glossary for guardrailing

        # user-binded instances
        # NOTE: this approach allows UUID + subsetstore to be read-only
        self._uuid = sl.reactive(sl.get_session_id())
        self._kernel_id = sl.reactive(sl.get_kernel_id())
        self._subset_store = sl.reactive(SubsetStore())

    def load_dataset(self,
                     release: Optional[str] = None,
                     datatype: Optional[str] = None) -> bool:
        """ load the HDF5 dataset for the dashboard """
        # use attributes if not manually overridden
        if not release:
            release = self.release
        if not datatype:
            datatype = self.datatype

        # start with standard open operation
        # TODO: redux version via envvar?
        df = open_file(
            f"{release}/explorerAll{datatype.capitalize()}-{VASTRA}.hdf5")
        columns = load_column_json(release, datatype)

        if (df is None) and (columns is None):
            logger.critical(
                "Part of dataset load failed! ensure everything is setup (files, envvars)"
            )
            return False

        # set reactives
        self.df.set(df)
        self.columns.set(columns)

        return True

    @property
    def release(self) -> str:
        """Current release of app (dr19, etc)"""
        return str(self._release.value)

    @property
    def datatype(self) -> str:
        """Current datatype of app (star or visit)"""
        return str(self._datatype.value)

    def get_default_dataset(self) -> str:
        """Method version to get the default dataset of app (star or visit). Used for defaulting the Subset dataclass"""
        datatype = self._datatype.value
        return "mwmlite" if datatype == "star" else "thepayne"

    @property
    def uuid(self) -> str:
        """User ID; Solara Session ID"""
        return str(self._uuid.value)

    @property
    def kernel_id(self) -> str:
        """Virtual kernel ID"""
        return str(self._kernel_id.value)

    @property
    def subset_store(self):
        """Internal subset backend"""
        return self._subset_store.value

    def __repr__(self) -> str:
        """Show relevant properties of class as string."""
        return "\n".join(
            f"{k:15}: {v}" for k, v in {
                "uuid": self.uuid,
                "kernel_id": self.kernel_id,
                "df": hex(id(self.df.value)),  # dataframe mem address
                "subset_backend": hex(id(self.subset_store)),
                "release": self.release,
                "datatype": self.datatype,
            }.items())

datatype property #

Current datatype of app (star or visit)

kernel_id property #

Virtual kernel ID

release property #

Current release of app (dr19, etc)

subset_store property #

Internal subset backend

uuid property #

User ID; Solara Session ID

__repr__() #

Show relevant properties of class as string.

Source code in src/sdss_explorer/dashboard/dataclass/state.py 
214
215
216
217
218
219
220
221
222
223
224
def __repr__(self) -> str:
    """Show relevant properties of class as string."""
    return "\n".join(
        f"{k:15}: {v}" for k, v in {
            "uuid": self.uuid,
            "kernel_id": self.kernel_id,
            "df": hex(id(self.df.value)),  # dataframe mem address
            "subset_backend": hex(id(self.subset_store)),
            "release": self.release,
            "datatype": self.datatype,
        }.items())

get_default_dataset() #

Method version to get the default dataset of app (star or visit). Used for defaulting the Subset dataclass

Source code in src/sdss_explorer/dashboard/dataclass/state.py 
194
195
196
197
def get_default_dataset(self) -> str:
    """Method version to get the default dataset of app (star or visit). Used for defaulting the Subset dataclass"""
    datatype = self._datatype.value
    return "mwmlite" if datatype == "star" else "thepayne"

load_dataset(release=None, datatype=None) #

load the HDF5 dataset for the dashboard

Source code in src/sdss_explorer/dashboard/dataclass/state.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def load_dataset(self,
                 release: Optional[str] = None,
                 datatype: Optional[str] = None) -> bool:
    """ load the HDF5 dataset for the dashboard """
    # use attributes if not manually overridden
    if not release:
        release = self.release
    if not datatype:
        datatype = self.datatype

    # start with standard open operation
    # TODO: redux version via envvar?
    df = open_file(
        f"{release}/explorerAll{datatype.capitalize()}-{VASTRA}.hdf5")
    columns = load_column_json(release, datatype)

    if (df is None) and (columns is None):
        logger.critical(
            "Part of dataset load failed! ensure everything is setup (files, envvars)"
        )
        return False

    # set reactives
    self.df.set(df)
    self.columns.set(columns)

    return True

SubsetState#

SubsetState specifically holds Subset dataclasses, which are used to generate the components in the sidebar. These also hold specific vx.DataFrame instances which correspond to plots.

Subset#

Subset dataclass.

Attributes:

Name Type Description
name str

subset name
expresssion str

custom filter expression
dataset str

specific dataset
flags list[str]

toggled quick flags list
carton list[str]

selected targeting cartons
mapper list[str]

selected mapper programs
crossmatch str

multi-line string of crossmatch identifiers
cmtype str

type of crossmatch
df DataFrame

vaex dataframe, corresponds to dataset
columns list[str]

specific columns, used internally to guardrail users and reset valid columns
Source code in src/sdss_explorer/dashboard/dataclass/subsets.py 
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
@dataclasses.dataclass(frozen=True)
class Subset:
    """Subset dataclass.

    Attributes:
        name: subset name
        expresssion: custom filter expression
        dataset: specific dataset
        flags: toggled quick flags list
        carton: selected targeting cartons
        mapper: selected mapper programs
        crossmatch: multi-line string of crossmatch identifiers
        cmtype: type of crossmatch
        df: vaex dataframe, corresponds to dataset
        columns: specific columns, used internally to guardrail users and reset valid columns

    """

    name: str = "A"
    expression: str = ""
    dataset: str = dataclasses.field(default_factory=State.get_default_dataset)
    flags: list[str] = dataclasses.field(
        default_factory=lambda: ["purely non-flagged"])
    carton: list[str] = dataclasses.field(default_factory=list)
    mapper: list[str] = dataclasses.field(default_factory=list)
    crossmatch: str = ""
    cmtype: str = "gaia_dr3"
    df: vx.DataFrame = State.df.value
    columns: list[str] = dataclasses.field(default_factory=list)

    def __repr__(self) -> str:
        return "\n".join(
            f"{k:15}: {v}" for k, v in {
                "name": self.name,
                "id": State.uuid,
                "df": hex(id(self.df)),
                "expression": self.expression,
                "dataset": self.dataset,
                "carton": self.carton,
                "mapper": self.mapper,
            }.items())

API#

User-facing subset reactive class, with functions for add/remove/etc.

Attributes:

Name Type Description
index Reactive[int]

index, used to ensure unique keys
subsets Reactive[dict[str, Subset]]

dictionary of keys to subsets
Source code in src/sdss_explorer/dashboard/dataclass/subsets.py 
 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
 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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class SubsetData:
    """User-facing subset reactive class, with functions for add/remove/etc.

    Attributes:
        index (solara.Reactive[int]): index, used to ensure unique keys
        subsets (solara.Reactive[dict[str,Subset]]): dictionary of keys to subsets

    """

    def __init__(self):
        self.index = sl.reactive(1)
        self.subsets = sl.reactive({"s0": Subset()})

    def add_subset(self, name: str, **kwargs) -> bool:
        """Adds subset and subsetcard, generating new unique key for subset.

        Args:
            key: subset key
            kwargs (kwargs): keyword arguments to instantiate `Subset` with

        Returns:
            bool: `True` if successful, `False` if not.
        """
        for subset in self.subsets.value.values():
            if name == subset.name:
                Alert.update("Subset with name already exists!", color="error")
                return False
        if len(name) == 0:
            Alert.update("Please enter a name for the subset.",
                         color="warning")
            return False

        else:  # no name case
            # add subset
            key = self.index.value
            subsets = self.subsets.value.copy()
            subsets.update({"s" + str(key): Subset(name=name, **kwargs)})
            self.subsets.set(dict(**subsets))

            # iterate index
            self.index.set(self.index.value + 1)
            return True

    def update_subset(self, key: str, **kwargs) -> bool:
        """Updates subset of key with specified kwargs

        Args:
            key: subset key
            kwargs (kwargs): keyword arguments to update with

        Returns:
            bool: `True` if successful, `False` if not.
        """
        if key not in self.subsets.value.keys():
            Alert.update(
                f"BUG: subset update failed! Key {key} not in hashmap.",
                color="error")
            return False

        subsets = self.subsets.value.copy()
        subset = self.subsets.value[key]
        subsets[key] = dataclasses.replace(subset, **kwargs)
        self.subsets.set(dict(**subsets))
        return True

    def rename_subset(self, key: str, newname: str) -> bool:
        """Renames a subset.

        Args:
            key: subset key
            newname: new name for subset

        Returns:
            bool: `True` if successful, `False` if not.
        """
        if key not in self.subsets.value.keys():
            Alert.update("BUG: subset clone failed! Key not in hashmap.",
                         color="error")
            return False
        elif len(newname) == 0:
            Alert.update("Enter a name for the subset!", color="warning")
            return False
        elif newname == self.subsets.value[key].name:
            Alert.update("Name is the same!", color="warning")
            return False
        elif newname in [ss.name for ss in self.subsets.value.values()]:
            Alert.update("Name already exists!", color="warning")
            return False
        else:
            return self.update_subset(key, name=newname)

    def clone_subset(self, key: str) -> bool:
        """Clones a subset.

        Args:
            key: subset key

        Returns:
            bool: `True` if successful, `False` if not.
        """
        if key not in self.subsets.value.keys():
            Alert.update("BUG: subset clone failed! Key not in hashmap.",
                         color="error")
            return False
        else:
            subsets = self.subsets.value.copy()
            subset = self.subsets.value[key]
            # TODO: better name logic
            name = "copy of " + subset.name
            newkey = "s" + str(self.index.value)
            self.index.set(self.index.value + 1)

            subsets.update({newkey: dataclasses.replace(subset, name=name)})
            self.subsets.set(dict(**subsets))
            return True

    def remove_subset(self, key: str):
        """
        Removes a subset from the Subset master list.

        Args:
            key: subset key
        """
        # dict comprehension for reconstruction
        self.subsets.value = dict(**{
            k: v
            for k, v in self.subsets.value.items() if k != key
        })

add_subset(name, **kwargs) #

Adds subset and subsetcard, generating new unique key for subset.

Parameters:

Name Type Description Default
key

subset key

 required
kwargs kwargs

keyword arguments to instantiate Subset with

 {}

Returns:

Name Type Description
bool bool

True if successful, False if not.
Source code in src/sdss_explorer/dashboard/dataclass/subsets.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def add_subset(self, name: str, **kwargs) -> bool:
    """Adds subset and subsetcard, generating new unique key for subset.

    Args:
        key: subset key
        kwargs (kwargs): keyword arguments to instantiate `Subset` with

    Returns:
        bool: `True` if successful, `False` if not.
    """
    for subset in self.subsets.value.values():
        if name == subset.name:
            Alert.update("Subset with name already exists!", color="error")
            return False
    if len(name) == 0:
        Alert.update("Please enter a name for the subset.",
                     color="warning")
        return False

    else:  # no name case
        # add subset
        key = self.index.value
        subsets = self.subsets.value.copy()
        subsets.update({"s" + str(key): Subset(name=name, **kwargs)})
        self.subsets.set(dict(**subsets))

        # iterate index
        self.index.set(self.index.value + 1)
        return True

clone_subset(key) #

Clones a subset.

Parameters:

Name Type Description Default
key str

subset key

 required

Returns:

Name Type Description
bool bool

True if successful, False if not.
Source code in src/sdss_explorer/dashboard/dataclass/subsets.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def clone_subset(self, key: str) -> bool:
    """Clones a subset.

    Args:
        key: subset key

    Returns:
        bool: `True` if successful, `False` if not.
    """
    if key not in self.subsets.value.keys():
        Alert.update("BUG: subset clone failed! Key not in hashmap.",
                     color="error")
        return False
    else:
        subsets = self.subsets.value.copy()
        subset = self.subsets.value[key]
        # TODO: better name logic
        name = "copy of " + subset.name
        newkey = "s" + str(self.index.value)
        self.index.set(self.index.value + 1)

        subsets.update({newkey: dataclasses.replace(subset, name=name)})
        self.subsets.set(dict(**subsets))
        return True

remove_subset(key) #

Removes a subset from the Subset master list.

Parameters:

Name Type Description Default
key str

subset key

 required
Source code in src/sdss_explorer/dashboard/dataclass/subsets.py 
171
172
173
174
175
176
177
178
179
180
181
182
def remove_subset(self, key: str):
    """
    Removes a subset from the Subset master list.

    Args:
        key: subset key
    """
    # dict comprehension for reconstruction
    self.subsets.value = dict(**{
        k: v
        for k, v in self.subsets.value.items() if k != key
    })

rename_subset(key, newname) #

Renames a subset.

Parameters:

Name Type Description Default
key str

subset key

 required
newname str

new name for subset

 required

Returns:

Name Type Description
bool bool

True if successful, False if not.
Source code in src/sdss_explorer/dashboard/dataclass/subsets.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def rename_subset(self, key: str, newname: str) -> bool:
    """Renames a subset.

    Args:
        key: subset key
        newname: new name for subset

    Returns:
        bool: `True` if successful, `False` if not.
    """
    if key not in self.subsets.value.keys():
        Alert.update("BUG: subset clone failed! Key not in hashmap.",
                     color="error")
        return False
    elif len(newname) == 0:
        Alert.update("Enter a name for the subset!", color="warning")
        return False
    elif newname == self.subsets.value[key].name:
        Alert.update("Name is the same!", color="warning")
        return False
    elif newname in [ss.name for ss in self.subsets.value.values()]:
        Alert.update("Name already exists!", color="warning")
        return False
    else:
        return self.update_subset(key, name=newname)

update_subset(key, **kwargs) #

Updates subset of key with specified kwargs

Parameters:

Name Type Description Default
key str

subset key

 required
kwargs kwargs

keyword arguments to update with

 {}

Returns:

Name Type Description
bool bool

True if successful, False if not.
Source code in src/sdss_explorer/dashboard/dataclass/subsets.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def update_subset(self, key: str, **kwargs) -> bool:
    """Updates subset of key with specified kwargs

    Args:
        key: subset key
        kwargs (kwargs): keyword arguments to update with

    Returns:
        bool: `True` if successful, `False` if not.
    """
    if key not in self.subsets.value.keys():
        Alert.update(
            f"BUG: subset update failed! Key {key} not in hashmap.",
            color="error")
        return False

    subsets = self.subsets.value.copy()
    subset = self.subsets.value[key]
    subsets[key] = dataclasses.replace(subset, **kwargs)
    self.subsets.set(dict(**subsets))
    return True

VCData#

Used to hold the current virtual columns. Handles add and remove operations in the backend.

API#

Virtual column data class

Attributes:

Name Type Description
columns dict[str, str]

list of virtual columns by their name and expression.
Source code in src/sdss_explorer/dashboard/dataclass/vcdata.py 
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
class VCList:
    """Virtual column data class

    Attributes:
        columns (dict[str,str]): list of virtual columns by their name and expression.
    """

    def __init__(self) -> None:
        self.columns = sl.reactive({})

    def add_column(self, name, expression):
        """Add a virtual column to the DataFrame.

        Args:
            name (str): name of vc to add
            expression (str): custom expression
        """
        columns = self.columns.value.copy()
        columns.update({name: expression})
        self.columns.set(columns)

        State.df.value.add_virtual_column(name, expression)
        for subset in SubsetState.subsets.value.values():
            if name not in subset.df.virtual_columns.keys():
                logger.debug("adding vc " + str(name) + " to subset " +
                             str(subset.name))
                subset.df.add_virtual_column(name, expression)

    def delete_column(self, name):
        """Remove a virtual column from the DataFrame.

        Warning:
            This could lead to race conditions. Untested under high stress.

        Args:
            name (str): name of vc to remove
        """
        # trigger plot resets FIRST
        columns = self.columns.value.copy()
        columns.pop(name)

        # first, if any other VC's depend on this one, they go first
        removed = [name]
        for other_name in set(
                columns.keys()):  # prevents changing size during iteration
            if name in State.df.value[other_name].variables():
                columns.pop(other_name)  # drop from UI datastruct
                removed.append(other_name)  # save this removal for later

        # remove all in a single update to prevent race conditions
        self.columns.set(columns)

        # now delete all removed virtual columns from dataframe(s)
        for column in removed:
            State.df.value.delete_virtual_column(name)
            for subset in SubsetState.subsets.value.values():
                if column in subset.df.virtual_columns.keys():
                    subset.df.delete_virtual_column(column)

    def __repr__(self) -> str:
        return str(dict(self.columns.value))

add_column(name, expression) #

Add a virtual column to the DataFrame.

Parameters:

Name Type Description Default
name str

name of vc to add

 required
expression str

custom expression

 required
Source code in src/sdss_explorer/dashboard/dataclass/vcdata.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def add_column(self, name, expression):
    """Add a virtual column to the DataFrame.

    Args:
        name (str): name of vc to add
        expression (str): custom expression
    """
    columns = self.columns.value.copy()
    columns.update({name: expression})
    self.columns.set(columns)

    State.df.value.add_virtual_column(name, expression)
    for subset in SubsetState.subsets.value.values():
        if name not in subset.df.virtual_columns.keys():
            logger.debug("adding vc " + str(name) + " to subset " +
                         str(subset.name))
            subset.df.add_virtual_column(name, expression)

delete_column(name) #

Remove a virtual column from the DataFrame.

Warning

This could lead to race conditions. Untested under high stress.

Parameters:

Name Type Description Default
name str

name of vc to remove

 required
Source code in src/sdss_explorer/dashboard/dataclass/vcdata.py 
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
def delete_column(self, name):
    """Remove a virtual column from the DataFrame.

    Warning:
        This could lead to race conditions. Untested under high stress.

    Args:
        name (str): name of vc to remove
    """
    # trigger plot resets FIRST
    columns = self.columns.value.copy()
    columns.pop(name)

    # first, if any other VC's depend on this one, they go first
    removed = [name]
    for other_name in set(
            columns.keys()):  # prevents changing size during iteration
        if name in State.df.value[other_name].variables():
            columns.pop(other_name)  # drop from UI datastruct
            removed.append(other_name)  # save this removal for later

    # remove all in a single update to prevent race conditions
    self.columns.set(columns)

    # now delete all removed virtual columns from dataframe(s)
    for column in removed:
        State.df.value.delete_virtual_column(name)
        for subset in SubsetState.subsets.value.values():
            if column in subset.df.virtual_columns.keys():
                subset.df.delete_virtual_column(column)

GridState#

Used to control the grid layout

API#

Class holding current state of grid layout.

Note

All modifying functions are nested as subfunctions of ObjectGrid.

Attributes:

Name Type Description
grid_layout Reactive[list[dict[str, int]]]

list of grid layout properties per item
objects Reactive[list[Any]]

list of widgets to render
states Reactive[list[PlotState]]

list of states, for exporting
index Reactive[int]

index, used for ensuring unique state between widgets
Source code in src/sdss_explorer/dashboard/dataclass/gridstate.py 
 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
class GridData:
    """
    Class holding current state of grid layout.

    Note:
        All modifying functions are nested as subfunctions of `ObjectGrid`.

    Attributes:
        grid_layout (sl.Reactive[list[dict[str,int]]]): list of grid layout properties per item
        objects (sl.Reactive[list[Any]]): list of widgets to render
        states (sl.Reactive[list[PlotState]]): list of states, for exporting
        index (sl.Reactive[int]): index, used for ensuring unique state between widgets
    """

    def __init__(self, objects=[], layout=[], states=[]) -> None:
        self.grid_layout = sl.reactive(layout)
        self.objects = sl.reactive(objects)
        self.states = sl.reactive(states)
        self.index = sl.reactive(len(objects))

    def __repr__(self) -> str:
        return str({
            "uuid": State.uuid,
            "objects": self.objects.value,
            "layout": self.grid_layout.value,
        })

PlotState#

PlotState is a specific class of plot settings that is instantiated and used in individual plot inistances. See more in Plots.