-
Notifications
You must be signed in to change notification settings - Fork 3
/
api.py
121 lines (92 loc) · 3.23 KB
/
api.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
import re
import yaml
import tempfile
import subprocess
import networkx as nx
import settings
TYPES = {
'svg': 'image/svg+xml',
'png': 'image/png',
'pdf': 'application/pdf',
'xml': 'application/xml',
}
class JujuSVGException(Exception):
def __init__(self, cmd, msg):
self.cmd = cmd
self.msg = msg
class BundleFormatException(Exception):
def __init__(self, msg):
self.msg = msg
def parse_bundle_id(bundle_id):
m = re.match(r'cs(:~[a-z0-9-_]+/|:)bundle/[a-z0-9-]+-[0-9]+', bundle_id)
if m:
bundle_path = m.group(0).replace('cs:', '')
else:
bundle_path = 'bundle/{0}'.format(bundle_id)
return ('https://api.jujucharms.com/v5/%s/archive/bundle.yaml' % bundle_path,
'https://api.jujucharms.com/v5/%s/diagram.svg' % bundle_path)
def split_rel(r):
return r.split(':', 1)
def mapply(func, g, **kwargs):
args = {}
for name in func.__code__.co_varnames:
if name in kwargs:
args[name] = kwargs[name]
return func(g, **args)
# https://gist.github.com/bcsaller/adca309ba7abef2e8caf#file-place_bundle-py-L46
def layout(bundle, algo, scale=500.0):
g = nx.MultiGraph()
for service in bundle['services']:
g.add_node(service)
for relation in bundle['relations']:
src = split_rel(relation[0])[0]
tgts = relation[1]
if isinstance(tgts, str):
tgts = [tgts]
for tgt in tgts:
tgt = split_rel(tgt)[0]
g.add_edge(src, tgt)
pos = mapply(algo, g, k=45, iterations=100)
for service, data in list(bundle['services'].items()):
data['annotations'] = {
"gui-x": float(pos[service][0]) * scale,
"gui-y": float(pos[service][1]) * scale,
}
return g
def process_bundle(bundle):
if len(bundle) > 1 and 'services' not in bundle:
raise BundleFormatException('This has multiple deployments')
if 'services' not in bundle:
if 'services' not in next(iter(bundle.values())):
raise BundleFormatException("This probably isn't a bundle...")
bundle = next(iter(bundle.values()))
annotations = False
for service, srvc_data in bundle['services'].items():
if 'annotations' in list(srvc_data.keys()):
annotations = True
break
if not annotations:
#layout = 'circular'
#algo = getattr(nx, layout + '_layout', None)
layout(bundle, nx.circular_layout)
with tempfile.NamedTemporaryFile(mode='w') as f:
f.write(yaml.dump(bundle, default_flow_style=False))
f.flush()
try:
cmd = [settings.JUJSVG, f.name]
svg = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
raise JujuSVGException(' '.join(e.cmd), e.output)
return svg
def output_bundle(svg, format='svg'):
if format == 'svg':
return svg
with tempfile.NamedTemporaryFile() as f:
f.write(svg)
f.flush()
try:
cmd = [settings.RSVG, '-f', format, f.name]
out = subprocess.check_output(cmd)
except subprocess.CalledProcessError as e:
raise JujuSVGException(' '.join(e.cmd), e.output)
return out