15176787354

Committed 21 May 2025 01:37PM UTC coverage: 87.493% (-0.003%) from 87.496%

Build # 15176787354

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1075 from mantidproject/1073_catch_oserror

Catch OSError exception when attempting to save to read-only folder

Run Details

3057 of 3494 relevant lines covered (87.49%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.12

/src/mslice/util/mantid/algorithm_wrapper.py

from uuid import uuid4

from mslice.models.workspacemanager.workspace_provider import (
    add_workspace,
    get_workspace_handle,
)
from mantid.api import AnalysisDataService, AlgorithmManager, Workspace

from mslice.workspace import wrap_workspace
from mslice.workspace.base import WorkspaceBase as MsliceWorkspace
from mslice.workspace.workspace import Workspace as MsliceWorkspace2D


def _parse_ws_names(args, kwargs):
    input_workspace = kwargs.get("InputWorkspace", None)
    if input_workspace:
        kwargs["InputWorkspace"] = _name_or_wrapper_to_workspace(input_workspace)
    elif len(args) > 0:
        if isinstance(args[0], MsliceWorkspace) or isinstance(args[0], str):
            input_workspace = get_workspace_handle(args[0])
        args = (_name_or_wrapper_to_workspace(args[0]),) + args[1:]

    output_name = ""
    if "OutputWorkspace" in kwargs:
        output_name = kwargs.pop("OutputWorkspace")

    for key in kwargs.keys():
        if input_workspace is None and "LHS" in key:
            input_workspace = get_workspace_handle(kwargs[key])
        if "Input" not in key and "Output" not in key:
            if isinstance(kwargs[key], MsliceWorkspace):
                kwargs[key] = _name_or_wrapper_to_workspace(kwargs[key])

    return input_workspace, output_name, args, kwargs


def _alg_has_outputws(wrapped_alg):
    alg = AlgorithmManager.create(wrapped_alg.__name__)
    return any(["OutputWorkspace" in prop.name for prop in alg.getProperties()])


def wrap_algorithm(algorithm):
    def alg_wrapper(*args, **kwargs):
        input_workspace, output_name, args, kwargs = _parse_ws_names(args, kwargs)

        args = tuple(
            [
                _name_or_wrapper_to_workspace(arg)
                if isinstance(arg, MsliceWorkspace)
                else arg
                for arg in args
            ]
        )

        if "InputWorkspaces" in kwargs:
            kwargs["InputWorkspaces"] = [
                _name_or_wrapper_to_workspace(arg) for arg in kwargs["InputWorkspaces"]
            ]

        for ky in [k for k in kwargs.keys() if "Workspace" in k]:
            if isinstance(kwargs[ky], str) and "__MSL" not in kwargs[ky]:
                kwargs[ky] = _name_or_wrapper_to_workspace(kwargs[ky])

        if _alg_has_outputws(algorithm):
            ads_name = (
                "__MSL" + output_name if output_name else "__MSLTMP" + str(uuid4())[:8]
            )
            store = kwargs.pop("store", True)
            if not store:
                ads_name += "_HIDDEN"
            result = algorithm(*args, OutputWorkspace=ads_name, **kwargs)
        else:
            result = algorithm(*args, **kwargs)

        if isinstance(result, Workspace):
            if isinstance(input_workspace, MsliceWorkspace2D) and isinstance(
                result, type(input_workspace.raw_ws)
            ):
                result = get_workspace_handle(input_workspace).rewrap(result)
                result.name = output_name
            else:
                result = wrap_workspace(result, output_name)
            if store:
                add_workspace(result, output_name)
                from mslice.app import is_gui

                if is_gui():
                    from mslice.app.presenters import get_slice_plotter_presenter

                    get_slice_plotter_presenter().update_displayed_workspaces()
        return result

    return alg_wrapper


def _name_or_wrapper_to_workspace(input_ws):
    if isinstance(input_ws, MsliceWorkspace):
        return input_ws.raw_ws
    elif isinstance(input_ws, str):
        return get_workspace_handle(input_ws).raw_ws
    else:
        return input_ws


def add_to_ads(workspaces):
    try:
        workspaces = iter(workspaces)
    except TypeError:
        workspaces = [workspaces]
    for workspace in workspaces:
        startid = (
            (5 if workspace.name.startswith("__mat") else 2)
            if workspace.name.startswith("__")
            else 0
        )
        AnalysisDataService.Instance().addOrReplace(
            workspace.name[startid:], workspace.raw_ws
        )


def remove_from_ads(workspacename):
    if AnalysisDataService.Instance().doesExist(workspacename):
        AnalysisDataService.Instance().remove(workspacename)
    # Remove hidden workspaces from ADS
    hiddenworkspace = "__MSL" + workspacename
    if AnalysisDataService.Instance().doesExist(hiddenworkspace):
        AnalysisDataService.Instance().remove(hiddenworkspace)
    hiddenworkspace = "__MSL__mat" + workspacename
    if AnalysisDataService.Instance().doesExist(hiddenworkspace):
        AnalysisDataService.Instance().remove(hiddenworkspace)
    hiddenworkspace = "__MSL" + workspacename + "_HIDDEN"
    if AnalysisDataService.Instance().doesExist(hiddenworkspace):
        AnalysisDataService.Instance().remove(hiddenworkspace)

1	from uuid import uuid4	1✔
2
3	from mslice.models.workspacemanager.workspace_provider import (	1✔
4	add_workspace,
5	get_workspace_handle,
6	)
7	from mantid.api import AnalysisDataService, AlgorithmManager, Workspace	1✔
8
9	from mslice.workspace import wrap_workspace	1✔
10	from mslice.workspace.base import WorkspaceBase as MsliceWorkspace	1✔
11	from mslice.workspace.workspace import Workspace as MsliceWorkspace2D	1✔
12
13
14	def _parse_ws_names(args, kwargs):	1✔
15	input_workspace = kwargs.get("InputWorkspace", None)	1✔
16	if input_workspace:	1✔
17	kwargs["InputWorkspace"] = _name_or_wrapper_to_workspace(input_workspace)	1✔
18	elif len(args) > 0:	1✔
19	if isinstance(args[0], MsliceWorkspace) or isinstance(args[0], str):	1✔
20	input_workspace = get_workspace_handle(args[0])	1✔
21	args = (_name_or_wrapper_to_workspace(args[0]),) + args[1:]	1✔
22
23	output_name = ""	1✔
24	if "OutputWorkspace" in kwargs:	1✔
25	output_name = kwargs.pop("OutputWorkspace")	1✔
26
27	for key in kwargs.keys():	1✔
28	if input_workspace is None and "LHS" in key:	1✔
29	input_workspace = get_workspace_handle(kwargs[key])	1✔
30	if "Input" not in key and "Output" not in key:	1✔
31	if isinstance(kwargs[key], MsliceWorkspace):	1✔
32	kwargs[key] = _name_or_wrapper_to_workspace(kwargs[key])	1✔
33
34	return input_workspace, output_name, args, kwargs	1✔
35
36
37	def _alg_has_outputws(wrapped_alg):	1✔
38	alg = AlgorithmManager.create(wrapped_alg.__name__)	1✔
39	return any(["OutputWorkspace" in prop.name for prop in alg.getProperties()])	1✔
40
41
42	def wrap_algorithm(algorithm):	1✔
43	def alg_wrapper(args, *kwargs):	1✔
44	input_workspace, output_name, args, kwargs = _parse_ws_names(args, kwargs)	1✔
45
46	args = tuple(	1✔
47	[
48	_name_or_wrapper_to_workspace(arg)
49	if isinstance(arg, MsliceWorkspace)
50	else arg
51	for arg in args
52	]
53	)
54
55	if "InputWorkspaces" in kwargs:	1✔
56	kwargs["InputWorkspaces"] = [	1✔
57	_name_or_wrapper_to_workspace(arg) for arg in kwargs["InputWorkspaces"]
58	]
59
60	for ky in [k for k in kwargs.keys() if "Workspace" in k]:	1✔
61	if isinstance(kwargs[ky], str) and "__MSL" not in kwargs[ky]:	1✔
62	kwargs[ky] = _name_or_wrapper_to_workspace(kwargs[ky])	1✔
63
64	if _alg_has_outputws(algorithm):	1✔
65	ads_name = (	1✔
66	"__MSL" + output_name if output_name else "__MSLTMP" + str(uuid4())[:8]
67	)
68	store = kwargs.pop("store", True)	1✔
69	if not store:	1✔
70	ads_name += "_HIDDEN"	1✔
71	result = algorithm(args, OutputWorkspace=ads_name, *kwargs)	1✔
72	else:
73	result = algorithm(args, *kwargs)	1✔
74
75	if isinstance(result, Workspace):	1✔
76	if isinstance(input_workspace, MsliceWorkspace2D) and isinstance(	1✔
77	result, type(input_workspace.raw_ws)
78	):
79	result = get_workspace_handle(input_workspace).rewrap(result)	1✔
80	result.name = output_name	1✔
81	else:
82	result = wrap_workspace(result, output_name)	1✔
83	if store:	1✔
84	add_workspace(result, output_name)	1✔
85	from mslice.app import is_gui	1✔
86
87	if is_gui():	1✔
88	from mslice.app.presenters import get_slice_plotter_presenter	×
89
90	get_slice_plotter_presenter().update_displayed_workspaces()	×
91	return result	1✔
92
93	return alg_wrapper	1✔
94
95
96	def _name_or_wrapper_to_workspace(input_ws):	1✔
97	if isinstance(input_ws, MsliceWorkspace):	1✔
98	return input_ws.raw_ws	1✔
99	elif isinstance(input_ws, str):	1✔
100	return get_workspace_handle(input_ws).raw_ws	1✔
101	else:
102	return input_ws	1✔
103
104
105	def add_to_ads(workspaces):	1✔
106	try:	1✔
107	workspaces = iter(workspaces)	1✔
108	except TypeError:	1✔
109	workspaces = [workspaces]	1✔
110	for workspace in workspaces:	1✔
111	startid = (	1✔
112	(5 if workspace.name.startswith("__mat") else 2)
113	if workspace.name.startswith("__")
114	else 0
115	)
116	AnalysisDataService.Instance().addOrReplace(	1✔
117	workspace.name[startid:], workspace.raw_ws
118	)
119
120
121	def remove_from_ads(workspacename):	1✔
122	if AnalysisDataService.Instance().doesExist(workspacename):	1✔
123	AnalysisDataService.Instance().remove(workspacename)	1✔
124	# Remove hidden workspaces from ADS
125	hiddenworkspace = "__MSL" + workspacename	1✔
126	if AnalysisDataService.Instance().doesExist(hiddenworkspace):	1✔
127	AnalysisDataService.Instance().remove(hiddenworkspace)	1✔
128	hiddenworkspace = "__MSL__mat" + workspacename	1✔
129	if AnalysisDataService.Instance().doesExist(hiddenworkspace):	1✔
130	AnalysisDataService.Instance().remove(hiddenworkspace)	×
131	hiddenworkspace = "__MSL" + workspacename + "_HIDDEN"	1✔
132	if AnalysisDataService.Instance().doesExist(hiddenworkspace):	1✔
133	AnalysisDataService.Instance().remove(hiddenworkspace)	×

mantidproject / mslice / 15176787354

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous