Integrate attr libraries
This commit is contained in:
Marcell Vazquez-Chanlatte
parent
fb2e79b807
commit
472fd45ce2
9 changed files with 200 additions and 265 deletions
305
mtl/ast.py
305
mtl/ast.py
|
|
@ -1,7 +1,9 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from collections import deque, namedtuple
|
||||
from functools import lru_cache
|
||||
from typing import Union, NamedTuple
|
||||
|
||||
import attr
|
||||
import funcy as fn
|
||||
from lenses import lens, bind
|
||||
|
||||
|
|
@ -24,131 +26,161 @@ def flatten_binary(phi, op, dropT, shortT):
|
|||
return op(tuple(fn.mapcat(f, phi.args)))
|
||||
|
||||
|
||||
class AST(object):
|
||||
__slots__ = ()
|
||||
|
||||
def __or__(self, other):
|
||||
return flatten_binary(Or((self, other)), Or, BOT, TOP)
|
||||
|
||||
def __and__(self, other):
|
||||
return flatten_binary(And((self, other)), And, TOP, BOT)
|
||||
|
||||
def __invert__(self):
|
||||
if isinstance(self, Neg):
|
||||
return self.arg
|
||||
return Neg(self)
|
||||
|
||||
def __rshift__(self, t):
|
||||
if self in (BOT, TOP):
|
||||
return self
|
||||
|
||||
phi = self
|
||||
for _ in range(t):
|
||||
phi = Next(phi)
|
||||
|
||||
return phi
|
||||
|
||||
def __call__(self, trace, time=0):
|
||||
return mtl.pointwise_sat(self)(trace, time)
|
||||
|
||||
@property
|
||||
def children(self):
|
||||
return tuple()
|
||||
|
||||
def walk(self):
|
||||
"""Walk of the AST."""
|
||||
pop = deque.pop
|
||||
children = deque([self])
|
||||
while len(children) > 0:
|
||||
node = pop(children)
|
||||
yield node
|
||||
children.extend(node.children)
|
||||
|
||||
@property
|
||||
def params(self):
|
||||
def get_params(leaf):
|
||||
if isinstance(leaf, ModalOp):
|
||||
if isinstance(leaf.interval[0], Param):
|
||||
yield leaf.interval[0]
|
||||
if isinstance(leaf.interval[1], Param):
|
||||
yield leaf.interval[1]
|
||||
|
||||
return set(fn.mapcat(get_params, self.walk()))
|
||||
|
||||
def set_params(self, val):
|
||||
phi = param_lens(self)
|
||||
return phi.modify(lambda x: float(val.get(x, val.get(str(x), x))))
|
||||
|
||||
@property
|
||||
def atomic_predicates(self):
|
||||
return set(AP_lens.collect()(self))
|
||||
|
||||
def inline_context(self, context):
|
||||
phi, phi2 = self, None
|
||||
|
||||
def update(ap):
|
||||
return context.get(ap, ap)
|
||||
|
||||
while phi2 != phi:
|
||||
phi2, phi = phi, AP_lens.modify(update)(phi)
|
||||
|
||||
return phi
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
def _or(exp1, exp2):
|
||||
return flatten_binary(Or((exp1, exp2)), Or, BOT, TOP)
|
||||
|
||||
|
||||
class _Top(AST):
|
||||
__slots__ = ()
|
||||
def _and(exp1, exp2):
|
||||
return flatten_binary(And((exp1, exp2)), And, TOP, BOT)
|
||||
|
||||
|
||||
def _neg(exp):
|
||||
if isinstance(exp, _Bot):
|
||||
return _Top()
|
||||
elif isinstance(exp, _Top):
|
||||
return _Bot()
|
||||
elif isinstance(exp, Neg):
|
||||
return exp.arg
|
||||
return Neg(exp)
|
||||
|
||||
|
||||
def _eval(exp, trace, time=0):
|
||||
return mtl.pointwise_sat(exp)(trace, time)
|
||||
|
||||
|
||||
def _timeshift(exp, t):
|
||||
if exp in (BOT, TOP):
|
||||
return exp
|
||||
|
||||
for _ in range(t):
|
||||
exp = Next(exp)
|
||||
return exp
|
||||
|
||||
|
||||
def _walk(exp):
|
||||
"""Walk of the AST."""
|
||||
pop = deque.pop
|
||||
children = deque([exp])
|
||||
while len(children) > 0:
|
||||
node = pop(children)
|
||||
yield node
|
||||
children.extend(node.children)
|
||||
|
||||
|
||||
def _params(exp):
|
||||
def get_params(leaf):
|
||||
if isinstance(leaf, ModalOp):
|
||||
if isinstance(leaf.interval[0], Param):
|
||||
yield leaf.interval[0]
|
||||
if isinstance(leaf.interval[1], Param):
|
||||
yield leaf.interval[1]
|
||||
|
||||
return set(fn.mapcat(get_params, exp.walk()))
|
||||
|
||||
|
||||
def _set_symbols(node, val):
|
||||
children = tuple(_set_symbols(c, val) for c in node.children)
|
||||
|
||||
if hasattr(node, 'interval'):
|
||||
return node.evolve(
|
||||
arg=children[0],
|
||||
interval=_update_itvl(node.interval, val),
|
||||
)
|
||||
elif isinstance(node, AtomicPred):
|
||||
return val.get(node.id, node)
|
||||
elif hasattr(node, 'args'):
|
||||
return node.evolve(args=children)
|
||||
elif hasattr(node, 'arg'):
|
||||
return node.evolve(arg=children[0])
|
||||
return node
|
||||
|
||||
|
||||
def _inline_context(exp, context):
|
||||
phi, phi2 = exp, None
|
||||
while phi2 != phi:
|
||||
phi2, phi = phi, _set_symbols(phi, context)
|
||||
|
||||
return phi
|
||||
|
||||
|
||||
def _atomic_predicates(exp):
|
||||
return set(bind(exp).Recur(AtomicPred).collect())
|
||||
|
||||
|
||||
class Param(NamedTuple):
|
||||
name: str
|
||||
|
||||
def __repr__(self):
|
||||
return self.name
|
||||
|
||||
|
||||
def ast_class(cls):
|
||||
cls.__or__ = _or
|
||||
cls.__and__ = _and
|
||||
cls.__invert__ = _neg
|
||||
cls.__call__ = _eval
|
||||
cls.__rshift__ = _timeshift
|
||||
cls.__getitem__ = _inline_context
|
||||
cls.walk = _walk
|
||||
cls.params = property(_params)
|
||||
cls.atomic_predicates = property(_atomic_predicates)
|
||||
cls.evolve = attr.evolve
|
||||
|
||||
if not hasattr(cls, "children"):
|
||||
cls.children = property(lambda _: ())
|
||||
|
||||
|
||||
|
||||
return attr.s(frozen=True, auto_attribs=True, repr=False, slots=True)(cls)
|
||||
|
||||
|
||||
def _update_itvl(itvl, lookup):
|
||||
def _update_param(p):
|
||||
if not isinstance(p, Param) or p.name not in lookup:
|
||||
return p
|
||||
|
||||
val = lookup[p.name]
|
||||
return val if isinstance(lookup, Param) else float(val)
|
||||
|
||||
return Interval(*map(_update_param, itvl))
|
||||
|
||||
|
||||
@ast_class
|
||||
class _Top:
|
||||
def __repr__(self):
|
||||
return "TRUE"
|
||||
|
||||
def __invert__(self):
|
||||
return BOT
|
||||
|
||||
|
||||
class _Bot(AST):
|
||||
__slots__ = ()
|
||||
|
||||
@ast_class
|
||||
class _Bot:
|
||||
def __repr__(self):
|
||||
return "FALSE"
|
||||
|
||||
def __invert__(self):
|
||||
return TOP
|
||||
|
||||
|
||||
TOP = _Top()
|
||||
BOT = _Bot()
|
||||
|
||||
|
||||
class AtomicPred(namedtuple("AP", ["id"]), AST):
|
||||
__slots__ = ()
|
||||
@ast_class
|
||||
class AtomicPred:
|
||||
id: str
|
||||
|
||||
def __repr__(self):
|
||||
return f"{self.id}"
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
@property
|
||||
def children(self):
|
||||
return tuple()
|
||||
|
||||
|
||||
class Interval(namedtuple('I', ['lower', 'upper'])):
|
||||
__slots__ = ()
|
||||
class Interval(NamedTuple):
|
||||
lower: Union[float, Param]
|
||||
upper: Union[float, Param]
|
||||
|
||||
def __repr__(self):
|
||||
return f"[{self.lower},{self.upper}]"
|
||||
|
||||
|
||||
class NaryOpMTL(namedtuple('NaryOp', ['args']), AST):
|
||||
__slots__ = ()
|
||||
|
||||
@ast_class
|
||||
class NaryOpMTL:
|
||||
OP = "?"
|
||||
args: "Node" # TODO: when 3.7 is more common replace with type union.
|
||||
|
||||
def __repr__(self):
|
||||
return "(" + f" {self.OP} ".join(f"{x}" for x in self.args) + ")"
|
||||
|
|
@ -159,28 +191,18 @@ class NaryOpMTL(namedtuple('NaryOp', ['args']), AST):
|
|||
|
||||
|
||||
class Or(NaryOpMTL):
|
||||
__slots__ = ()
|
||||
|
||||
OP = "|"
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
class And(NaryOpMTL):
|
||||
__slots__ = ()
|
||||
|
||||
OP = "&"
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
class ModalOp(namedtuple('ModalOp', ['interval', 'arg']), AST):
|
||||
__slots__ = ()
|
||||
@ast_class
|
||||
class ModalOp:
|
||||
OP = '?'
|
||||
interval: Interval
|
||||
arg: "Node"
|
||||
|
||||
def __repr__(self):
|
||||
if self.interval.lower == 0 and self.interval.upper == float('inf'):
|
||||
|
|
@ -193,25 +215,17 @@ class ModalOp(namedtuple('ModalOp', ['interval', 'arg']), AST):
|
|||
|
||||
|
||||
class F(ModalOp):
|
||||
__slots__ = ()
|
||||
OP = "< >"
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
class G(ModalOp):
|
||||
__slots__ = ()
|
||||
OP = "[ ]"
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
class Until(namedtuple('ModalOp', ['arg1', 'arg2']), AST):
|
||||
__slots__ = ()
|
||||
@ast_class
|
||||
class Until:
|
||||
arg1: "Node"
|
||||
arg2: "Node"
|
||||
|
||||
def __repr__(self):
|
||||
return f"({self.arg1} U {self.arg2})"
|
||||
|
|
@ -220,13 +234,10 @@ class Until(namedtuple('ModalOp', ['arg1', 'arg2']), AST):
|
|||
def children(self):
|
||||
return (self.arg1, self.arg2)
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
class Neg(namedtuple('Neg', ['arg']), AST):
|
||||
__slots__ = ()
|
||||
@ast_class
|
||||
class Neg:
|
||||
arg: "Node"
|
||||
|
||||
def __repr__(self):
|
||||
return f"~{self.arg}"
|
||||
|
|
@ -235,13 +246,10 @@ class Neg(namedtuple('Neg', ['arg']), AST):
|
|||
def children(self):
|
||||
return (self.arg,)
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
class Next(namedtuple('Next', ['arg']), AST):
|
||||
__slots__ = ()
|
||||
@ast_class
|
||||
class Next:
|
||||
arg: "Node"
|
||||
|
||||
def __repr__(self):
|
||||
return f"@{self.arg}"
|
||||
|
|
@ -250,30 +258,7 @@ class Next(namedtuple('Next', ['arg']), AST):
|
|||
def children(self):
|
||||
return (self.arg,)
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
class Param(namedtuple('Param', ['name']), AST):
|
||||
__slots__ = ()
|
||||
|
||||
def __repr__(self):
|
||||
return self.name
|
||||
|
||||
def __hash__(self):
|
||||
# TODO: compute hash based on contents
|
||||
return hash(repr(self))
|
||||
|
||||
|
||||
@lru_cache()
|
||||
def param_lens(phi, *, getter=False):
|
||||
return bind(phi).Recur(Param)
|
||||
|
||||
|
||||
def type_pred(*args):
|
||||
ast_types = set(args)
|
||||
return lambda x: type(x) in ast_types
|
||||
|
||||
|
||||
AP_lens = lens.Recur(AtomicPred)
|
||||
|
|
|
|||
|
|
@ -16,17 +16,14 @@ FALSE_TRACE = const_trace(False)
|
|||
|
||||
|
||||
def negate_trace(x):
|
||||
out = x.operation(TRUE_TRACE, op.xor)
|
||||
out.domain = x.domain
|
||||
return out
|
||||
return x.operation(TRUE_TRACE, op.xor)
|
||||
|
||||
|
||||
def pointwise_sat(phi, dt=0.1):
|
||||
ap_names = [z.id for z in phi.atomic_predicates]
|
||||
|
||||
def _eval_mtl(x, t=0):
|
||||
evaluated = fn.project(x, ap_names)
|
||||
return bool(eval_mtl(phi, dt)(evaluated)[t])
|
||||
return bool(eval_mtl(phi, dt)(x)[t])
|
||||
|
||||
return _eval_mtl
|
||||
|
||||
|
|
@ -37,9 +34,7 @@ def eval_mtl(phi, dt):
|
|||
|
||||
|
||||
def or_traces(xs):
|
||||
out = orf(*xs)
|
||||
out.domain = xs[0].domain
|
||||
return out
|
||||
return orf(*xs)
|
||||
|
||||
|
||||
@eval_mtl.register(mtl.Or)
|
||||
|
|
@ -55,9 +50,7 @@ def eval_mtl_or(phi, dt):
|
|||
|
||||
|
||||
def and_traces(xs):
|
||||
out = andf(*xs)
|
||||
out.domain = xs[0].domain
|
||||
return out
|
||||
return andf(*xs)
|
||||
|
||||
|
||||
@eval_mtl.register(mtl.And)
|
||||
|
|
@ -73,6 +66,10 @@ def eval_mtl_and(phi, dt):
|
|||
|
||||
|
||||
def apply_until(y):
|
||||
if len(y) == 1:
|
||||
left, right = y.first_value()
|
||||
yield (0, min(left, right))
|
||||
return
|
||||
periods = list(y.iterperiods())
|
||||
phi2_next = False
|
||||
for t, _, (phi1, phi2) in periods[::-1]:
|
||||
|
|
@ -87,7 +84,7 @@ def eval_mtl_until(phi, dt):
|
|||
def _eval(x):
|
||||
y1, y2 = f1(x), f2(x)
|
||||
y = y1.operation(y2, lambda a, b: (a, b))
|
||||
out = traces.TimeSeries(apply_until(y), domain=y1.domain)
|
||||
out = traces.TimeSeries(apply_until(y))
|
||||
out.compact()
|
||||
|
||||
return out
|
||||
|
|
@ -111,7 +108,7 @@ def eval_mtl_g(phi, dt):
|
|||
def process_intervals(x):
|
||||
# Need to add last interval
|
||||
intervals = fn.chain(x.iterintervals(), [(
|
||||
x.last(),
|
||||
x.first_item(),
|
||||
(float('inf'), None),
|
||||
)])
|
||||
for (start, val), (end, val2) in intervals:
|
||||
|
|
@ -121,10 +118,10 @@ def eval_mtl_g(phi, dt):
|
|||
|
||||
if b == float('inf'):
|
||||
def _eval(x):
|
||||
y = f(x)
|
||||
val = len(y.slice(a, b)) == 1 and y[a]
|
||||
return traces.TimeSeries(
|
||||
[(y.domain.start(), val)], domain=y.domain)
|
||||
y = f(x).slice(a, b)
|
||||
y.compact()
|
||||
val = len(y) == 1 and y[a]
|
||||
return const_trace(val)
|
||||
else:
|
||||
def _eval(x):
|
||||
y = f(x)
|
||||
|
|
@ -132,7 +129,8 @@ def eval_mtl_g(phi, dt):
|
|||
return y
|
||||
|
||||
out = traces.TimeSeries(process_intervals(y)).slice(
|
||||
y.domain.start(), y.domain.end())
|
||||
y.first_key(), float('inf')
|
||||
)
|
||||
out.compact()
|
||||
return out
|
||||
|
||||
|
|
@ -158,7 +156,7 @@ def eval_mtl_next(phi, dt):
|
|||
def _eval(x):
|
||||
y = f(x)
|
||||
out = traces.TimeSeries(((t - dt, v) for t, v in y))
|
||||
out = out.slice(y.domain.start(), y.domain.end())
|
||||
out = out.slice(0, float('inf'))
|
||||
out.compact()
|
||||
|
||||
return out
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ oo = float('inf')
|
|||
|
||||
|
||||
def get_times(x, tau, lo, hi):
|
||||
end = min(v.domain.end() for v in x.values())
|
||||
end = min(v.last_key() for v in x.values())
|
||||
|
||||
lo, hi = map(float, (lo, hi))
|
||||
hi = hi + tau if hi + tau <= end else end
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ class MTLVisitor(NodeVisitor):
|
|||
self.default_interval = ast.Interval(0.0, H)
|
||||
|
||||
def binop_inner(self, _, children):
|
||||
if isinstance(children[0], ast.AST):
|
||||
if not isinstance(children[0], list):
|
||||
return children
|
||||
|
||||
((left, _, _, _, right),) = children
|
||||
|
|
|
|||
|
|
@ -23,23 +23,17 @@ TODO: Automatically generate input time series.
|
|||
"""
|
||||
|
||||
x = {
|
||||
"x":
|
||||
traces.TimeSeries([(0, 1), (0.1, 1), (0.2, 4)], domain=(0, 10)),
|
||||
"y":
|
||||
traces.TimeSeries([(0, 2), (0.1, 4), (0.2, 2)], domain=(0, 10)),
|
||||
"ap1":
|
||||
traces.TimeSeries([(0, True), (0.1, True), (0.2, False)], domain=(0, 10)),
|
||||
"ap1": traces.TimeSeries([(0, True), (0.1, True), (0.2, False)]),
|
||||
"ap2":
|
||||
traces.TimeSeries([(0, False), (0.2, True), (0.5, False)], domain=(0, 10)),
|
||||
traces.TimeSeries([(0, False), (0.2, True), (0.5, False)]),
|
||||
"ap3":
|
||||
traces.TimeSeries([(0, True), (0.1, True), (0.3, False)], domain=(0, 10)),
|
||||
traces.TimeSeries([(0, True), (0.1, True), (0.3, False)]),
|
||||
"ap4":
|
||||
traces.TimeSeries(
|
||||
[(0, False), (0.1, False), (0.3, False)], domain=(0, 10)),
|
||||
traces.TimeSeries([(0, False), (0.1, False), (0.3, False)]),
|
||||
"ap5":
|
||||
traces.TimeSeries([(0, False), (0.1, False), (0.3, True)], domain=(0, 10)),
|
||||
traces.TimeSeries([(0, False), (0.1, False), (0.3, True)]),
|
||||
"ap6":
|
||||
traces.TimeSeries([(0, True)], domain=(0, 10)),
|
||||
traces.TimeSeries([(0, True), (float('inf'), True)]),
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -118,7 +112,7 @@ def test_fastboolean_smoketest():
|
|||
assert not mtl_eval(x, 0)
|
||||
|
||||
with raises(NotImplementedError):
|
||||
mtl.fastboolean_eval.pointwise_sat(mtl.ast.AST())
|
||||
mtl.fastboolean_eval.pointwise_sat(None)
|
||||
|
||||
|
||||
def test_callable_interface():
|
||||
|
|
|
|||
|
|
@ -10,11 +10,7 @@ def test_params1(a, b, c):
|
|||
phi = mtl.parse('G[a, b] x')
|
||||
assert {x.name for x in phi.params} == {'a', 'b'}
|
||||
|
||||
phi2 = phi.set_params({'a': a, 'b': b})
|
||||
phi2 = phi[{'a': a, 'b': b}]
|
||||
assert phi2.params == set()
|
||||
assert phi2 == mtl.parse(f'G[{a}, {b}](x)')
|
||||
|
||||
|
||||
@given(MetricTemporalLogicStrategy)
|
||||
def test_hash_stable(phi):
|
||||
assert hash(phi) == hash(mtl.parse(str(phi)))
|
||||
|
||||
|
|
|
|||
|
|
@ -1,19 +1,26 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from hypothesis import event, given
|
||||
from traces import TimeSeries
|
||||
|
||||
import mtl
|
||||
from mtl.hypothesis import MetricTemporalLogicStrategy
|
||||
|
||||
|
||||
@given(MetricTemporalLogicStrategy)
|
||||
def test_invertable_repr(phi):
|
||||
event(str(phi))
|
||||
assert str(phi) == str(mtl.parse(str(phi)))
|
||||
TS = {
|
||||
"ap1": TimeSeries([(0, True), (0.1, True), (0.2, False)]),
|
||||
"ap2": TimeSeries([(0, False), (0.2, True), (0.5, False)]),
|
||||
"ap3": TimeSeries([(0, True), (0.1, True), (0.3, False)]),
|
||||
"ap4": TimeSeries([(0, False), (0.1, False), (0.3, False)]),
|
||||
"ap5": TimeSeries([(0, False), (0.1, False), (0.3, True)]),
|
||||
}
|
||||
|
||||
|
||||
@given(MetricTemporalLogicStrategy)
|
||||
def test_hash_inheritance(phi):
|
||||
assert hash(repr(phi)) == hash(phi)
|
||||
def test_stablizing_repr(phi):
|
||||
for _ in range(10):
|
||||
phi, phi2 = mtl.parse(str(phi)), phi
|
||||
|
||||
assert phi == phi2
|
||||
|
||||
|
||||
def test_sugar_smoke():
|
||||
|
|
|
|||
|
|
@ -5,43 +5,25 @@ from hypothesis import given
|
|||
from pytest import raises
|
||||
|
||||
CONTEXT = {
|
||||
mtl.parse('ap1'): mtl.parse('x'),
|
||||
mtl.parse('ap2'): mtl.parse('(y U z)'),
|
||||
mtl.parse('ap3'): mtl.parse('x'),
|
||||
mtl.parse('ap4'): mtl.parse('(x -> y -> z)'),
|
||||
mtl.parse('ap5'): mtl.parse('(ap1 <-> y <-> z)'),
|
||||
'ap1': mtl.parse('x'),
|
||||
'ap2': mtl.parse('(y U z)'),
|
||||
'ap3': mtl.parse('x'),
|
||||
'ap4': mtl.parse('(x -> y -> z)'),
|
||||
'ap5': mtl.parse('(ap1 <-> y <-> z)'),
|
||||
}
|
||||
APS = set(CONTEXT.keys())
|
||||
|
||||
|
||||
@given(MetricTemporalLogicStrategy)
|
||||
def test_f_neg_or_canonical_form(phi):
|
||||
phi2 = mtl.utils.f_neg_or_canonical_form(phi)
|
||||
phi3 = mtl.utils.f_neg_or_canonical_form(phi2)
|
||||
assert phi2 == phi3
|
||||
assert not any(
|
||||
isinstance(x, (mtl.ast.G, mtl.ast.And)) for x in phi2.walk())
|
||||
|
||||
|
||||
def test_f_neg_or_canonical_form_not_implemented():
|
||||
with raises(NotImplementedError):
|
||||
mtl.utils.f_neg_or_canonical_form(mtl.ast.AST())
|
||||
|
||||
|
||||
def test_inline_context_rigid():
|
||||
phi = mtl.parse('G ap1')
|
||||
phi2 = phi.inline_context(CONTEXT)
|
||||
assert phi2 == mtl.parse('G x')
|
||||
assert phi[CONTEXT] == mtl.parse('G x')
|
||||
|
||||
phi = mtl.parse('G ap5')
|
||||
phi2 = phi.inline_context(CONTEXT)
|
||||
assert phi2 == mtl.parse('G(x <-> y <-> z)')
|
||||
assert phi[CONTEXT] == mtl.parse('G(x <-> y <-> z)')
|
||||
|
||||
|
||||
@given(MetricTemporalLogicStrategy)
|
||||
def test_inline_context(phi):
|
||||
phi2 = phi.inline_context(CONTEXT)
|
||||
assert not (APS & phi2.atomic_predicates)
|
||||
assert not (APS & phi[CONTEXT].atomic_predicates)
|
||||
|
||||
|
||||
@given(MetricTemporalLogicStrategy, MetricTemporalLogicStrategy)
|
||||
|
|
|
|||
39
mtl/utils.py
39
mtl/utils.py
|
|
@ -13,35 +13,8 @@ from mtl.ast import (And, F, G, Interval, Neg, Or, Next, Until,
|
|||
oo = float('inf')
|
||||
|
||||
|
||||
def f_neg_or_canonical_form(phi):
|
||||
if isinstance(phi, (AtomicPred, _Top, _Bot)):
|
||||
return phi
|
||||
|
||||
children = [f_neg_or_canonical_form(s) for s in phi.children]
|
||||
if isinstance(phi, (And, G)):
|
||||
children = [Neg(s) for s in children]
|
||||
children = tuple(sorted(children, key=str))
|
||||
|
||||
if isinstance(phi, Or):
|
||||
return Or(children)
|
||||
elif isinstance(phi, And):
|
||||
return Neg(Or(children))
|
||||
elif isinstance(phi, Neg):
|
||||
return Neg(*children)
|
||||
elif isinstance(phi, Next):
|
||||
return Next(*children)
|
||||
elif isinstance(phi, Until):
|
||||
return Until(*children)
|
||||
elif isinstance(phi, F):
|
||||
return F(phi.interval, *children)
|
||||
elif isinstance(phi, G):
|
||||
return Neg(F(phi.interval, *children))
|
||||
else:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def const_trace(x, start=0):
|
||||
return traces.TimeSeries([(start, x)], domain=traces.Domain(start, oo))
|
||||
def const_trace(x):
|
||||
return traces.TimeSeries([(0, x), (oo, x)])
|
||||
|
||||
|
||||
def require_discretizable(func):
|
||||
|
|
@ -85,9 +58,9 @@ def _discretize(phi, dt, horizon):
|
|||
if not isinstance(phi, (F, G, Until)):
|
||||
children = tuple(_discretize(arg, dt, horizon) for arg in phi.children)
|
||||
if isinstance(phi, (And, Or)):
|
||||
return bind(phi).args.set(children)
|
||||
return phi.evolve(args=children)
|
||||
elif isinstance(phi, (Neg, Next)):
|
||||
return bind(phi).arg.set(children[0])
|
||||
return phi.evolve(arg=children[0])
|
||||
|
||||
raise NotImplementedError
|
||||
|
||||
|
|
@ -120,9 +93,9 @@ def _distribute_next(phi, i=0):
|
|||
children = tuple(_distribute_next(c, i) for c in phi.children)
|
||||
|
||||
if isinstance(phi, (And, Or)):
|
||||
return bind(phi).args.set(children)
|
||||
return phi.evolve(args=children)
|
||||
elif isinstance(phi, (Neg, Next)):
|
||||
return bind(phi).arg.set(children[0])
|
||||
return phi.evolve(arg=children[0])
|
||||
|
||||
|
||||
def is_discretizable(phi, dt):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue