scripts.config_generator.savu_config._rem() - Code Metrics - Inspection of "Merge pull request #777 from jessicavers/master" - DiamondLightSource/Savu - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 2fd334...552d1f )

by Nicola

created 2021-06-30 12:25 UTC

scripts.config_generator.savu_config._rem() A

↳ Parent: scripts.config_generator.savu_config

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	2
dl	0
loc	7
rs	10
c	0
b	0
f	0

# Copyright 2014 Diamond Light Source Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
.. module:: savu_config
   :platform: Unix
   :synopsis: A command line tool for creating Savu plugin lists

.. moduleauthor:: Nicola Wadeson <[email protected]>

"""

import re
import sys
import logging

logger = logging.getLogger('documentationLog')
logger_rst = logging.getLogger('documentationRst')

import warnings
with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    from .content import Content
    from .completer import Completer
    from .display_formatter import ListDisplay, DispDisplay, \
        CiteDisplay
    from . import arg_parsers as parsers
    from savu.plugins import utils as pu
    from . import config_utils as utils
    from .config_utils import parse_args
    from .config_utils import error_catcher


def _help(content, args):
    """ Display the help information"""
    print('%s Savu configurator commands %s\n' % tuple(['*'*21]*2))
    for key in list(commands.keys()):
        doc = commands[key].__doc__
        if doc:
            print("%8s : %s" % (key, commands[key].__doc__))
    print("\n%s\n* For more information about individual commands type "
          "'<command> -h' *\n%s\n" % tuple(['*'*70]*2))
    return content


@parse_args
@error_catcher
def _open(content, args):
    """ Open an existing process list."""
    content.fopen(args.file, update=True, skip=args.skip)
    _ref(content, '* -n')
    _disp(content, '-q')
    return content


@parse_args
@error_catcher
def _disp(content, args):
    """ Display the plugins in the current list."""
    range_dict = content.split_plugin_string(args.start, args.stop,
                                             subelem_view=True)
    formatter = DispDisplay(content.plugin_list)
    verbosity = parsers._get_verbosity(args)
    level = 'advanced' if args.all else content.disp_level
    datasets = True if args.datasets else False
    content.display(formatter, current_level=level, verbose=verbosity,
                    datasets=datasets, disp_level=args.level, **range_dict)
    return content


@parse_args
@error_catcher
def _list(content, args):
    """ List the available plugins. """
    list_content = Content()
    utils._populate_plugin_list(list_content, pfilter=args.string)
    if not len(list_content.plugin_list.plugin_list):
        print("No result found.")
        return content

    # Sort the list of dictionaries.
    # Sort by the name of the plugin within each dictionary.
    list_content.plugin_list.plugin_list = \
        sorted(list_content.plugin_list.plugin_list,
                key=lambda i: i['name'])

    formatter = ListDisplay(list_content.plugin_list)
    verbosity = parsers._get_verbosity(args)
    list_content.display(formatter, verbose=verbosity)
    return content


@parse_args
@error_catcher
def _save(content, args):
    """ Save the current process list to file."""
    out_file = content.filename if args.input else args.filepath
    content.check_file(out_file)
    print()
    DispDisplay(content.plugin_list)._notices()
    content.save(out_file, check=input("Are you sure you want to save the "
                 "current data to %s [y/N]" % (out_file)),
                 template=args.template)
    return content


@parse_args
@error_catcher
def _mod(content, args):
    """ Modify plugin parameters."""
    pos_str, subelem, dims, command = \
        content.separate_plugin_subelem(args.param, False)
    if 'expand' in command:
        # Run the start stop step view for that dimension alone
        _expand(content, f"{pos_str} {dims} {True}")
    else:
        content.check_required_args(args.value, True)
        # Get the parameter name for the display later
        args.param = content.get_param_arg_str(pos_str, subelem)
        content_modified = content.modify(pos_str, subelem,
                                          ' '.join(args.value),
                                          dim=command)
        if content_modified:
            # Display the selected parameter only
            _disp(content, str(args.param))
    return content


@parse_args
@error_catcher
def _expand(content, args):
    """ Expand the plugin preview parameter. """
    content.set_preview_display(args.off, args.dim, args.dim_view,
                                args.plugin_pos)
    if content.expand_dim is not None:
        if content.expand_dim == "all" and args.dim is not None:
            is_modified = content.modify_dimensions(args.plugin_pos, args.dim)
            if is_modified:
                _disp(content, f"{args.plugin_pos}.preview")
        else:
            _disp(content, f"{args.plugin_pos}.preview")
            #content.set_preview_display(args.off, "all")
    return content


@parse_args
@error_catcher
def _set(content, args):
    """ Set the status of the plugin to be on or off. """
    for pos in args.plugin_pos:
        content.on_and_off(pos, args.status.upper())
    _disp(content, '-q')
    return content


@parse_args
@error_catcher
def _add(content, args):
    """ Add a plugin to the list. """
    elems = content.get_positions()
    final = str(int(re.findall(r'\d+', elems[-1])[0])+1) if elems else 1
    content.add(args.name, args.pos if args.pos else str(final))
    _disp(content, '-q')
    return content


@parse_args
@error_catcher
def _ref(content, args):
    """ Refresh a plugin (update it). """
    positions = content.get_positions() if args.pos == ['*'] else args.pos
    for pos_str in positions:
        content.refresh(pos_str, defaults=args.defaults)
        if not args.nodisp:
            _disp(content, pos_str)
    return content


@parse_args
@error_catcher
def _cite(content, args):
    """ Display plugin citations."""
    range_dict = content.split_plugin_string(args.start, args.stop)
    formatter = CiteDisplay(content.plugin_list)
    content.display(formatter, **range_dict)
    return content


@parse_args
@error_catcher
def _rem(content, args):
    """ Remove plugin(s) from the list. """
    content.remove(content.find_position(args.pos))
    _disp(content, '-q')
    return content


@parse_args
@error_catcher
def _move(content, args):
    """ Move a plugin to a different position in the list."""
    content.move(args.orig_pos, args.new_pos)
    _disp(content, '-q')
    return content


@parse_args
@error_catcher
def _coll(content, arg):
    """ List all plugin collections. """
    colls = Completer([])._get_collections()
    print('\n')
    for c in colls:
        print('%s\n %s' % ('-'*40, c))
    print('-'*40, '\n')
    return content

@parse_args
@error_catcher
def _clear(content, arg):
    """ Clear the current plugin list."""
    content.clear(check=input("Are you sure you want to clear the current "
                  "plugin list? [y/N]"))
    return content

@parse_args
@error_catcher
def _exit(content, arg):
    """ Close the program."""
    content.set_finished(check=input("Are you sure? [y/N]"))
    return content

@parse_args
@error_catcher
def _level(content, args):
    """ Set a visibility level for the parameters."""
    content.level(args.level)
    return content

@parse_args
@error_catcher
def _history(content, arg):
    """ View the history of previous commands """
    hlen = utils.readline.get_current_history_length()
    for i in range(hlen):
        print("%5i : %s" % (i, utils.readline.get_history_item(i)))
    return content


commands = {'open': _open,
            'help': _help,
            'disp': _disp,
            'list': _list,
            'save': _save,
            'mod': _mod,
            'set': _set,
            'add': _add,
            'rem': _rem,
            'move': _move,
            'ref': _ref,
            'level': _level,
            'expand': _expand,
            'cite': _cite,
            'coll': _coll,
            'clear': _clear,
            'exit': _exit,
            'history': _history}

def get_description():
    """ For each command, enter the function and save the docstring to a
    dictionary
    """
    command_desc_dict = {command: function_name.__doc__
                         for command, function_name in commands.items()}
    return command_desc_dict

def main(test=False):
    """
    :param test: If test is True the last argument from sys.argv is removed,
                 as it contains the directory of the test scripts, which fails the
                 parsing of the arguments as it has an unexpected argument.

                 If test is False then nothing is touched.
    """

    print("Running the configurator")
    # required for running the tests locally or on travis
    # drops the last argument from pytest which is the test file/module
    if test:
        try:
            # find where the /scripts argument is
            index_of_scripts_argument = ["scripts" in arg for arg in sys.argv].index(True)
            # remove it, including every arguments after it (e.g --cov)
            sys.argv = sys.argv[:index_of_scripts_argument]
        except ValueError:
            # scripts was not part of the arguments passed in by the test
            pass

    args = parsers._config_arg_parser(doc=False)
    if args.error:
        utils.error_level = 1

    print("Starting Savu Config tool (please wait for prompt)")

    _reduce_logging_level()

    content = Content(level="advanced" if args.disp_all else 'basic')

    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        # imports all the (working) plugin modules
        content.failed = utils.populate_plugins(error_mode=args.error,
                                                examples=args.examples)

    comp = Completer(commands=commands, plugin_list=pu.plugins)
    utils._set_readline(comp.complete)


    # if file flag is passed then open it here
    if args.file:
        commands['open'](content, args.file)

    print("\n*** Press Enter for a list of available commands. ***\n")

    utils.load_history_file(utils.histfile)
    accumulative_output = ''
    while True:
        try:
            in_text = input(">>> ").strip()
            in_list = in_text.split(' ', 1)
            _write_command_to_log(in_text)

        except KeyboardInterrupt:
            print()
            continue
        except EOFError:
            # makes possible exiting on CTRL + D (EOF, like Python interpreter)
            break

        command, arg = in_list if len(in_list) == 2 else in_list+['']
        command = command if command else 'help'
        if command not in commands:
            print("I'm sorry, that's not a command I recognise. Press Enter "
                  "for a list of available commands.")
        else:
            content = commands[command](content, arg)
            if logger.handlers:
                # Write the command output to a logger
                accumulative_output = _write_output_to_log(accumulative_output)

        if content.is_finished():
            break

    print("Thanks for using the application")


def _write_command_to_log(in_text):
    logger.debug('TEST COMMAND: ' + in_text)
    block_text = _log_file_code('bash')
    logger_rst.debug(block_text + pu.indent('>>> ' + in_text))


def _write_output_to_log(accumulative_output):
    """ Separate the current command output, format it correctly
    and write this to the restructured text log file.

    :param accumulative_output:
    :return: accumulative_output
    """
    current_output = sys.stdout.getvalue()

    # Find all previous command output
    length_Str = len(accumulative_output)
    if length_Str:
        # If there is previous output, then select only new lines from the
        # current command. This will demonstrate clearly the output of each
        # individual command
        current_output = current_output[length_Str:]

    # Characters used for command line colour
    unicode_chars = [u'\x1b[100m', u'\x1b[0m', u'\x1b[97m',
                     u'\x1b[49m', u'\x1b[46m', u'\x1b[39m',
                     u'\x1b[36m']
    # Remove unicode characters which cannot be displayed
    for code in unicode_chars:
        current_output = current_output.replace(code, '')

    # Indent the text for the rst file format
    indent_current_output = pu.indent_multi_line_str(current_output)
    # Write to the rst log file
    logger_rst.debug(indent_current_output)
    return sys.stdout.getvalue()


def _log_file_code(type):
    block_text = '''.. code-block:: '''+str(type)+'''

'''
    return block_text


def _reduce_logging_level():
    import logging
    logging.getLogger().setLevel(logging.CRITICAL)

if __name__ == '__main__':
    main()


1			# Copyright 2014 Diamond Light Source Ltd.
2			#
3			# Licensed under the Apache License, Version 2.0 (the "License");
4			# you may not use this file except in compliance with the License.
5			# You may obtain a copy of the License at
6			#
7			# http://www.apache.org/licenses/LICENSE-2.0
8			#
9			# Unless required by applicable law or agreed to in writing, software
10			# distributed under the License is distributed on an "AS IS" BASIS,
11			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12			# See the License for the specific language governing permissions and
13			# limitations under the License.
14
15			"""
16			.. module:: savu_config
17			:platform: Unix
18			:synopsis: A command line tool for creating Savu plugin lists
19
20			.. moduleauthor:: Nicola Wadeson <[email protected]>
21
22			"""
23
24			import re
25			import sys
26			import logging
27
28			logger = logging.getLogger('documentationLog')
29			logger_rst = logging.getLogger('documentationRst')
30
31			import warnings
32			with warnings.catch_warnings():
33			warnings.simplefilter("ignore")
34			from .content import Content
35			from .completer import Completer
36			from .display_formatter import ListDisplay, DispDisplay, \
37			CiteDisplay
38			from . import arg_parsers as parsers
39			from savu.plugins import utils as pu
40			from . import config_utils as utils
41			from .config_utils import parse_args
42			from .config_utils import error_catcher
43
44
45			def _help(content, args):
46			""" Display the help information"""
47			print('%s Savu configurator commands %s\n' % tuple([''21]*2))
48			for key in list(commands.keys()):
49			doc = commands[key].__doc__
50			if doc:
51			print("%8s : %s" % (key, commands[key].__doc__))
52			print("\n%s\n* For more information about individual commands type "
53			"'<command> -h' \n%s\n" % tuple([''70]2))
54			return content
55
56
57			@parse_args
58			@error_catcher
59			def _open(content, args):
60			""" Open an existing process list."""
61			content.fopen(args.file, update=True, skip=args.skip)
62			_ref(content, '* -n')
63			_disp(content, '-q')
64			return content
65
66
67			@parse_args
68			@error_catcher
69			def _disp(content, args):
70			""" Display the plugins in the current list."""
71			range_dict = content.split_plugin_string(args.start, args.stop,
72			subelem_view=True)
73			formatter = DispDisplay(content.plugin_list)
74			verbosity = parsers._get_verbosity(args)
75			level = 'advanced' if args.all else content.disp_level
76			datasets = True if args.datasets else False
77			content.display(formatter, current_level=level, verbose=verbosity,
78			datasets=datasets, disp_level=args.level, **range_dict)
79			return content
80
81
82			@parse_args
83			@error_catcher
84			def _list(content, args):
85			""" List the available plugins. """
86			list_content = Content()
87			utils._populate_plugin_list(list_content, pfilter=args.string)
88			if not len(list_content.plugin_list.plugin_list):
89			print("No result found.")
90			return content
91
92			# Sort the list of dictionaries.
93			# Sort by the name of the plugin within each dictionary.
94			list_content.plugin_list.plugin_list = \
95			sorted(list_content.plugin_list.plugin_list,
96			key=lambda i: i['name'])
97
98			formatter = ListDisplay(list_content.plugin_list)
99			verbosity = parsers._get_verbosity(args)
100			list_content.display(formatter, verbose=verbosity)
101			return content
102
103
104			@parse_args
105			@error_catcher
106			def _save(content, args):
107			""" Save the current process list to file."""
108			out_file = content.filename if args.input else args.filepath
109			content.check_file(out_file)
110			print()
111			DispDisplay(content.plugin_list)._notices()
112			content.save(out_file, check=input("Are you sure you want to save the "
113			"current data to %s [y/N]" % (out_file)),
114			template=args.template)
115			return content
116
117
118			@parse_args
119			@error_catcher
120			def _mod(content, args):
121			""" Modify plugin parameters."""
122			pos_str, subelem, dims, command = \
123			content.separate_plugin_subelem(args.param, False)
124			if 'expand' in command:
125			# Run the start stop step view for that dimension alone
126			_expand(content, f"{pos_str} {dims} {True}")
127			else:
128			content.check_required_args(args.value, True)
129			# Get the parameter name for the display later
130			args.param = content.get_param_arg_str(pos_str, subelem)
131			content_modified = content.modify(pos_str, subelem,
132			' '.join(args.value),
133			dim=command)
134			if content_modified:
135			# Display the selected parameter only
136			_disp(content, str(args.param))
137			return content
138
139
140			@parse_args
141			@error_catcher
142			def _expand(content, args):
143			""" Expand the plugin preview parameter. """
144			content.set_preview_display(args.off, args.dim, args.dim_view,
145			args.plugin_pos)
146			if content.expand_dim is not None:
147			if content.expand_dim == "all" and args.dim is not None:
148			is_modified = content.modify_dimensions(args.plugin_pos, args.dim)
149			if is_modified:
150			_disp(content, f"{args.plugin_pos}.preview")
151			else:
152			_disp(content, f"{args.plugin_pos}.preview")
153			#content.set_preview_display(args.off, "all")
154			return content
155
156
157			@parse_args
158			@error_catcher
159			def _set(content, args):
160			""" Set the status of the plugin to be on or off. """
161			for pos in args.plugin_pos:
162			content.on_and_off(pos, args.status.upper())
163			_disp(content, '-q')
164			return content
165
166
167			@parse_args
168			@error_catcher
169			def _add(content, args):
170			""" Add a plugin to the list. """
171			elems = content.get_positions()
172			final = str(int(re.findall(r'\d+', elems[-1])[0])+1) if elems else 1
173			content.add(args.name, args.pos if args.pos else str(final))
174			_disp(content, '-q')
175			return content
176
177
178			@parse_args
179			@error_catcher
180			def _ref(content, args):
181			""" Refresh a plugin (update it). """
182			positions = content.get_positions() if args.pos == ['*'] else args.pos
183			for pos_str in positions:
184			content.refresh(pos_str, defaults=args.defaults)
185			if not args.nodisp:
186			_disp(content, pos_str)
187			return content
188
189
190			@parse_args
191			@error_catcher
192			def _cite(content, args):
193			""" Display plugin citations."""
194			range_dict = content.split_plugin_string(args.start, args.stop)
195			formatter = CiteDisplay(content.plugin_list)
196			content.display(formatter, **range_dict)
197			return content
198
199
200			@parse_args
201			@error_catcher
202			def _rem(content, args):
203			""" Remove plugin(s) from the list. """
204			content.remove(content.find_position(args.pos))
205			_disp(content, '-q')
206			return content
207
208
209			@parse_args
210			@error_catcher
211			def _move(content, args):
212			""" Move a plugin to a different position in the list."""
213			content.move(args.orig_pos, args.new_pos)
214			_disp(content, '-q')
215			return content
216
217
218			@parse_args
219			@error_catcher
220			def _coll(content, arg):
221			""" List all plugin collections. """
222			colls = Completer([])._get_collections()
223			print('\n')
224			for c in colls:
225			print('%s\n %s' % ('-'*40, c))
226			print('-'*40, '\n')
227			return content
228
229			@parse_args
230			@error_catcher
231			def _clear(content, arg):
232			""" Clear the current plugin list."""
233			content.clear(check=input("Are you sure you want to clear the current "
234			"plugin list? [y/N]"))
235			return content
236
237			@parse_args
238			@error_catcher
239			def _exit(content, arg):
240			""" Close the program."""
241			content.set_finished(check=input("Are you sure? [y/N]"))
242			return content
243
244			@parse_args
245			@error_catcher
246			def _level(content, args):
247			""" Set a visibility level for the parameters."""
248			content.level(args.level)
249			return content
250
251			@parse_args
252			@error_catcher
253			def _history(content, arg):
254			""" View the history of previous commands """
255			hlen = utils.readline.get_current_history_length()
256			for i in range(hlen):
257			print("%5i : %s" % (i, utils.readline.get_history_item(i)))
258			return content
259
260
261			commands = {'open': _open,
262			'help': _help,
263			'disp': _disp,
264			'list': _list,
265			'save': _save,
266			'mod': _mod,
267			'set': _set,
268			'add': _add,
269			'rem': _rem,
270			'move': _move,
271			'ref': _ref,
272			'level': _level,
273			'expand': _expand,
274			'cite': _cite,
275			'coll': _coll,
276			'clear': _clear,
277			'exit': _exit,
278			'history': _history}
279
280			def get_description():
281			""" For each command, enter the function and save the docstring to a
282			dictionary
283			"""
284			command_desc_dict = {command: function_name.__doc__
285			for command, function_name in commands.items()}
286			return command_desc_dict
287
288			def main(test=False):
289			"""
290			:param test: If test is True the last argument from sys.argv is removed,
291			as it contains the directory of the test scripts, which fails the
292			parsing of the arguments as it has an unexpected argument.
293
294			If test is False then nothing is touched.
295			"""
296
297			print("Running the configurator")
298			# required for running the tests locally or on travis
299			# drops the last argument from pytest which is the test file/module
300			if test:
301			try:
302			# find where the /scripts argument is
303			index_of_scripts_argument = ["scripts" in arg for arg in sys.argv].index(True)
304			# remove it, including every arguments after it (e.g --cov)
305			sys.argv = sys.argv[:index_of_scripts_argument]
306			except ValueError:
307			# scripts was not part of the arguments passed in by the test
308			pass
309
310			args = parsers._config_arg_parser(doc=False)
311			if args.error:
312			utils.error_level = 1
313
314			print("Starting Savu Config tool (please wait for prompt)")
315
316			_reduce_logging_level()
317
318			content = Content(level="advanced" if args.disp_all else 'basic')
319
320			with warnings.catch_warnings():
321			warnings.simplefilter("ignore")
322			# imports all the (working) plugin modules
323			content.failed = utils.populate_plugins(error_mode=args.error,
324			examples=args.examples)
325
326			comp = Completer(commands=commands, plugin_list=pu.plugins)
327			utils._set_readline(comp.complete)
328
329
330			# if file flag is passed then open it here
331			if args.file:
332			commands['open'](content, args.file)
333
334			print("\n* Press Enter for a list of available commands. *\n")
335
336			utils.load_history_file(utils.histfile)
337			accumulative_output = ''
338			while True:
339			try:
340			in_text = input(">>> ").strip()
341			in_list = in_text.split(' ', 1)
342			_write_command_to_log(in_text)
343
344			except KeyboardInterrupt:
345			print()
346			continue
347			except EOFError:
348			# makes possible exiting on CTRL + D (EOF, like Python interpreter)
349			break
350
351			command, arg = in_list if len(in_list) == 2 else in_list+['']
352			command = command if command else 'help'
353			if command not in commands:
354			print("I'm sorry, that's not a command I recognise. Press Enter "
355			"for a list of available commands.")
356			else:
357			content = commands[command](content, arg)
358			if logger.handlers:
359			# Write the command output to a logger
360			accumulative_output = _write_output_to_log(accumulative_output)
361
362			if content.is_finished():
363			break
364
365			print("Thanks for using the application")
366
367
368			def _write_command_to_log(in_text):
369			logger.debug('TEST COMMAND: ' + in_text)
370			block_text = _log_file_code('bash')
371			logger_rst.debug(block_text + pu.indent('>>> ' + in_text))
372
373
374			def _write_output_to_log(accumulative_output):
375			""" Separate the current command output, format it correctly
376			and write this to the restructured text log file.
377
378			:param accumulative_output:
379			:return: accumulative_output
380			"""
381			current_output = sys.stdout.getvalue()
382
383			# Find all previous command output
384			length_Str = len(accumulative_output)
385			if length_Str:
386			# If there is previous output, then select only new lines from the
387			# current command. This will demonstrate clearly the output of each
388			# individual command
389			current_output = current_output[length_Str:]
390
391			# Characters used for command line colour
392			unicode_chars = [u'\x1b[100m', u'\x1b[0m', u'\x1b[97m',
393			u'\x1b[49m', u'\x1b[46m', u'\x1b[39m',
394			u'\x1b[36m']
395			# Remove unicode characters which cannot be displayed
396			for code in unicode_chars:
397			current_output = current_output.replace(code, '')
398
399			# Indent the text for the rst file format
400			indent_current_output = pu.indent_multi_line_str(current_output)
401			# Write to the rst log file
402			logger_rst.debug(indent_current_output)
403			return sys.stdout.getvalue()
404
405
406			def _log_file_code(type):
407			block_text = '''.. code-block:: '''+str(type)+'''
408
409			'''
410			return block_text
411
412
413			def _reduce_logging_level():
414			import logging
415			logging.getLogger().setLevel(logging.CRITICAL)
416
417			if __name__ == '__main__':
418			main()
419

DiamondLightSource / Savu

Push — master ( 2fd334...552d1f )

scripts.config_generator.savu_config._rem() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like