Converting workshop checker to use Reporter class.
[rnaseq-cwl-training.git] / bin / util.py
1 import sys
2 import json
3 import yaml
4 from subprocess import Popen, PIPE
5
6
7 class Reporter(object):
8     """Collect and report errors."""
9
10     def __init__(self):
11         """Constructor."""
12
13         super(Reporter, self).__init__()
14         self.messages = []
15
16
17     def check_field(self, filename, name, values, key, expected):
18         """Check that a dictionary has an expected value."""
19
20         if key not in values:
21             self.add(filename, '{0} does not contain {1}', name, key)
22         elif values[key] != expected:
23             self.add(filename, '{0} {1} is {2} not {3}', name, key, values[key], expected)
24
25
26     def check(self, condition, location, fmt, *args):
27         """Append error if condition not met."""
28
29         if not condition:
30             self.add(location, fmt, *args)
31
32
33     def add(self, location, fmt, *args):
34         """Append error unilaterally."""
35
36         if isinstance(location, type(None)):
37             coords = ''
38         elif isinstance(location, str):
39             coords = '{0}: '.format(location)
40         elif isinstance(location, tuple):
41             filename, line_number = location
42             coords = '{0}:{1}: '.format(*location)
43         else:
44             assert False, 'Unknown location "{0}"/{1}'.format(location, type(location))
45
46         self.messages.append(coords + fmt.format(*args))
47
48
49     def report(self, stream=sys.stdout):
50         """Report all messages."""
51
52         if not self.messages:
53             return
54         for m in sorted(self.messages):
55             print(m, file=stream)
56
57
58 def read_markdown(parser, path):
59     """
60     Get YAML and AST for Markdown file, returning
61     {'metadata':yaml, 'metadata_len':N, 'text':text, 'lines':[(i, line, len)], 'doc':doc}.
62     """
63
64     # Split and extract YAML (if present).
65     with open(path, 'r') as reader:
66         body = reader.read()
67     metadata_raw, metadata_yaml, body = split_metadata(path, body)
68
69     # Split into lines.
70     metadata_len = 0 if metadata_raw is None else metadata_raw.count('\n')
71     lines = [(metadata_len+i+1, line, len(line)) for (i, line) in enumerate(body.split('\n'))]
72
73     # Parse Markdown.
74     cmd = 'ruby {0}'.format(parser)
75     p = Popen(cmd, shell=True, stdin=PIPE, stdout=PIPE, close_fds=True, universal_newlines=True)
76     stdout_data, stderr_data = p.communicate(body)
77     doc = json.loads(stdout_data)
78
79     return {
80         'metadata': metadata_yaml,
81         'metadata_len': metadata_len,
82         'text': body,
83         'lines': lines,
84         'doc': doc
85     }
86
87
88 def split_metadata(path, text):
89     """
90     Get raw (text) metadata, metadata as YAML, and rest of body.
91     If no metadata, return (None, None, body).
92     """
93
94     metadata_raw = None
95     metadata_yaml = None
96     metadata_len = None
97
98     pieces = text.split('---', 2)
99     if len(pieces) == 3:
100         metadata_raw = pieces[1]
101         text = pieces[2]
102         try:
103             metadata_yaml = yaml.load(metadata_raw)
104         except yaml.YAMLError as e:
105             print('Unable to parse YAML header in {0}:\n{1}'.format(path, e), file=sys.stderr)
106             sys.exit(1)
107
108     return metadata_raw, metadata_yaml, text