Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 0 additions & 63 deletions compliance_tool/aas_compliance_tool/compliance_check_aasx.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,69 +88,6 @@ def check_deserialization(file_path: str, state_manager: ComplianceToolStateMana
return identifiable_store, files, new_cp


def check_schema(file_path: str, state_manager: ComplianceToolStateManager) -> None:
"""
Checks a given file against the official json schema and reports any issues using the given
:class:`~basyx.aas.compliance_tool.state_manager.ComplianceToolStateManager`

Opens the file and checks if the data inside is stored in XML or JSON. Then calls the respective compliance tool
schema check
"""
logger = logging.getLogger('compliance_check')
logger.addHandler(state_manager)
logger.propagate = False
logger.setLevel(logging.INFO)

# create handler to get logger info
logger_deserialization = logging.getLogger(aasx.__name__)
logger_deserialization.addHandler(state_manager)
logger_deserialization.propagate = False
logger_deserialization.setLevel(logging.INFO)

state_manager.add_step('Open file')
try:
# open given file
reader = aasx.AASXReader(file_path)
state_manager.set_step_status_from_log()
except ValueError as error:
logger.error(error)
state_manager.set_step_status_from_log()
state_manager.add_step('Read file')
state_manager.set_step_status(Status.NOT_EXECUTED)
return

try:
# read given file (Find XML and JSON parts)
state_manager.add_step('Read file')
core_rels = reader.reader.get_related_parts_by_type()
try:
aasx_origin_part = core_rels[aasx.RELATIONSHIP_TYPE_AASX_ORIGIN][0]
except IndexError as e:
raise ValueError("Not a valid AASX file: aasx-origin Relationship is missing.") from e
state_manager.set_step_status(Status.SUCCESS)
for aas_part in reader.reader.get_related_parts_by_type(aasx_origin_part)[
aasx.RELATIONSHIP_TYPE_AAS_SPEC]:
content_type = reader.reader.get_content_type(aas_part)
extension = aas_part.split("/")[-1].split(".")[-1]
with reader.reader.open_part(aas_part) as p:
if content_type.split(";")[0] in (
"text/xml", "application/xml") or content_type == "" and extension == "xml":
logger.debug("Parsing AAS objects from XML stream in OPC part {} ...".format(aas_part))
compliance_check_xml._check_schema(p, state_manager)
elif content_type.split(";")[0] == "application/json" \
or content_type == "" and extension == "json":
logger.debug("Parsing AAS objects from JSON stream in OPC part {} ...".format(aas_part))
compliance_check_json._check_schema(io.TextIOWrapper(p, encoding='utf-8-sig'), state_manager)
else:
raise ValueError("Could not determine part format of AASX part {} (Content Type: {}, extension: {}"
.format(aas_part, content_type, extension))
except ValueError as error:
logger.error(error)
state_manager.set_step_status(Status.FAILED)
finally:
reader.close()


def check_aas_example(file_path: str, state_manager: ComplianceToolStateManager, **kwargs) -> None:
"""
Checks if a file contains all elements of the aas example and reports any issues using the given
Expand Down
78 changes: 0 additions & 78 deletions compliance_tool/aas_compliance_tool/compliance_check_json.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,84 +23,6 @@
from aas_compliance_tool.state_manager import ComplianceToolStateManager, Status


JSON_SCHEMA_FILE = os.path.join(os.path.dirname(__file__), 'schemas/aasJSONSchema.json')


def check_schema(file_path: str, state_manager: ComplianceToolStateManager) -> None:
"""
Checks a given file against the official json schema and reports any issues using the given
:class:`~basyx.aas.compliance_tool.state_manager.ComplianceToolStateManager`

Add the steps: `Open file`, `Read file and check if it is conform to the json syntax` and `Validate file against
official json schema`

:param file_path: Path to the file which should be checked
:param state_manager: :class:`~basyx.aas.compliance_tool.state_manager.ComplianceToolStateManager` to log the steps
"""
logger = logging.getLogger('compliance_check')
logger.addHandler(state_manager)
logger.propagate = False
logger.setLevel(logging.INFO)

state_manager.add_step('Open file')
try:
# open given file
file_to_be_checked = open(file_path, 'r', encoding='utf-8-sig')
except IOError as error:
state_manager.set_step_status(Status.FAILED)
logger.error(error)
state_manager.add_step('Read file and check if it is conform to the json syntax')
state_manager.set_step_status(Status.NOT_EXECUTED)
state_manager.add_step('Validate file against official json schema')
state_manager.set_step_status(Status.NOT_EXECUTED)
return
return _check_schema(file_to_be_checked, state_manager)


def _check_schema(file_to_be_checked: IO[str], state_manager: ComplianceToolStateManager):
logger = logging.getLogger('compliance_check')
logger.addHandler(state_manager)
logger.propagate = False
logger.setLevel(logging.INFO)

try:
with file_to_be_checked:
state_manager.set_step_status(Status.SUCCESS)
# read given file and check if it is conform to the json syntax
state_manager.add_step('Read file and check if it is conform to the json syntax')
json_to_be_checked = json.load(file_to_be_checked)
state_manager.set_step_status(Status.SUCCESS)
except json.decoder.JSONDecodeError as error:
state_manager.set_step_status(Status.FAILED)
logger.error(error)
state_manager.add_step('Validate file against official json schema')
state_manager.set_step_status(Status.NOT_EXECUTED)
return

# load json schema
with open(JSON_SCHEMA_FILE, 'r', encoding='utf-8-sig') as json_file:
aas_json_schema = json.load(json_file)
state_manager.add_step('Validate file against official json schema')

# validate given file against schema
try:
import jsonschema # type: ignore
except ImportError as error:
state_manager.set_step_status(Status.NOT_EXECUTED)
logger.error("Python package 'jsonschema' is required for validating the JSON file.", error)
return

try:
jsonschema.validate(instance=json_to_be_checked, schema=aas_json_schema)
except jsonschema.exceptions.ValidationError as error:
state_manager.set_step_status(Status.FAILED)
logger.error(error)
return

state_manager.set_step_status(Status.SUCCESS)
return


def check_deserialization(file_path: str, state_manager: ComplianceToolStateManager,
file_info: Optional[str] = None) -> model.DictIdentifiableStore:
"""
Expand Down
77 changes: 0 additions & 77 deletions compliance_tool/aas_compliance_tool/compliance_check_xml.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,83 +23,6 @@
from aas_compliance_tool.state_manager import ComplianceToolStateManager, Status


XML_SCHEMA_FILE = os.path.join(os.path.dirname(__file__), 'schemas/aasXMLSchema.xsd')


def check_schema(file_path: str, state_manager: ComplianceToolStateManager) -> None:
"""
Checks a given file against the official xml schema and reports any issues using the given
:class:`~basyx.aas.compliance_tool.state_manager.ComplianceToolStateManager`

Add the steps: `Open file`, `Read file`, `Check if it is conform to the xml syntax` and `Validate file against
official xml schema`

:param file_path: Path to the file which should be checked
:param state_manager: :class:`~basyx.aas.compliance_tool.state_manager.ComplianceToolStateManager` to log the steps
"""
logger = logging.getLogger('compliance_check')
logger.addHandler(state_manager)
logger.propagate = False
logger.setLevel(logging.INFO)

state_manager.add_step('Open file')
try:
# open given file
file_to_be_checked = open(file_path, 'rb')
state_manager.set_step_status(Status.SUCCESS)
except IOError as error:
state_manager.set_step_status(Status.FAILED)
logger.error(error)
state_manager.add_step('Read file and check if it is conform to the xml syntax')
state_manager.set_step_status(Status.NOT_EXECUTED)
state_manager.add_step('Validate file against official xml schema')
state_manager.set_step_status(Status.NOT_EXECUTED)
return
return _check_schema(file_to_be_checked, state_manager)


def _check_schema(file_to_be_checked, state_manager):
logger = logging.getLogger('compliance_check')
logger.addHandler(state_manager)
logger.propagate = False
logger.setLevel(logging.INFO)

state_manager.add_step('Read file and check if it is conform to the xml syntax')
try:
# read given file and check if it is conform to the xml syntax
parser = etree.XMLParser(remove_blank_text=True, remove_comments=True)
etree.parse(file_to_be_checked, parser)
state_manager.set_step_status(Status.SUCCESS)
except etree.XMLSyntaxError as error:
state_manager.set_step_status(Status.FAILED)
logger.error(error)
state_manager.add_step('Validate file against official xml schema')
state_manager.set_step_status(Status.NOT_EXECUTED)
file_to_be_checked.close()
return
except Exception:
file_to_be_checked.close()
raise

# load aas xml schema
aas_xml_schema = etree.XMLSchema(file=XML_SCHEMA_FILE)
parser = etree.XMLParser(schema=aas_xml_schema)

state_manager.add_step('Validate file against official xml schema')
# validate given file against schema
try:
file_to_be_checked.seek(0) # Reset reading file offset (cursor) to the beginning of the file
with file_to_be_checked:
etree.parse(file_to_be_checked, parser=parser)
except etree.ParseError as error:
state_manager.set_step_status(Status.FAILED)
logger.error(error)
return

state_manager.set_step_status(Status.SUCCESS)
return


def check_deserialization(file_path: str, state_manager: ComplianceToolStateManager,
file_info: Optional[str] = None) -> model.DictIdentifiableStore:
"""
Expand Down
Loading
Loading