mine.cli.run() - Code Metrics - Inspection of "Release v1.6" - jacebrowning/mine - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#89)

by Jace

created 2019-05-21 02:25 UTC

mine.cli.run() F

↳ Parent: mine.cli

Complexity

Conditions

Size

Total Lines	83
Code Lines	53

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	25
CRAP Score	31.5036

Importance

Changes

Metric	Value
cc	15
eloc	53
nop	7
dl	0
loc	83
ccs	25
cts	43
cp	0.5814
crap	31.5036
rs	2.9998
c	0
b	0
f	0

How to fix Long Method Complexity

#!/usr/bin/env python

"""Command-line interface."""

import argparse
import subprocess
import sys
import time

import log
import yorm

from . import CLI, DESCRIPTION, VERSION, common, services
from .manager import get_manager
from .models import Application, Data


daemon = Application(CLI, filename=CLI)


def main(args=None):
    """Process command-line arguments and run the program."""

    # Shared options
    debug = argparse.ArgumentParser(add_help=False)
    debug.add_argument('-V', '--version', action='version', version=VERSION)
    group = debug.add_mutually_exclusive_group()
    group.add_argument(
        '-v', '--verbose', action='count', default=0, help="enable verbose logging"
    )
    group.add_argument(
        '-q',
        '--quiet',
        action='store_const',
        const=-1,
        dest='verbose',
        help="only display errors and prompts",
    )

    # Build main parser
    parser = argparse.ArgumentParser(
        prog=CLI,
        description=DESCRIPTION,
        formatter_class=common.HelpFormatter,
        parents=[debug],
    )
    parser.add_argument(
        '-d',
        '--daemon',
        metavar='DELAY',
        nargs='?',
        const=300,
        type=int,
        help="run continuously with delay [seconds]",
    )
    parser.add_argument('-f', '--file', help="custom settings file path")
    subs = parser.add_subparsers(help="", dest='command', metavar="<command>")

    # Build switch parser
    info = "start applications on another computer"
    sub = subs.add_parser(
        'switch',
        description=info.capitalize() + '.',
        help=info,
        formatter_class=common.HelpFormatter,
        parents=[debug],
    )
    sub.add_argument(
        'name', nargs='?', help="computer to queue for launch (default: current)"
    )

    # Build close parser
    info = "close applications on this computer"
    sub = subs.add_parser(
        'close',
        description=info.capitalize() + '.',
        help=info,
        formatter_class=common.HelpFormatter,
        parents=[debug],
    )

    # Build edit parser
    info = "launch the settings file for editing"
    sub = subs.add_parser(
        'edit',
        description=info.capitalize() + '.',
        help=info,
        formatter_class=common.HelpFormatter,
        parents=[debug],
    )

    # Build clean parser
    info = "display and delete conflicted files"
    sub = subs.add_parser(
        'clean',
        description=info.capitalize() + '.',
        help=info,
        formatter_class=common.HelpFormatter,
        parents=[debug],
    )
    sub.add_argument(
        '-f',
        '--force',
        action='store_true',
        help="actually delete the conflicted files",
    )

    # Parse arguments
    args = parser.parse_args(args=args)
    kwargs = {'delay': args.daemon}
    if args.command == 'switch':
        kwargs['switch'] = args.name if args.name else True
    elif args.command == 'close':
        kwargs['switch'] = False
    elif args.command == 'edit':
        kwargs['edit'] = True
    elif args.command == 'clean':
        kwargs['delete'] = True
        kwargs['force'] = args.force

    # Configure logging
    common.configure_logging(args.verbose)

    # Run the program
    try:
        log.debug("Running main command...")
        success = run(path=args.file, **kwargs)
    except KeyboardInterrupt:
        msg = "command canceled"
        if common.verbosity == common.MAX_VERBOSITY:
            log.exception(msg)
        else:
            log.debug(msg)
        success = False
    if success:
        log.debug("Command succeeded")
    else:
        log.debug("Command failed")
        sys.exit(1)


def run(
    path=None,
    cleanup=True,
    delay=None,
    switch=None,
    edit=False,
    delete=False,
    force=False,
):
    """Run the program.

    :param path: custom settings file path
    :param cleanup: remove unused items from the config
    :param delay: number of seconds to delay before repeating

    :param switch: computer name to queue for launch

    :param edit: launch the configuration file for editing

    :param delete: attempt to delete conflicted files
    :param force: actually delete conflicted files

    """  # pylint: disable=too-many-arguments,too-many-branches
    manager = get_manager()
    if not manager.is_running(services.APPLICATION):
        manager.start(services.APPLICATION)

    root = services.find_root()
    path = path or services.find_config_path(root=root)

    data = Data()
    yorm.sync(data, path)

    config = data.config
    status = data.status

    log.info("Identifying current computer...")
    computer = config.computers.get_current()
    log.info("Current computer: %s", computer)

    if edit:
        return manager.launch(path)
    if delete:
        return services.delete_conflicts(root, force=force)

    if switch is True:
        switch = computer
    elif switch is False:
        data.close_all_applications(config, manager)
    elif switch:
        switch = config.computers.match(switch)

    if switch:
        if switch != computer:
            data.close_all_applications(config, manager)
        data.queue_all_applications(config, status, switch)

    while True:
        data.launch_queued_applications(config, status, computer, manager)
        data.update_status(config, status, computer, manager)

        if delay is None:
            break

        log.info("Delaying %s seconds for files to sync...", delay)
        time.sleep(delay)

        step = 1
        elapsed = 0
        log.info("Waiting %s seconds for status changes...", delay)
        while elapsed < delay and not data.modified:
            time.sleep(step)
            elapsed += step

        services.delete_conflicts(root, config_only=True, force=True)

    if cleanup:
        data.prune_status(config, status)

    if delay is None:
        return _restart_daemon(manager)

    return True


def _restart_daemon(manager):
    cmd = "nohup {} --daemon --verbose >> /tmp/mine.log 2>&1 &".format(CLI)
    if daemon and not manager.is_running(daemon):
        log.warning("Daemon is not running, attempting to restart...")

        log.info("$ %s", cmd)
        subprocess.call(cmd, shell=True)
        if manager.is_running(daemon):
            return True

        log.error("Manually start daemon: %s", cmd)
        return False

    return True


if __name__ == '__main__':  # pragma: no cover (manual test)
    main()


1		#!/usr/bin/env python
2
3	1	"""Command-line interface."""
4
5	1	import argparse
6	1	import subprocess
7	1	import sys
8	1	import time
9	1
10		import log
11	1	import yorm
12
13	1	from . import CLI, DESCRIPTION, VERSION, common, services
14	1	from .manager import get_manager
15	1	from .models import Application, Data
16	1
17	1
18		daemon = Application(CLI, filename=CLI)
19
20	1
21	1	def main(args=None):
22		"""Process command-line arguments and run the program."""
23
24	1	# Shared options
25		debug = argparse.ArgumentParser(add_help=False)
26		debug.add_argument('-V', '--version', action='version', version=VERSION)
27		group = debug.add_mutually_exclusive_group()
28	1	group.add_argument(
29	1	'-v', '--verbose', action='count', default=0, help="enable verbose logging"
30	1	)
31	1	group.add_argument(
32		'-q',
33	1	'--quiet',
34		action='store_const',
35	1	const=-1,
36		dest='verbose',
37		help="only display errors and prompts",
38		)
39	1
40		# Build main parser
41	1	parser = argparse.ArgumentParser(
42		prog=CLI,
43	1	description=DESCRIPTION,
44	1	formatter_class=common.HelpFormatter,
45		parents=[debug],
46		)
47	1	parser.add_argument(
48	1	'-d',
49		'--daemon',
50	1	metavar='DELAY',
51		nargs='?',
52		const=300,
53		type=int,
54	1	help="run continuously with delay [seconds]",
55	1	)
56		parser.add_argument('-f', '--file', help="custom settings file path")
57		subs = parser.add_subparsers(help="", dest='command', metavar="<command>")
58
59	1	# Build switch parser
60	1	info = "start applications on another computer"
61		sub = subs.add_parser(
62		'switch',
63		description=info.capitalize() + '.',
64	1	help=info,
65	1	formatter_class=common.HelpFormatter,
66		parents=[debug],
67	1	)
68		sub.add_argument(
69		'name', nargs='?', help="computer to queue for launch (default: current)"
70		)
71	1
72	1	# Build close parser
73	1	info = "close applications on this computer"
74	1	sub = subs.add_parser(
75	1	'close',
76		description=info.capitalize() + '.',
77	1	help=info,
78	1	formatter_class=common.HelpFormatter,
79	1	parents=[debug],
80	1	)
81	1
82		# Build edit parser
83		info = "launch the settings file for editing"
84	1	sub = subs.add_parser(
85		'edit',
86		description=info.capitalize() + '.',
87	1	help=info,
88	1	formatter_class=common.HelpFormatter,
89	1	parents=[debug],
90	1	)
91	1
92	1	# Build clean parser
93	1	info = "display and delete conflicted files"
94		sub = subs.add_parser(
95	1	'clean',
96	1	description=info.capitalize() + '.',
97	1	help=info,
98	1	formatter_class=common.HelpFormatter,
99		parents=[debug],
100	1	)
101	1	sub.add_argument(
102		'-f',
103		'--force',
104	1	action='store_true',
105		help="actually delete the conflicted files",
106		)
107
108		# Parse arguments
109		args = parser.parse_args(args=args)
110		kwargs = {'delay': args.daemon}
111		if args.command == 'switch':
112		kwargs['switch'] = args.name if args.name else True
113		elif args.command == 'close':
114		kwargs['switch'] = False
115		elif args.command == 'edit':
116		kwargs['edit'] = True
117		elif args.command == 'clean':
118		kwargs['delete'] = True
119		kwargs['force'] = args.force
120
121		# Configure logging
122	1	common.configure_logging(args.verbose)
123	1
124	1	# Run the program
125		try:
126	1	log.debug("Running main command...")
127	1	success = run(path=args.file, **kwargs)
128		except KeyboardInterrupt:
129	1	msg = "command canceled"
130	1	if common.verbosity == common.MAX_VERBOSITY:
131		log.exception(msg)
132	1	else:
133	1	log.debug(msg)
134		success = False
135	1	if success:
136	1	log.debug("Command succeeded")
137	1	else:
138		log.debug("Command failed")
139	1	sys.exit(1)
140
141	1
142		def run(
143	1	path=None,
144		cleanup=True,
145		delay=None,
146	1	switch=None,
147		edit=False,
148	1	delete=False,
149		force=False,
150	1	):
151		"""Run the program.
152
153	1	:param path: custom settings file path
154		:param cleanup: remove unused items from the config
155		:param delay: number of seconds to delay before repeating
156
157		:param switch: computer name to queue for launch
158	1
159	1	:param edit: launch the configuration file for editing
160	1
161		:param delete: attempt to delete conflicted files
162	1	:param force: actually delete conflicted files
163	1
164		""" # pylint: disable=too-many-arguments,too-many-branches
165		manager = get_manager()
166		if not manager.is_running(services.APPLICATION):
167		manager.start(services.APPLICATION)
168
169		root = services.find_root()
170		path = path or services.find_config_path(root=root)
171
172		data = Data()
173		yorm.sync(data, path)
174
175		config = data.config
176		status = data.status
177	1
178	1	log.info("Identifying current computer...")
179		computer = config.computers.get_current()
180	1	log.info("Current computer: %s", computer)
181	1
182		if edit:
183		return manager.launch(path)
184		if delete:
185		return services.delete_conflicts(root, force=force)
186	1
187	1	if switch is True:
188	1	switch = computer
189	1	elif switch is False:
190		data.close_all_applications(config, manager)
191	1	elif switch:
192	1	switch = config.computers.match(switch)
193	1
194	1	if switch:
195		if switch != computer:
196	1	data.close_all_applications(config, manager)
197	1	data.queue_all_applications(config, status, switch)
198
199	1	while True:
200		data.launch_queued_applications(config, status, computer, manager)
201		data.update_status(config, status, computer, manager)
202
203		if delay is None:
204		break
205
206		log.info("Delaying %s seconds for files to sync...", delay)
207		time.sleep(delay)
208
209		step = 1
210		elapsed = 0
211		log.info("Waiting %s seconds for status changes...", delay)
212		while elapsed < delay and not data.modified:
213		time.sleep(step)
214		elapsed += step
215
216		services.delete_conflicts(root, config_only=True, force=True)
217
218		if cleanup:
219		data.prune_status(config, status)
220
221		if delay is None:
222		return _restart_daemon(manager)
223
224		return True
225
226
227		def _restart_daemon(manager):
228		cmd = "nohup {} --daemon --verbose >> /tmp/mine.log 2>&1 &".format(CLI)
229		if daemon and not manager.is_running(daemon):
230		log.warning("Daemon is not running, attempting to restart...")
231
232		log.info("$ %s", cmd)
233		subprocess.call(cmd, shell=True)
234		if manager.is_running(daemon):
235		return True
236
237		log.error("Manually start daemon: %s", cmd)
238		return False
239
240		return True
241
242
243		if __name__ == '__main__': # pragma: no cover (manual test)
244		main()
245

jacebrowning / mine

Pull Request — master (#89)

mine.cli.run() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like