Skip to content
Snippets Groups Projects
Unverified Commit 4939c5f9 authored by Andrea Franchini's avatar Andrea Franchini
Browse files

add CLI support

parent edeb4259
No related branches found
No related tags found
No related merge requests found
#!/usr/bin/env python3
import argparse
from mc_openapi.app_config import app
from mc_openapi.doml_mc import DOMLVersion
from mc_openapi.doml_mc.mc import ModelChecker
from mc_openapi.doml_mc.mc_result import MCResult
parser = argparse.ArgumentParser()
parser.add_argument("-d", "--doml", dest="doml", help="the DOMLX file to check")
parser.add_argument("-V", "--doml-version", dest="doml_version", default="V2_0", help="(optional) the version used by the DOMLX file")
parser.add_argument("-r", "--requirements", dest="requirements", help="the user-specified requirements file to check")
parser.add_argument("-c", "--check-consistency", dest="consistency", action='store_true', help="check additional built-in consistency requirements")
parser.add_argument("-t", "--threads", dest="threads", type=int, help="number of threads used by the model checker")
args = parser.parse_args()
if not args.doml:
# Start the webserver
app.run(port=8080)
else:
# Run only it via command line
doml_path = args.doml
reqs_path = args.requirements
try:
doml_ver = DOMLVersion[args.doml_version]
except:
print(f"Unknown DOML version '{args.doml_version}'")
versions = [ ver.name for ver in list(DOMLVersion)]
print(f"Available DOML versions = {versions}")
exit(1)
with open(doml_path, "rb") as xmif:
# Read DOML file from path
doml_xmi = xmif.read()
# Config the model checker
dmc = ModelChecker(doml_xmi, doml_ver)
user_reqs = None
if reqs_path:
with open(reqs_path, "r") as reqsf:
# Read the user requirements written in DSL
user_reqs = reqsf.read()
# Check satisfiability
results = dmc.check_requirements(
threads=args.threads,
user_requirements=user_reqs,
consistency_checks=args.consistency
)
res, msg = results.summarize()
if res == MCResult.sat:
print("sat")
else:
print(res.name)
print("\033[91m {}\033[00m".format(msg))
\ No newline at end of file
......@@ -2,6 +2,8 @@ from typing import Optional
from joblib import parallel_backend, Parallel, delayed
from multiprocessing import TimeoutError
from mc_openapi.dsl_parser.parser import Parser
from .intermediate_model.metamodel import (
DOMLVersion,
MetaModels,
......@@ -9,7 +11,7 @@ from .intermediate_model.metamodel import (
)
from .xmi_parser.doml_model import parse_doml_model
from .mc_result import MCResult, MCResults
from .imc import RequirementStore, IntermediateModelChecker
from .imc import Requirement, RequirementStore, IntermediateModelChecker
from .common_reqs import CommonRequirements
from .consistency_reqs import (
get_attribute_type_reqs,
......@@ -26,7 +28,13 @@ class ModelChecker:
self.metamodel = MetaModels[self.doml_version]
self.inv_assoc = InverseAssociations[self.doml_version]
def check_common_requirements(self, threads: int = 1, consistency_checks: bool = False, timeout: Optional[int] = None) -> MCResults:
def check_requirements(
self,
threads: int = 1,
user_requirements: Optional[str] = None,
consistency_checks: bool = False,
timeout: Optional[int] = None
) -> MCResults:
assert self.metamodel and self.inv_assoc
req_store = CommonRequirements[self.doml_version]
if consistency_checks:
......@@ -42,11 +50,6 @@ class ModelChecker:
rs = RequirementStore(req_store.get_all_requirements()[rfrom:rto])
return imc.check_requirements(rs)
if threads <= 1:
imc = IntermediateModelChecker(self.metamodel, self.inv_assoc, self.intermediate_model)
reqs = imc.check_requirements(req_store, timeout=(0 if timeout is None else timeout))
return reqs
else:
def split_reqs(n_reqs: int, n_split: int):
slice_size = n_reqs // n_split
rto = 0
......@@ -58,9 +61,18 @@ class ModelChecker:
try:
with parallel_backend('loky', n_jobs=threads):
results = Parallel(timeout=timeout)(delayed(worker)(rfrom, rto) for rfrom, rto in split_reqs(len(req_store), threads))
ret = MCResults([])
for res in results:
ret.add_results(res)
if user_requirements:
imc = IntermediateModelChecker(self.metamodel, self.inv_assoc, self.intermediate_model)
parser = Parser(imc.smt_encoding, imc.smt_sorts)
user_reqs_store = parser.parse(user_requirements)
user_results = imc.check_requirements(user_reqs_store)
ret.add_results(user_results)
return ret
except TimeoutError:
return MCResults([(MCResult.dontknow, "")])
......@@ -13,7 +13,7 @@ def post(body, requirement=None):
doml_xmi = body
try:
dmc = ModelChecker(doml_xmi, DOMLVersion.V2_0)
results = dmc.check_common_requirements(threads=2, consistency_checks=False, timeout=50)
results = dmc.check_requirements(threads=2, user_requirements=None, consistency_checks=False, timeout=50)
res, msg = results.summarize()
if res == MCResult.sat:
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment