Skip to content

env_functions

Environment functions

check_file_access(filepath, append='_new')

Check path has write access, if not, return appended path with write access

Source code in mmg_toolbox/utils/env_functions.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def check_file_access(filepath: str, append: str = '_new') -> str:
    """Check path has write access, if not, return appended path with write access"""
    # return filepath if it already exists and is writable
    if os.path.exists(filepath) and os.access(filepath, os.W_OK):
        return filepath
    tries = 0
    max_tries = 3

    path, name = os.path.split(filepath)
    if not os.access(path, os.W_OK):
        raise OSError(f"new file cannot be written as path is not writable: '{path}'")

    if os.path.exists(filepath):
        # filepath is not writeable, amend name
        name, ext = os.path.splitext(name)
        while os.path.exists(filepath) and not os.access(filepath, os.W_OK):
            if tries > max_tries:
                raise Exception(f"File is not writable: {filepath}")
            name += append
            filepath = os.path.join(path, name + ext)
            tries += 1
    return filepath

find_scan_files(scan_file, directory, instrument=None, extension='.nxs')

Recursively search a directory for scan files, returns a generator

fn = find_scan_files(12345, '/dls/i16/data')
filepath = next(fn, None)  # returns first filepath with '12345' in the title
all_paths = list(fn)

Parameters:

Name Type Description Default
scan_file str | int

scan number or part of filename

 required
directory str | None

directory to search for scan files, or None to use instrument

 required
instrument str | None

if directory is None, use instrument to use DLS data directory (None to use current beamline)

 None
extension str

extension to search for scan files

 '.nxs'

Returns:

Type Description
Iterator[str]

generator object for search
Source code in mmg_toolbox/utils/env_functions.py 
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
def find_scan_files(scan_file: str | int, directory: str | None, instrument: str | None = None, extension: str = '.nxs') -> Iterator[str]:
    """
    Recursively search a directory for scan files, returns a generator

        fn = find_scan_files(12345, '/dls/i16/data')
        filepath = next(fn, None)  # returns first filepath with '12345' in the title
        all_paths = list(fn)

    :param scan_file: scan number or part of filename
    :param directory: directory to search for scan files, or None to use instrument
    :param instrument: if directory is None, use instrument to use DLS data directory (None to use current beamline)
    :param extension: extension to search for scan files
    :return: generator object for search
    """
    scan_file = str(scan_file)
    if directory is None:
        if instrument is None:
            instrument = get_beamline()
        directory = os.path.join(DLS, instrument.lower(), 'data')

    def scantree(path: str):
        for entry in os.scandir(path):
            if os.access(entry.path, os.R_OK) and not entry.name.startswith('.'):
                if entry.is_file() and entry.name.endswith(extension) and scan_file in entry.name:
                    yield entry.path
                elif entry.is_dir():
                    yield from scantree(entry.path)
    return scantree(directory)

get_beamline(default='', filename=None)

Return current beamline from filename/filepath or environment variable

Source code in mmg_toolbox/utils/env_functions.py 
56
57
58
59
60
61
def get_beamline(default='', filename: str | None = None) -> str:
    """Return current beamline from filename/filepath or environment variable"""
    env_bl = os.environ.get(BEAMLINE, default)
    if filename is None:
        return env_bl
    return get_beamline_from_directory(filename, env_bl)

get_beamline_from_directory(directory, default='')

Return current beamline from given directory. Finds the last instance of [ijk]##-#.

Source code in mmg_toolbox/utils/env_functions.py 
64
65
66
67
def get_beamline_from_directory(directory: str, default: str = ''):
    """Return current beamline from given directory. Finds the last instance of [ijk]##-#."""
    beamlines = re.findall('/([a-zA-Z][0-9]{2}-?[1-9]?)', directory)
    return next(iter(beamlines[::-1]), default)

get_data_directory()

Return the default data directory

Source code in mmg_toolbox/utils/env_functions.py 
75
76
77
78
79
80
81
def get_data_directory():
    """Return the default data directory"""
    beamline = get_beamline()
    year = datetime.now().year
    if beamline:
        return f"/dls/{beamline}/data/{year}"
    return os.path.expanduser('~')

get_dls_visits(instrument=None, year=None)

Return dict of {visit: path} for each visible visit in the beamline directory

Source code in mmg_toolbox/utils/env_functions.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def get_dls_visits(instrument: str | None = None, year: str | int | None = None) -> dict[str, str]:
    """Return dict of {visit: path} for each visible visit in the beamline directory"""
    if instrument is None:
        instrument = get_beamline()
    if year is None:
        year = datetime.now().year

    dls_dir = os.path.join(DLS, instrument.lower(), 'data', str(year))
    if os.path.isdir(dls_dir):
        return {
            os.path.basename(path): path
            for path in sorted(
                (file.path for file in os.scandir(dls_dir)
                 if file.is_dir() and os.access(file.path, os.R_OK)),
                key=lambda x: os.path.getmtime(x), reverse=True
            )
        }
    return {}

get_first_file(folder, extension='.nxs')

Return first scan in folder

Source code in mmg_toolbox/utils/env_functions.py 
114
115
116
def get_first_file(folder: str, extension='.nxs') -> str:
    """Return first scan in folder"""
    return next(iter(list_files(folder, extension=extension)))

get_last_scan_number(folder)

Return latest scan number

Source code in mmg_toolbox/utils/env_functions.py 
138
139
140
def get_last_scan_number(folder: str) -> int:
    """Return latest scan number"""
    return get_scan_numbers(folder)[-1]

get_notebook_directory(data_directory)

Return the notebook directory of the visit

Source code in mmg_toolbox/utils/env_functions.py 
89
90
91
def get_notebook_directory(data_directory: str):
    """Return the notebook directory of the visit"""
    return os.path.join(data_directory, 'processed', 'notebooks')

get_processing_directory(data_directory)

Return the processing directory of the visit

Source code in mmg_toolbox/utils/env_functions.py 
84
85
86
def get_processing_directory(data_directory: str):
    """Return the processing directory of the visit"""
    return os.path.join(data_directory, 'processing')

get_scan_notebooks(scan, data_directory=None)

Return list of processed jupyter notebooks for scan

Source code in mmg_toolbox/utils/env_functions.py 
179
180
181
182
183
184
185
186
187
188
189
190
def get_scan_notebooks(scan: int | str, data_directory: str | None = None) -> list[str]:
    """Return list of processed jupyter notebooks for scan"""
    try:
        data_directory, filename = os.path.split(scan)
        scan = get_scan_number(filename)
    except TypeError:
        pass
    notebook_directory = get_notebook_directory(data_directory)
    if os.path.isdir(notebook_directory):
        notebooks = list_files(notebook_directory, '.ipynb')
        return [notebook for notebook in notebooks if str(scan) in os.path.basename(notebook)]
    return []

get_scan_numbers(folder, extension='.nxs')

Return ordered list of scans numbers from nexus files in directory

Source code in mmg_toolbox/utils/env_functions.py 
119
120
121
122
123
124
def get_scan_numbers(folder: str, extension: str = '.nxs') -> list[int]:
    """Return ordered list of scans numbers from nexus files in directory"""
    return sorted(
        number for filename in list_files(folder, extension=extension)
        if (number := get_scan_number(filename)) > 0
    )

get_user(default='')

Return current user from environment variable

Source code in mmg_toolbox/utils/env_functions.py 
70
71
72
def get_user(default=''):
    """Return current user from environment variable"""
    return next((os.environ[u] for u in USER if u in os.environ), default)

last_folder_update(folder)

Returns datetime timestamp of last folder update

Source code in mmg_toolbox/utils/env_functions.py 
173
174
175
176
def last_folder_update(folder: str) -> datetime:
    """Returns datetime timestamp of last folder update"""
    modified = os.path.getmtime(folder)
    return datetime.fromtimestamp(modified)

open_jupyter_lab(directory=None)

Open a new terminal and start a Jupyter lab terminal (linux only)

Source code in mmg_toolbox/utils/env_functions.py 
238
239
240
241
242
243
244
def open_jupyter_lab(directory: str | None = None):
    """
    Open a new terminal and start a Jupyter lab terminal (linux only)
    """
    dir_cmd = f"cd {directory};" or ""
    shell_cmd = f"gnome-terminal -- bash -c \"{dir_cmd}jupyter lab; exec bash\""
    subprocess.Popen(shell_cmd, shell=True)

open_terminal(command)

Open a new terminal window (linux only) and run a command

Source code in mmg_toolbox/utils/env_functions.py 
205
206
207
208
209
210
def open_terminal(command: str):
    """
    Open a new terminal window (linux only) and run a command
    """
    shell_cmd = f"gnome-terminal -- bash -c \"{command}; exec bash\""
    subprocess.Popen(shell_cmd, shell=True)

run_command(command)

Run shell command, print output to terminal

Source code in mmg_toolbox/utils/env_functions.py 
193
194
195
196
197
198
199
200
201
202
def run_command(command: str):
    """
    Run shell command, print output to terminal
    """
    print('\n\n\n################# Starting ###################')
    print(f"Running command:\n{command}\n\n\n")
    output = subprocess.run(command, shell=True, capture_output=True)
    print(output.stdout.decode())
    print(output.stderr.decode())
    print('\n\n\n################# Finished ###################\n\n\n')

run_jupyter_notebook(notebook_filename)

Run a jupyter notebook

Source code in mmg_toolbox/utils/env_functions.py 
230
231
232
233
234
235
def run_jupyter_notebook(notebook_filename: str):
    """
    Run a jupyter notebook
    """
    command = f"jupyter notebook {notebook_filename}"
    run_command(command)

run_python_script(script_filename)

Run shell command, print output to terminal

Source code in mmg_toolbox/utils/env_functions.py 
213
214
215
216
217
218
def run_python_script(script_filename: str):
    """
    Run shell command, print output to terminal
    """
    command = f"{sys.executable} {script_filename}"
    run_command(command)

run_python_string(script)

Run shell command, print output to terminal

Source code in mmg_toolbox/utils/env_functions.py 
221
222
223
224
225
226
227
def run_python_string(script: str):
    """
    Run shell command, print output to terminal
    """
    # command = f"{sys.executable} -c {script}"
    # run_command(command)
    subprocess.run([sys.executable, '-c', script])

scan_number_mapping(*folders, extension='.nxs')

Build mapping of scan number to scan file

Source code in mmg_toolbox/utils/env_functions.py 
127
128
129
130
131
132
133
134
135
def scan_number_mapping(*folders: str, extension: str = '.nxs') -> dict[int, str]:
    """Build mapping of scan number to scan file"""
    mapping = {
        number: filename
        for folder in folders
        for filename in list_files(folder, extension=extension)
        if (number := get_scan_number(filename)) > 0
    }
    return dict(sorted(mapping.items()))