Source code for momba.model.automata

# -*- coding:utf-8 -*-
#
# Copyright (C) 2019-2021, Saarland University
# Copyright (C) 2019-2021, Maximilian Köhl <koehl@cs.uni-saarland.de>

from __future__ import annotations

import dataclasses as d
import typing as t

import collections

from mxu.maps import FrozenMap

from . import actions, errors, expressions, observations, types

if t.TYPE_CHECKING:
    from . import context


Annotation = t.Mapping[str, t.Union[int, str, float]]


[docs]@d.dataclass(frozen=True, eq=False) class Instance: """ Represents an automaton instance. Attributes ---------- automaton: The instaniated automaton. arguments: The arguments passed to the parameters of the automaton. input_enabled: The set of action types for which the instance is input enabled. comment: An optional comment describing the instance. """ automaton: Automaton arguments: t.Tuple[expressions.Expression, ...] = () input_enable: t.FrozenSet[actions.ActionType] = frozenset() comment: t.Optional[str] = None def __post_init__(self) -> None: if len(self.arguments) != len(self.automaton.parameters): raise errors.ModelingError("invalid number of arguments for automaton")
ProgressInvariant = t.Optional[expressions.Expression] TransientValues = t.AbstractSet["Assignment"]
[docs]@d.dataclass(frozen=True, eq=False) class Location: """ Represents a location of an automaton. Attributes: name: The optional name of the location. progress_invariant: An optional expression `progress_invariant` specifies the invariant subject to which the time may progress in the location. As per the JANI specification, the invariant is only allowed for models using real-valued clocks (see :class:`~momba.model.ModelType`). transient_values: A set of assignments for transient variables. These assignments define the values of the transient variables when in the location. """ name: t.Optional[str] = None progress_invariant: ProgressInvariant = None transient_values: TransientValues = frozenset()
[docs] def validate(self, automaton: Automaton) -> None: """ Validates the location for the given automaton. Raises :class:`~errors.ModelingError` if the location is invalid to add to the automaton according to the JANI specification. For instance, if the location has a `progress_invariant` but the model type does not support clocks. """ scope = automaton.scope if self.progress_invariant is not None: if not scope.ctx.model_type.uses_clocks: raise errors.ModelingError( f"location invariant is not allowed for model type {scope.ctx.model_type}" ) if scope.get_type(self.progress_invariant) != types.BOOL: raise errors.InvalidTypeError( f"type of invariant in location {self} is not `types.BOOL`" ) if self.transient_values: if not are_compatible(self.transient_values): raise errors.IncompatibleAssignmentsError( f"incompatible assignments for transient values in location {self}" ) for assignment in self.transient_values: if assignment.index != 0: raise errors.ModelingError( f"index of assignment {assignment} to transient variable is non-zero" ) assignment.validate(scope)
[docs]@d.dataclass(frozen=True) class Assignment: """ Represents an assignment. Attributes ---------- target: The target of the assignment. value: The value to assign. index: The index of the assignment. """ target: expressions.Expression value: expressions.Expression index: int = 0
[docs] def validate(self, scope: context.Scope) -> None: """ Validates the assignment in the given scope. Raises :class:`~errors.ModelingError` if the target is not assignable from the value type. """ target_type = self.target.infer_target_type(scope) value_type = scope.get_type(self.value) if not target_type.is_assignable_from(value_type): raise errors.InvalidTypeError( f"cannot assign {value_type} to {target_type}" )
[docs]def are_compatible(assignments: t.Iterable[Assignment]) -> bool: """ Checks whether the given assignments are compatible according to the JANI specification. """ groups: t.DefaultDict[int, t.Set[expressions.Expression]] = collections.defaultdict( set ) for assignment in assignments: target = assignment.target if target in groups[assignment.index]: return False groups[assignment.index].add(target) return True
[docs]@d.dataclass(frozen=True) class Destination: """ Represents a destination of an edge. Attributes ---------- location: The target location of the destination. probability: An optional expression for the probability of the destination. assignments: A set of assignments to be executed when going to the destination. """ location: Location probability: t.Optional[expressions.Expression] = None assignments: t.Tuple[Assignment, ...] = () def _validate(self, automaton: Automaton, scope: context.Scope) -> None: self.location.validate(automaton) if self.location not in automaton.locations: raise errors.ModelingError( f"source location of edge {self} is not a location of the automaton {automaton}" ) if not are_compatible(self.assignments): raise errors.IncompatibleAssignmentsError( f"assignments on edge {self} are not compatible" ) if self.probability is not None: if not scope.get_type(self.probability).is_numeric: raise errors.InvalidTypeError("probability value must be numeric") for assignment in self.assignments: assignment.validate(scope)
[docs]@d.dataclass(frozen=True) class Edge: """ Represents an edge of an automaton. Attributes ---------- location: The source location of the edge. destinations: The destinations of the edge. action_pattern: The optional action pattern of the edge. guard: The optional guard of the edge. rate: The optional rate of the edge. annotation: An optional annotation of the edge. """ location: Location destinations: t.Tuple[Destination, ...] action_pattern: t.Optional[actions.ActionPattern] = None guard: t.Optional[expressions.Expression] = None rate: t.Optional[expressions.Expression] = None annotation: t.Optional[Annotation] = None observation: t.FrozenSet[observations.Observation] = frozenset()
[docs] def create_edge_scope(self, parent: context.Scope) -> context.Scope: """ Creates an *edge scope* with the given parent scope. .. warning:: Used for *value passing* an experimental Momba feature. Value passing is not part of the official JANI specification. """ scope = parent.create_child_scope() if self.action_pattern is not None: self.action_pattern.declare_in(scope) return scope
[docs] def validate(self, automaton: Automaton) -> None: """ Validates the edge for the given automaton. Raises :class:`~errors.ModelingError` if the edge is invalid to add to the automaton according to the JANI specification. For instance, if source location is not a location of the automaton. """ scope = automaton.scope self.location.validate(automaton) if self.location not in automaton.locations: raise errors.ModelingError( f"source location of edge {self} is not a location of the automaton {automaton}" ) if self.rate is not None and not scope.get_type(self.rate).is_numeric: raise errors.InvalidTypeError(f"type of rate on edge {self} is not numeric") edge_scope = self.create_edge_scope(scope) if self.guard is not None and edge_scope.get_type(self.guard) != types.BOOL: raise errors.InvalidTypeError( f"type of guard on edge {self} is not `types.BOOL`" ) for destination in self.destinations: destination._validate(automaton, edge_scope)
[docs]class Automaton: """ Represents an automaton. Attributes ---------- ctx: The :class:`Context` associated with the automaton. scope: The local :class:`Scope` of the automaton. name: The optional name of the automaton. """ ctx: context.Context scope: context.Scope name: t.Optional[str] _parameters: t.List[str] _locations: t.Set[Location] _initial_locations: t.Set[Location] _initial_restriction: t.Optional[expressions.Expression] _edges: t.List[Edge] _incoming_edges: t.DefaultDict[Location, t.Set[Edge]] _outgoing_edges: t.DefaultDict[Location, t.Set[Edge]] def __init__(self, ctx: context.Context, *, name: t.Optional[str] = None) -> None: self.ctx = ctx self.scope = self.ctx.global_scope.create_child_scope() self.name = name self._parameters = [] self._locations = set() self._initial_locations = set() self._initial_restriction = None self._edges = [] self._incoming_edges = collections.defaultdict(set) self._outgoing_edges = collections.defaultdict(set) def __repr__(self) -> str: return f"<Automaton name={self.name!r} at 0x{id(self):x}>" @property def locations(self) -> t.AbstractSet[Location]: """ The set of locations of the automaton. """ return self._locations @property def initial_locations(self) -> t.AbstractSet[Location]: """ The set of initial locations of the automaton. """ return self._initial_locations @property def edges(self) -> t.Sequence[Edge]: """ The set of edges of the automaton. """ return self._edges @property def initial_restriction(self) -> t.Optional[expressions.Expression]: """ An optional restriction to be satisfied by initial states. This property can be set *only once* per automaton. The expression has to be boolean. Raises :class:`~errors.ModelingError` when set twice or to a non-boolean expression. """ return self._initial_restriction @initial_restriction.setter def initial_restriction(self, initial_restriction: expressions.Expression) -> None: if self._initial_restriction is not None: raise errors.ModelingError( "restriction of initial environment has already been set" ) if self.scope.get_type(initial_restriction) != types.BOOL: raise errors.ModelingError( "restriction of initial valuations must have type `types.BOOL`" ) self._initial_restriction = initial_restriction @property def parameters(self) -> t.Sequence[str]: """ The sequence of parameters of the automaton. """ return self._parameters
[docs] def add_location(self, location: Location, *, initial: bool = False) -> None: """ Adds a location to the automaton. The flag `initial` specifies whether the location is an initial location. Raises :class:`~errors.ModelingError` when the location cannot be added to the automaton. See the method :meth:`Location.validate` for details. """ location.validate(self) self._locations.add(location) if initial: self._initial_locations.add(location)
[docs] def create_location( self, name: t.Optional[str] = None, *, progress_invariant: t.Optional[expressions.Expression] = None, transient_values: t.AbstractSet[Assignment] = frozenset(), initial: bool = False, ) -> Location: """ Creates a location with the given name and adds it to the automaton. The optional expression `progress_invariant` specifies the invariant subject to which the time may progress in the location. As per the JANI specification, the invariant is only allowed for models using real-valued clocks (see :class:`~momba.model.ModelType`). The parameter `transient_values` should be a set of assignments for transient variables. These assignments define the values of the transient variables when in the location. The flag `initial` specifies whether the location is an initial location. Raises the same exceptions as :meth:`add_location`. """ location = Location(name, progress_invariant, transient_values) self.add_location(location, initial=initial) return location
[docs] def add_edge(self, edge: Edge) -> None: """ Adds an edge to the automaton. Raises :class:`~errors.ModelingError` when the edge cannot be added to the automaton. See the method :meth:`Edge.validate` for details. """ edge.validate(self) self._edges.append(edge) self._locations.add(edge.location) self._outgoing_edges[edge.location].add(edge) for destination in edge.destinations: self._locations.add(destination.location) self._incoming_edges[destination.location].add(edge)
[docs] def create_edge( self, source: Location, destinations: t.Iterable[Destination], *, action_pattern: t.Optional[actions.ActionPattern] = None, guard: t.Optional[expressions.Expression] = None, rate: t.Optional[expressions.Expression] = None, annotation: t.Optional[Annotation] = None, observations: t.AbstractSet[observations.Observation] = frozenset(), ) -> None: """ Creates an edge and adds it to the automaton. The parameters are passed to :class:`Edge`. Raises the same exceptions as :meth:`add_edge`. """ edge = Edge( source, tuple(destinations), action_pattern, guard, rate, FrozenMap(annotation), frozenset(observations), ) self.add_edge(edge)
[docs] def get_incoming_edges(self, location: Location) -> t.AbstractSet[Edge]: """ Returns the set of outgoing edges from the given location. """ return self._incoming_edges[location]
[docs] def get_outgoing_edges(self, location: Location) -> t.AbstractSet[Edge]: """ Returns the set of incoming edges to the given location. """ return self._outgoing_edges[location]
[docs] def declare_variable( self, name: str, typ: types.Type, *, is_transient: t.Optional[bool] = None, initial_value: t.Optional[expressions.Expression] = None, ) -> None: """ Declares a variable in the local scope of the automaton. Passes the parameters to :meth:`Scope.declare_variable`. """ self.scope.declare_variable( name, typ, is_transient=is_transient, initial_value=initial_value )
[docs] def declare_parameter( self, name: str, typ: types.Type, *, default_value: t.Optional[expressions.Expression] = None, ) -> None: """ Declarse a parameter of the automaton. Passes the parameters to :meth:`Scope.declare_constant` where `value` is `default_value`. """ self.scope.declare_constant(name, typ, value=default_value) self._parameters.append(name)
[docs] def create_instance( self, *, arguments: t.Sequence[expressions.ValueOrExpression] = (), input_enable: t.AbstractSet[actions.ActionType] = frozenset(), comment: t.Optional[str] = None, ) -> Instance: """ Creates an instance of the automaton for composition. Passes the parameters to :class:`Instance`. """ return Instance( self, arguments=tuple(map(expressions.ensure_expr, arguments)), input_enable=frozenset(input_enable), )
Assignments = t.Union[ t.Iterable["Assignment"], t.Mapping[str, "expressions.Expression"] ]
[docs]def create_destination( location: Location, *, probability: t.Optional[expressions.Expression] = None, assignments: Assignments = (), ) -> Destination: """ Creates a destination with the given target location. This is a convenience function for creating destinations where assignments can be provied as a mapping. We recommend using it instead of creating :class:`Destination` objects directly. """ if isinstance(assignments, t.Mapping): return Destination( location, probability, assignments=tuple( Assignment(expressions.Name(name), value) for name, value in assignments.items() ), ) else: return Destination(location, probability, tuple(assignments))