-
Notifications
You must be signed in to change notification settings - Fork 0
/
cli.py
244 lines (195 loc) · 8.24 KB
/
cli.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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
130
131
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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
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
233
234
235
236
237
238
239
240
241
242
243
244
#!/usr/bin/env python
"""Command-line interface."""
import sys
import time
import argparse
from . import CLI, VERSION, DESCRIPTION
from . import common
from . import services
from .data import Data
from .manager import get_manager
import yorm
log = common.logger(__name__)
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")
shared = {'formatter_class': common.HelpFormatter,
'parents': [debug]}
# Build main parser
parser = argparse.ArgumentParser(prog=CLI, description=DESCRIPTION,
**shared)
parser.add_argument('-d', '--daemon', metavar='DELAY', nargs='?', const=60,
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, **shared)
sub.add_argument('name', nargs='?',
help="computer to queue for launch (default: current)")
# Build clean parser
info = "display and delete conflicted files"
sub = subs.add_parser('clean', description=info.capitalize() + '.',
help=info, **shared)
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 == '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 cancelled"
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,
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 delete: attempt to delete conflicted files
:param force: actually delete conflicted files
"""
manager = get_manager()
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 cleanup:
clean(config, status)
if delete:
return services.delete_conflicts(root, force=force)
if switch is True:
switch = computer
elif switch:
switch = config.computers.match(switch)
if switch:
queue(config, status, switch)
while True:
launch(config, status, computer, manager)
update(config, status, computer, manager)
if delay is None:
break
log.info("delaying for %s seconds...", delay)
time.sleep(delay)
return True
# TODO: consider moving this logic to `data`
def clean(config, status):
"""Remove undefined applications and computers."""
log.info("cleaning up applications and computers...")
for appstatus in status.applications.copy():
if not config.applications.find(appstatus.application):
status.applications.remove(appstatus)
log.info("removed application: %s", appstatus)
else:
for computerstate in appstatus.computers.copy():
if not config.computers.find(computerstate.computer):
appstatus.computers.remove(computerstate)
log.info("removed computer: %s", computerstate)
# TODO: consider moving this logic to `data`
def queue(config, status, computer):
"""Queue applications for launch."""
log.info("queuing applications for launch...")
for application in config.applications:
if application.queued:
log.debug("queuing %s on %s...", application, computer)
status.queue(application, computer)
else:
log.debug("skipping %s (not queued)...", application)
# TODO: consider moving this logic to `data`
def launch(config, status, computer, manager):
"""Launch applications that have been queued."""
log.info("launching queued applications...")
for app_status in status.applications:
if app_status.next:
application = config.applications.get(app_status.application)
show_queued(application, app_status.next)
if app_status.next == computer:
latest = status.get_latest(application)
if latest in (computer, None):
if not manager.is_running(application):
manager.start(application)
app_status.next = None
else:
show_waiting(application, latest)
elif manager.is_running(application):
manager.stop(application)
# TODO: consider moving this logic to `data`
def update(config, status, computer, manager):
"""Update each application's status."""
log.info("recording application status...")
for application in config.applications:
if manager.is_running(application):
latest = status.get_latest(application)
if computer != latest:
if status.is_running(application, computer):
# case 1: application just launched remotely
manager.stop(application)
status.stop(application, computer)
show_running(application, latest)
show_stopped(application, computer)
else:
# case 2: application just launched locally
status.start(application, computer)
show_running(application, computer)
else:
# case 3: application already running locally
pass
else:
if status.is_running(application, computer):
# case 4: application just closed locally
status.stop(application, computer)
show_stopped(application, computer)
else:
# case 5: application already closed locally
pass
def show_queued(application, computer):
"""Display the state of a queued application."""
print("{} is queued for {}".format(application, computer))
def show_waiting(application, computer):
"""Display the old state of a running application."""
print("{} is still running on {}".format(application, computer))
def show_running(application, computer):
"""Display the new state of a running application."""
print("{} is now running on {}".format(application, computer))
def show_started(application, computer):
"""Display the new state of a started application."""
print("{} is now started on {}".format(application, computer))
def show_stopped(application, computer):
"""Display the new state of a stopped application."""
print("{} is now stopped on {}".format(application, computer))
if __name__ == '__main__': # pragma: no cover (manual test)
main()