Skip to content

Cctbx

cctbx

Source code in mfx/cctbx.py 
  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
 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
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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
class cctbx:
    def __init__(self):
        from mfx.macros import get_exp
        self.experiment = str(get_exp())


    def geom_refine(
        self,
        user: str,
        group: str,
        level: int = None,
        facility: str = "NERSC",
        exp: str = ''):
        """Launch CCTBX XFEL GUI.

        Parameters:

            user (str): username for computer account at facility.

            group (str): the trial and rungroup number in the format 000_rg005.
            Default is newest trial_rungroup

            level (int): the level of geometry refinement.
            0 = whole detector and 1 = individual detector panels.
            Default is to systematically do both.

            facility (str): Default: "NERSC". Options: "S3DF, NERSC".

            exp (str): experiment number in format 'mfxp1047723'.
                       If none selected default is the current experiment.

            debug (bool): Default: False.
        """
        import logging
        import os

        if exp != '':
            experiment = exp
        else:
            experiment = self.experiment

        proc = [
            f"ssh -Yt {user}@s3dflogin "
            f"python /sdf/group/lcls/ds/tools/mfx/scripts/cctbx/geom_refine.py "
            f"-e {experiment} -f {facility} -g {group} -l {level} "
            ]

        logging.info(proc)

        if facility == 'NERSC':
            logging.warning(f"Have you renewed your token with sshproxy today?")
            token = input("(y/n)? ")

            if token.lower() == "n":
                self.sshproxy(user)

        os.system(proc[0])


    def average(
        self,
        user: str,
        run: int,
        facility: str = "NERSC",
        exp: str = '',
        debug: bool = False):
        """Launch CCTBX XFEL GUI.

        Parameters:

            user (str): username for computer account at facility.

            run (int): Enter -r for the run number

            facility (str): Default: "NERSC". Options: "S3DF, NERSC".

            exp (str): experiment number in format 'mfxp1047723'.
                       If none selected default is the current experiment.

            debug (bool): Default: False.
        """
        import logging
        import os
        import subprocess

        if exp != '':
            experiment = exp
        else:
            experiment = self.experiment

        proc = [
            f"ssh -Yt {user}@s3dflogin "
            f"python /sdf/group/lcls/ds/tools/mfx/scripts/cctbx/average.py "
            f"-e {experiment} -f {facility} -d {str(debug)} -r {run}"
            ]

        logging.info(proc)

        if facility == 'NERSC':
            logging.warning(f"Have you renewed your token with sshproxy today?")
            token = input("(y/n)? ")

            if token.lower() == "n":
                self.sshproxy(user)

        if debug:
            os.system(proc[0])
        else:
            subprocess.Popen(
                proc, shell=True,
                stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)


    def image_viewer(
        self,
        user: str,
        run: int,
        image_type: str,
        group: str = None,
        facility: str = "NERSC",
        exp: str = '',
        debug: bool = False):
        """Launch CCTBX XFEL GUI.

        Parameters:

            user (str): username for computer account at facility.

            run (int): Enter -r for the run number

            image_type (str): Enter -t for type of image view

            group (str): the trial and rungroup number in the format 000_rg005.
            Default is newest trial_rungroup

            facility (str): Default: "NERSC". Options: "S3DF, NERSC".

            exp (str): experiment number in format 'mfxp1047723'.
                       If none selected default is the current experiment.

            debug (bool): Default: False.
        """
        import logging
        import os
        import subprocess

        if exp != '':
            experiment = exp
        else:
            experiment = self.experiment

        proc = [
            f"ssh -Yt {user}@s3dflogin "
            f"python /sdf/group/lcls/ds/tools/mfx/scripts/cctbx/image_viewer.py "
            f"-e {experiment} -f {facility} -d {str(debug)} -t {image_type} -r {run} -g {group}"
            ]

        logging.info(proc)

        if facility == 'NERSC':
            logging.warning(f"Have you renewed your token with sshproxy today?")
            token = input("(y/n)? ")

            if token.lower() == "n":
                self.sshproxy(user)

        if debug:
            os.system(proc[0])
        else:
            subprocess.Popen(
                proc, shell=True,
                stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)


    def sshproxy(
        self,
        user: str,
    ):

        """Launch sshproxy check for getting NERSC token if needed

        Parameters:

            user (str): username for computer account at facility.

            debug (bool): Default: False.
        """
        import logging
        import os
        import subprocess
        logging.info(f"Creating new sshproxy token for {user}.")
        proc = [
            f"ssh -Yt {user}@s3dflogin "
            f"/sdf/group/lcls/ds/tools/mfx/scripts/cctbx/sshproxy.sh "
            f"-c cctbx -u {user}"
            ]

        logging.info(proc)

        os.system(proc[0])


    def xfel_gui(
        self,
        user: str,
        facility: str = "NERSC",
        exp: str  = '',
        debug: bool = False,
    ):
        """Launch CCTBX XFEL GUI.

        Parameters:

            user (str): username for computer account at facility.

            facility (str): Default: "NERSC". Options: "S3DF, NERSC".

            exp (str): experiment number in format 'mfxp1047723'.
                       If none selected default is the current experiment.

            debug (bool): Default: False.
        """
        import logging
        import os
        import subprocess

        if exp != '':
            experiment = exp
        else:
            experiment = self.experiment

        proc = [
            f"ssh -Yt {user}@s3dflogin "
            f"/sdf/group/lcls/ds/tools/mfx/scripts/cctbx/cctbx.sh "
            f"{user} {experiment} {facility} 1 {str(debug)} "
            ]

        logging.info(proc)

        if facility == 'NERSC':
            logging.warning(f"Have you renewed your token with sshproxy today?")
            token = input("(y/n)? ")

            if token.lower() == "n":
                self.sshproxy(user)

        if debug:
            os.system(proc[0])
        else:
            subprocess.Popen(
                proc, shell=True,
                stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)


    def notch_check(self, user, runs=[]):
        import logging
        import subprocess
        import sys
        if len(runs) > 0:
            run_list = []
            for run in runs:
                run_list.append(f'{experiment}:{run}')
            logging.info(f'Selected runs: {run_list}')
            runlist = ' '
            runlist = runlist.join(run_list)
            logging.info(f'Selected runs: {runlist}')
        else:
            logging.warning(f'No selected runs. Program will exit.')
            sys.exit()

        proc = [
            f'ssh -YAC {user}@s3dflogin '
            f'/sdf/group/lcls/ds/tools/mfx/scripts/cctbx/cctbx_notch_check.sh "{self.runlist}"'
            ]

        logging.info(proc)

        subprocess.Popen(
            proc, shell=True, 
            stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)

average(user, run, facility='NERSC', exp='', debug=False)

Launch CCTBX XFEL GUI.

Parameters:

user (str): username for computer account at facility.

run (int): Enter -r for the run number

facility (str): Default: "NERSC". Options: "S3DF, NERSC".

exp (str): experiment number in format 'mfxp1047723'.
           If none selected default is the current experiment.

debug (bool): Default: False.
Source code in mfx/cctbx.py 
 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
def average(
    self,
    user: str,
    run: int,
    facility: str = "NERSC",
    exp: str = '',
    debug: bool = False):
    """Launch CCTBX XFEL GUI.

    Parameters:

        user (str): username for computer account at facility.

        run (int): Enter -r for the run number

        facility (str): Default: "NERSC". Options: "S3DF, NERSC".

        exp (str): experiment number in format 'mfxp1047723'.
                   If none selected default is the current experiment.

        debug (bool): Default: False.
    """
    import logging
    import os
    import subprocess

    if exp != '':
        experiment = exp
    else:
        experiment = self.experiment

    proc = [
        f"ssh -Yt {user}@s3dflogin "
        f"python /sdf/group/lcls/ds/tools/mfx/scripts/cctbx/average.py "
        f"-e {experiment} -f {facility} -d {str(debug)} -r {run}"
        ]

    logging.info(proc)

    if facility == 'NERSC':
        logging.warning(f"Have you renewed your token with sshproxy today?")
        token = input("(y/n)? ")

        if token.lower() == "n":
            self.sshproxy(user)

    if debug:
        os.system(proc[0])
    else:
        subprocess.Popen(
            proc, shell=True,
            stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)

geom_refine(user, group, level=None, facility='NERSC', exp='')

Launch CCTBX XFEL GUI.

Parameters:

user (str): username for computer account at facility.

group (str): the trial and rungroup number in the format 000_rg005.
Default is newest trial_rungroup

level (int): the level of geometry refinement.
0 = whole detector and 1 = individual detector panels.
Default is to systematically do both.

facility (str): Default: "NERSC". Options: "S3DF, NERSC".

exp (str): experiment number in format 'mfxp1047723'.
           If none selected default is the current experiment.

debug (bool): Default: False.
Source code in mfx/cctbx.py 
 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
def geom_refine(
    self,
    user: str,
    group: str,
    level: int = None,
    facility: str = "NERSC",
    exp: str = ''):
    """Launch CCTBX XFEL GUI.

    Parameters:

        user (str): username for computer account at facility.

        group (str): the trial and rungroup number in the format 000_rg005.
        Default is newest trial_rungroup

        level (int): the level of geometry refinement.
        0 = whole detector and 1 = individual detector panels.
        Default is to systematically do both.

        facility (str): Default: "NERSC". Options: "S3DF, NERSC".

        exp (str): experiment number in format 'mfxp1047723'.
                   If none selected default is the current experiment.

        debug (bool): Default: False.
    """
    import logging
    import os

    if exp != '':
        experiment = exp
    else:
        experiment = self.experiment

    proc = [
        f"ssh -Yt {user}@s3dflogin "
        f"python /sdf/group/lcls/ds/tools/mfx/scripts/cctbx/geom_refine.py "
        f"-e {experiment} -f {facility} -g {group} -l {level} "
        ]

    logging.info(proc)

    if facility == 'NERSC':
        logging.warning(f"Have you renewed your token with sshproxy today?")
        token = input("(y/n)? ")

        if token.lower() == "n":
            self.sshproxy(user)

    os.system(proc[0])

image_viewer(user, run, image_type, group=None, facility='NERSC', exp='', debug=False)

Launch CCTBX XFEL GUI.

Parameters:

user (str): username for computer account at facility.

run (int): Enter -r for the run number

image_type (str): Enter -t for type of image view

group (str): the trial and rungroup number in the format 000_rg005.
Default is newest trial_rungroup

facility (str): Default: "NERSC". Options: "S3DF, NERSC".

exp (str): experiment number in format 'mfxp1047723'.
           If none selected default is the current experiment.

debug (bool): Default: False.
Source code in mfx/cctbx.py 
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
def image_viewer(
    self,
    user: str,
    run: int,
    image_type: str,
    group: str = None,
    facility: str = "NERSC",
    exp: str = '',
    debug: bool = False):
    """Launch CCTBX XFEL GUI.

    Parameters:

        user (str): username for computer account at facility.

        run (int): Enter -r for the run number

        image_type (str): Enter -t for type of image view

        group (str): the trial and rungroup number in the format 000_rg005.
        Default is newest trial_rungroup

        facility (str): Default: "NERSC". Options: "S3DF, NERSC".

        exp (str): experiment number in format 'mfxp1047723'.
                   If none selected default is the current experiment.

        debug (bool): Default: False.
    """
    import logging
    import os
    import subprocess

    if exp != '':
        experiment = exp
    else:
        experiment = self.experiment

    proc = [
        f"ssh -Yt {user}@s3dflogin "
        f"python /sdf/group/lcls/ds/tools/mfx/scripts/cctbx/image_viewer.py "
        f"-e {experiment} -f {facility} -d {str(debug)} -t {image_type} -r {run} -g {group}"
        ]

    logging.info(proc)

    if facility == 'NERSC':
        logging.warning(f"Have you renewed your token with sshproxy today?")
        token = input("(y/n)? ")

        if token.lower() == "n":
            self.sshproxy(user)

    if debug:
        os.system(proc[0])
    else:
        subprocess.Popen(
            proc, shell=True,
            stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)

sshproxy(user)

Launch sshproxy check for getting NERSC token if needed

Parameters:

user (str): username for computer account at facility.

debug (bool): Default: False.
Source code in mfx/cctbx.py 
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
def sshproxy(
    self,
    user: str,
):

    """Launch sshproxy check for getting NERSC token if needed

    Parameters:

        user (str): username for computer account at facility.

        debug (bool): Default: False.
    """
    import logging
    import os
    import subprocess
    logging.info(f"Creating new sshproxy token for {user}.")
    proc = [
        f"ssh -Yt {user}@s3dflogin "
        f"/sdf/group/lcls/ds/tools/mfx/scripts/cctbx/sshproxy.sh "
        f"-c cctbx -u {user}"
        ]

    logging.info(proc)

    os.system(proc[0])

xfel_gui(user, facility='NERSC', exp='', debug=False)

Launch CCTBX XFEL GUI.

Parameters:

user (str): username for computer account at facility.

facility (str): Default: "NERSC". Options: "S3DF, NERSC".

exp (str): experiment number in format 'mfxp1047723'.
           If none selected default is the current experiment.

debug (bool): Default: False.
Source code in mfx/cctbx.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def xfel_gui(
    self,
    user: str,
    facility: str = "NERSC",
    exp: str  = '',
    debug: bool = False,
):
    """Launch CCTBX XFEL GUI.

    Parameters:

        user (str): username for computer account at facility.

        facility (str): Default: "NERSC". Options: "S3DF, NERSC".

        exp (str): experiment number in format 'mfxp1047723'.
                   If none selected default is the current experiment.

        debug (bool): Default: False.
    """
    import logging
    import os
    import subprocess

    if exp != '':
        experiment = exp
    else:
        experiment = self.experiment

    proc = [
        f"ssh -Yt {user}@s3dflogin "
        f"/sdf/group/lcls/ds/tools/mfx/scripts/cctbx/cctbx.sh "
        f"{user} {experiment} {facility} 1 {str(debug)} "
        ]

    logging.info(proc)

    if facility == 'NERSC':
        logging.warning(f"Have you renewed your token with sshproxy today?")
        token = input("(y/n)? ")

        if token.lower() == "n":
            self.sshproxy(user)

    if debug:
        os.system(proc[0])
    else:
        subprocess.Popen(
            proc, shell=True,
            stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)