import jsonschema
import json
#import fastjsonschema
class BehaverseJsonValidator():
"""
Loads Behaverse json file from local drive, aggregate json lines into a single
array, and validate the content and structure of the events.
"""
def __init__(self, json_schema_path: str = None):
if json_schema_path is not None:
with open(json_schema_path) as f:
self.schema = json.load(f)
@staticmethod
def __join_json_lines(src_path):
with open(src_path) as f:
return f"[{','.join(f.readlines())}]"
def convert(self, src_path, dest_path):
with open(dest_path, "w") as f:
f.write(self.__join_json_lines(src_path))
def validate_json(self, content):
"""
Validates a json object with respect to a predefined schema (json-schema-v7).
If json contains errornous content, a validation error will be raised.
"""
assert(self.schema is not None)
return jsonschema.validate(content, self.schema)
def validate_json_file(self, json_path: str):
"""
Loads and validates a json file. Respective validation error will be raised in case of invalid structure.
"""
with open(json_path) as f:
content = json.load(f)
return self.validate_json(content)
if __name__ == "__main__":
"""
Example code showing how to use this class.
"""
original_json_file = "data/samples/nback_demo.json"
valid_json_file = "data/samples/nback_demo_v.json"
schema_file = "behaverse/validators/nback.schema.json"
valid_json_file = "behaverse/validators/test.json"
schema_file = "behaverse/validators/test.schema.json"
BehaverseJsonValidator(json_schema_path=schema_file).validate_json_file(valid_json_file)
# load and concat json lines
#JsonAggregator().convert(original_json_file, valid_json_file)
# load and validate json lines
#BehaverseJsonValidator(json_schema_path=schema_file).validate_json_file(valid_json_file)
print("Good news everyone! Behaverse JSON file is valid!")