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
|
#!/usr/bin/env python
# encoding: utf-8
"""
convert -- Convert PROV-JSON to RDF, PROV-N, PROV-XML, or graphical formats (SVG, PDF, PNG)
@author: Trung Dong Huynh
@copyright: 2016 University of Southampton, United Kingdom. All rights reserved.
@license: MIT License
@contact: trungdong@donggiang.com
@deffield updated: 2016-10-19
"""
from argparse import ArgumentParser, RawDescriptionHelpFormatter, FileType
import os
import sys
import logging
import traceback
from prov.model import ProvDocument
from prov import serializers
logger = logging.getLogger(__name__)
__all__ = []
__version__ = 0.1
__date__ = '2014-03-14'
__updated__ = '2016-10-19'
DEBUG = 0
TESTRUN = 0
PROFILE = 0
GRAPHVIZ_SUPPORTED_FORMATS = {
'bmp', 'canon', 'cmap', 'cmapx', 'cmapx_np', 'dot', 'eps', 'fig', 'gtk', 'gv', 'ico', 'imap', 'imap_np', 'ismap',
'jpe', 'jpeg', 'jpg', 'pdf', 'plain', 'plain-ext', 'png', 'ps', 'ps2', 'svg', 'svgz', 'tif', 'tiff', 'tk',
'vml', 'vmlz', 'x11', 'xdot', 'xlib'
}
class CLIError(Exception):
"""Generic exception to raise and log different fatal errors."""
def __init__(self, msg):
super(CLIError).__init__(type(self))
self.msg = "E: %s" % msg
def __str__(self):
return self.msg
def convert_file(infile, outfile, output_format):
prov_doc = ProvDocument.deserialize(infile)
# Formats not supported by prov.serializers
if output_format == 'provn':
outfile.write(prov_doc.get_provn().encode())
elif output_format in GRAPHVIZ_SUPPORTED_FORMATS:
from prov.dot import prov_to_dot
dot = prov_to_dot(prov_doc)
content = dot.create(format=output_format)
outfile.write(content)
else:
# Try supported serializers:
try:
prov_doc.serialize(outfile, format=output_format)
except serializers.DoNotExist:
raise CLIError('Output format "%s" is not supported.' % output_format)
def main(argv=None): # IGNORE:C0111
"""Command line options."""
if argv is None:
argv = sys.argv
else:
sys.argv.extend(argv)
program_name = os.path.basename(sys.argv[0])
program_version = "v%s" % __version__
program_build_date = str(__updated__)
program_version_message = '%%(prog)s %s (%s)' % (program_version, program_build_date)
program_shortdesc = __import__('__main__').__doc__.split("\n")[1]
program_license = '''%s
Created by Trung Dong Huynh on %s.
Copyright 2016 University of Southampton. All rights reserved.
Licensed under the MIT License
https://github.com/trungdong/prov/blob/master/LICENSE
Distributed on an "AS IS" basis without warranties
or conditions of any kind, either express or implied.
USAGE
''' % (program_shortdesc, str(__date__))
try:
# Setup argument parser
parser = ArgumentParser(description=program_license, formatter_class=RawDescriptionHelpFormatter)
parser.add_argument('-f', '--format', dest='format', action='store', default='json',
help='output format: json, xml, provn, or one supported by GraphViz (e.g. svg, pdf)')
parser.add_argument('infile', nargs='?', type=FileType('r'), default=sys.stdin)
parser.add_argument('outfile', nargs='?', type=FileType('wb'), default=sys.stdout)
parser.add_argument('-V', '--version', action='version', version=program_version_message)
args = None
try:
# Process arguments
args = parser.parse_args()
convert_file(args.infile, args.outfile, args.format.lower())
finally:
if args:
if args.infile:
args.infile.close()
if args.outfile:
args.outfile.close()
return 0
except KeyboardInterrupt:
# handle keyboard interrupt
return 0
except Exception as e:
if DEBUG or TESTRUN:
traceback.print_exc()
raise e
indent = len(program_name) * " "
sys.stderr.write(program_name + ": " + str(e) + "\n")
sys.stderr.write(indent + " for help use --help")
return 2
if __name__ == "__main__":
logging.basicConfig(level=(logging.DEBUG if DEBUG else logging.INFO))
if TESTRUN:
import doctest
doctest.testmod()
if PROFILE:
import cProfile
import pstats
profile_filename = 'converter_profile.txt'
cProfile.run('main()', profile_filename)
statsfile = open("profile_stats.txt", "wb")
p = pstats.Stats(profile_filename, stream=statsfile)
stats = p.strip_dirs().sort_stats('cumulative')
stats.print_stats()
statsfile.close()
sys.exit(0)
sys.exit(main())
|
