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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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

get_beamline(default='')

Return current beamline from environment variable

Source code in mmg_toolbox/utils/env_functions.py 
55
56
57
def get_beamline(default=''):
    """Return current beamline from environment variable"""
    return os.environ.get(BEAMLINE, default)

get_data_directory()

Return the default data directory

Source code in mmg_toolbox/utils/env_functions.py 
65
66
67
68
69
70
71
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 list of visits

Source code in mmg_toolbox/utils/env_functions.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_dls_visits(instrument: str | None = None, year: str | int | None = None) -> dict[str, ...]:
    """Return list of visits"""
    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(os.path.join(DLS, instrument, 'data', YEAR))
                 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 
104
105
106
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 
128
129
130
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 
79
80
81
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 
74
75
76
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 
139
140
141
142
143
144
145
146
147
148
149
150
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)

Return ordered list of scans numbers from nexus files in directory

Source code in mmg_toolbox/utils/env_functions.py 
109
110
111
112
113
114
def get_scan_numbers(folder: str) -> list[int]:
    """Return ordered list of scans numbers from nexus files in directory"""
    return sorted(
        number for filename in list_files(folder, extension='.nxs')
        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 
60
61
62
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 
133
134
135
136
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()

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

Source code in mmg_toolbox/utils/env_functions.py 
198
199
200
201
202
203
def open_jupyter_lab():
    """
    Open a new terminal and start a Jupyter lab terminal (linux only)
    """
    shell_cmd = f"gnome-terminal -- bash -c \"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 
165
166
167
168
169
170
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 
153
154
155
156
157
158
159
160
161
162
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 
190
191
192
193
194
195
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 
173
174
175
176
177
178
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 
181
182
183
184
185
186
187
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)

Build mapping of scan number to scan file

Source code in mmg_toolbox/utils/env_functions.py 
117
118
119
120
121
122
123
124
125
def scan_number_mapping(*folders: str) -> 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='.nxs')
        if (number := get_scan_number(filename)) > 0
    }
    return dict(sorted(mapping.items()))