|
10 | 10 | MutableMapping, |
11 | 11 | MutableSequence, |
12 | 12 | Optional, |
13 | | - Set, |
14 | 13 | Union, |
15 | 14 | cast, |
16 | 15 | ) |
17 | 16 |
|
18 | 17 | from cwl_utils import expression |
19 | | -from rdflib import Graph, URIRef |
20 | | -from rdflib.namespace import OWL, RDFS |
| 18 | +from cwl_utils.file_formats import check_format |
| 19 | +from rdflib import Graph |
21 | 20 | from ruamel.yaml.comments import CommentedMap |
22 | 21 | from schema_salad.avro.schema import Names, Schema, make_avsc_object |
23 | 22 | from schema_salad.exceptions import ValidationException |
@@ -76,64 +75,6 @@ def substitute(value, replace): # type: (str, str) -> str |
76 | 75 | return value + replace |
77 | 76 |
|
78 | 77 |
|
79 | | -def formatSubclassOf( |
80 | | - fmt: str, cls: str, ontology: Optional[Graph], visited: Set[str] |
81 | | -) -> bool: |
82 | | - """Determine if `fmt` is a subclass of `cls`.""" |
83 | | - if URIRef(fmt) == URIRef(cls): |
84 | | - return True |
85 | | - |
86 | | - if ontology is None: |
87 | | - return False |
88 | | - |
89 | | - if fmt in visited: |
90 | | - return False |
91 | | - |
92 | | - visited.add(fmt) |
93 | | - |
94 | | - uriRefFmt = URIRef(fmt) |
95 | | - |
96 | | - for _s, _p, o in ontology.triples((uriRefFmt, RDFS.subClassOf, None)): |
97 | | - # Find parent classes of `fmt` and search upward |
98 | | - if formatSubclassOf(o, cls, ontology, visited): |
99 | | - return True |
100 | | - |
101 | | - for _s, _p, o in ontology.triples((uriRefFmt, OWL.equivalentClass, None)): |
102 | | - # Find equivalent classes of `fmt` and search horizontally |
103 | | - if formatSubclassOf(o, cls, ontology, visited): |
104 | | - return True |
105 | | - |
106 | | - for s, _p, _o in ontology.triples((None, OWL.equivalentClass, uriRefFmt)): |
107 | | - # Find equivalent classes of `fmt` and search horizontally |
108 | | - if formatSubclassOf(s, cls, ontology, visited): |
109 | | - return True |
110 | | - |
111 | | - return False |
112 | | - |
113 | | - |
114 | | -def check_format( |
115 | | - actual_file: Union[CWLObjectType, List[CWLObjectType]], |
116 | | - input_formats: Union[List[str], str], |
117 | | - ontology: Optional[Graph], |
118 | | -) -> None: |
119 | | - """Confirm that the format present is valid for the allowed formats.""" |
120 | | - for afile in aslist(actual_file): |
121 | | - if not afile: |
122 | | - continue |
123 | | - if "format" not in afile: |
124 | | - raise ValidationException( |
125 | | - f"File has no 'format' defined: {json_dumps(afile, indent=4)}" |
126 | | - ) |
127 | | - for inpf in aslist(input_formats): |
128 | | - if afile["format"] == inpf or formatSubclassOf( |
129 | | - afile["format"], inpf, ontology, set() |
130 | | - ): |
131 | | - return |
132 | | - raise ValidationException( |
133 | | - f"File has an incompatible format: {json_dumps(afile, indent=4)}" |
134 | | - ) |
135 | | - |
136 | | - |
137 | 78 | class Builder(HasReqsHints): |
138 | 79 | def __init__( |
139 | 80 | self, |
|
0 commit comments