df350e7a7895bdfc656b1e64c3a66eead0a715d0
[rnaseq-cwl-training.git] / bin / util.py
1 import sys
2 import os
3 import json
4 from subprocess import Popen, PIPE
5
6 # Import this way to produce a more useful error message.
7 try:
8     import yaml
9 except ImportError:
10     print('Unable to import YAML module: please install PyYAML', file=sys.stderr)
11     sys.exit(1)
12
13
14 UNWANTED_FILES = [
15     '.nojekyll'
16 ]
17
18
19 REPORTER_NOT_SET = []
20
21 class Reporter(object):
22     """Collect and report errors."""
23
24     def __init__(self):
25         """Constructor."""
26
27         super(Reporter, self).__init__()
28         self.messages = []
29
30
31     def check_field(self, filename, name, values, key, expected=REPORTER_NOT_SET):
32         """Check that a dictionary has an expected value."""
33
34         if key not in values:
35             self.add(filename, '{0} does not contain {1}', name, key)
36         elif expected is REPORTER_NOT_SET:
37             pass
38         elif type(expected) in (tuple, set, list):
39             if values[key] not in expected:
40                 self.add(filename, '{0} {1} value {2} is not in {3}', name, key, values[key], expected)
41         elif values[key] != expected:
42             self.add(filename, '{0} {1} is {2} not {3}', name, key, values[key], expected)
43
44
45     def check(self, condition, location, fmt, *args):
46         """Append error if condition not met."""
47
48         if not condition:
49             self.add(location, fmt, *args)
50
51
52     def add(self, location, fmt, *args):
53         """Append error unilaterally."""
54
55         if isinstance(location, type(None)):
56             coords = ''
57         elif isinstance(location, str):
58             coords = '{0}: '.format(location)
59         elif isinstance(location, tuple):
60             filename, line_number = location
61             coords = '{0}:{1}: '.format(*location)
62         else:
63             assert False, 'Unknown location "{0}"/{1}'.format(location, type(location))
64
65         self.messages.append(coords + fmt.format(*args))
66
67
68     def report(self, stream=sys.stdout):
69         """Report all messages."""
70
71         if not self.messages:
72             return
73         for m in sorted(self.messages):
74             print(m, file=stream)
75
76
77 def read_markdown(parser, path):
78     """
79     Get YAML and AST for Markdown file, returning
80     {'metadata':yaml, 'metadata_len':N, 'text':text, 'lines':[(i, line, len)], 'doc':doc}.
81     """
82
83     # Split and extract YAML (if present).
84     with open(path, 'r') as reader:
85         body = reader.read()
86     metadata_raw, metadata_yaml, body = split_metadata(path, body)
87
88     # Split into lines.
89     metadata_len = 0 if metadata_raw is None else metadata_raw.count('\n')
90     lines = [(metadata_len+i+1, line, len(line)) for (i, line) in enumerate(body.split('\n'))]
91
92     # Parse Markdown.
93     cmd = 'ruby {0}'.format(parser)
94     p = Popen(cmd, shell=True, stdin=PIPE, stdout=PIPE, close_fds=True, universal_newlines=True)
95     stdout_data, stderr_data = p.communicate(body)
96     doc = json.loads(stdout_data)
97
98     return {
99         'metadata': metadata_yaml,
100         'metadata_len': metadata_len,
101         'text': body,
102         'lines': lines,
103         'doc': doc
104     }
105
106
107 def split_metadata(path, text):
108     """
109     Get raw (text) metadata, metadata as YAML, and rest of body.
110     If no metadata, return (None, None, body).
111     """
112
113     metadata_raw = None
114     metadata_yaml = None
115     metadata_len = None
116
117     pieces = text.split('---', 2)
118     if len(pieces) == 3:
119         metadata_raw = pieces[1]
120         text = pieces[2]
121         try:
122             metadata_yaml = yaml.load(metadata_raw)
123         except yaml.YAMLError as e:
124             print('Unable to parse YAML header in {0}:\n{1}'.format(path, e), file=sys.stderr)
125             sys.exit(1)
126
127     return metadata_raw, metadata_yaml, text
128
129
130 def load_yaml(filename):
131     """
132     Wrapper around YAML loading so that 'import yaml' is only needed
133     in one file.
134     """
135
136     try:
137         with open(filename, 'r') as reader:
138             return yaml.load(reader)
139     except (yaml.YAMLError, FileNotFoundError) as e:
140         print('Unable to load YAML file {0}:\n{1}'.format(filename, e), file=sys.stderr)
141         sys.exit(1)
142
143
144 def check_unwanted_files(dir_path, reporter):
145     """
146     Check that unwanted files are not present.
147     """
148
149     for filename in UNWANTED_FILES:
150         path = os.path.join(dir_path, filename)
151         reporter.check(not os.path.exists(path),
152                        path,
153                        "Unwanted file found")
154
155
156 def require(condition, message):
157     """Fail if condition not met."""
158
159     if not condition:
160         print(message, file=sys.stderr)
161         sys.exit(1)