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

contains_filetype(folder, extension='.nxs')

Return True if folder contains filetype

Source code in mmg_toolbox/utils/env_functions.py 
169
170
171
172
def contains_filetype(folder: str, extension: str = '.nxs') -> bool:
    """Return True if folder contains filetype"""
    gen = (file.path for file in os.scandir(folder) if file.is_file() and file.name.endswith(extension))
    return True if next(gen, False) else False

find_scan_files(scan_file, directory=None, 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

 None
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 
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
def find_scan_files(scan_file: str | int, directory: str | None = 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, max_visits=None, omit_empty=False)

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

visits = get_dls_visits('i16')

Parameters:

Name Type Description Default
instrument str | None

displays visits for this instrument, or None for current beamline.

 None
year str | int | None

displays visits in this year, or None for current year.

 None
max_visits int | None

maximum number of visits to display, nor None for all visits

 None
omit_empty bool

if True, omits visits containing no scans

 False
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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def get_dls_visits(instrument: str | None = None, year: str | int | None = None,
                   max_visits: int | None = None, omit_empty: bool = False) -> dict[str, str]:
    """
    Return dict of {visit: path} for each visible visit in the beamline directory

        visits = get_dls_visits('i16')

    :param instrument: displays visits for this instrument, or None for current beamline.
    :param year: displays visits in this year, or None for current year.
    :param max_visits: maximum number of visits to display, nor None for all visits
    :param omit_empty: if True, omits visits containing no scans
    """
    if instrument is None:
        instrument = get_beamline()
    if year is None:
        year = datetime.now().year

    if max_visits:
        visits = get_dls_visits(instrument, year, max_visits=None, omit_empty=omit_empty)
        return {visit: visits[visit] for visit in list(visits)[:max_visits]}

    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) and
                       (contains_filetype(file.path, '.nxs') if omit_empty else True)
                 ),
                key=lambda x: os.path.getmtime(x), reverse=True
            )
        }
    return {}

get_dls_visits_str(instrument=None, year=None, max_visits=3, omit_empty=True)

Return str of visits for each visible visit int he beamline directory

print(get_dls_visits_str('i16', 2025))

Parameters:

Name Type Description Default
instrument str | None

displays visits for this instrument, or None for current beamline.

 None
year str | int | None

displays visits in this year, or None for current year.

 None
max_visits int | None

maximum number of visits to display, nor None for all visits

 3
omit_empty bool

if True, omits visits containing no scans

 True

Returns:

Type Description
str

multiline string
Source code in mmg_toolbox/utils/env_functions.py 
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
def get_dls_visits_str(instrument: str | None = None, year: str | int | None = None,
                       max_visits: int | None = 3, omit_empty: bool = True) -> str:
    """
    Return str of visits for each visible visit int he beamline directory

        print(get_dls_visits_str('i16', 2025))

    :param instrument: displays visits for this instrument, or None for current beamline.
    :param year: displays visits in this year, or None for current year.
    :param max_visits: maximum number of visits to display, nor None for all visits
    :param omit_empty: if True, omits visits containing no scans
    :return: multiline string
    """
    instrument = instrument or get_beamline()
    year = year or datetime.now().year
    visits = get_dls_visits(instrument, year)
    if not visits:
        return f"No visits for instrument '{instrument}'"

    strings = []
    for visit, path in visits.items():
        scans = scan_number_mapping(path)
        if len(scans) > 0:
            last_scan_number = list(scans)[-1]
            last_scan_path = list(scans.values())[-1]
            last_scan_time = datetime.fromtimestamp(os.path.getmtime(last_scan_path), tz=timezone.utc)
            last_scan = f", #{last_scan_number:<8} ({last_scan_time:%Y-%m-%d %H:%M})"
        else:
            last_scan = ""
            if omit_empty:
                continue
        strings.append(f"{visit:10} Scans: {len(scans):4}{last_scan}")
        if len(strings) == max_visits:
            break
    return f"{instrument} visits in {year}\n" + '\n'.join(strings)

get_first_file(folder, extension='.nxs')

Return first scan in folder

Source code in mmg_toolbox/utils/env_functions.py 
175
176
177
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 
199
200
201
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 
241
242
243
244
245
246
247
248
249
250
251
252
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 
180
181
182
183
184
185
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 
235
236
237
238
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 
300
301
302
303
304
305
306
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 
267
268
269
270
271
272
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 
255
256
257
258
259
260
261
262
263
264
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 
292
293
294
295
296
297
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 
275
276
277
278
279
280
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 
283
284
285
286
287
288
289
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 
188
189
190
191
192
193
194
195
196
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()))