Skip to content

Constraints

masa.common.constraints.base.Constraint

Bases: Protocol

Protocol for stateful constraint monitors.

A Constraint is a monitor that consumes atomic proposition labels at each step and maintains internal state (e.g., cumulative cost, whether an LTL automaton is in an accepting/unsafe state, etc.).

Implementations are intended to be lightweight and compatible with Gymnasium wrappers: call reset at episode start and update after each environment transition using the label set from info["labels"].

Required interface

Implementations should provide:

  • reset: clear any episode state.
  • update: incorporate the current label set.
  • constraint_type: a stable identifier string for logging/dispatch.

Metrics interface

The protocol declares:

  • step_metric
  • episode_metric

constraint_type property 

constraint_type: str

A stable identifier for the constraint (e.g., "cmdp", "ltl_safety").

reset 

reset()

Reset any episode-dependent internal state.

Source code in masa/common/constraints/base.py 
88
89
def reset(self):
    """Reset any episode-dependent internal state."""

update 

update(labels: Iterable[str])

Update internal state given the current set of labels.

Parameters:

Name Type Description Default
labels Iterable[str]

Iterable of atomic proposition strings active at the current step (typically taken from info["labels"]).

 required
Source code in masa/common/constraints/base.py 
91
92
93
94
95
96
97
def update(self, labels: Iterable[str]):
    """Update internal state given the current set of labels.

    Args:
        labels: Iterable of atomic proposition strings active at the current
            step (typically taken from ``info["labels"]``).
    """

step_metric 

step_metric() -> Dict[str, float]

Return per-step logging metrics.

Metrics returned here should be:

  • cheap to compute,
  • non-destructive (do not mutate state),
  • meaningful at any time step.

Examples include running cumulative cost, a per-step violation flag, a current probability estimate, etc.

Returns:

Type Description
Dict[str, float]

Dictionary of scalar metrics (values should be JSON/log friendly).
Source code in masa/common/constraints/base.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def step_metric(self) -> Dict[str, float]:
    """Return per-step logging metrics.

    Metrics returned here should be:

    - cheap to compute,
    - non-destructive (do not mutate state),
    - meaningful at *any* time step.

    Examples include running cumulative cost, a per-step violation flag,
    a current probability estimate, etc.

    Returns:
        Dictionary of scalar metrics (values should be JSON/log friendly).
    """

episode_metric 

episode_metric() -> Dict[str, float]

Return end-of-episode logging metrics.

This is intended to summarize what matters for evaluation/logging at episode termination (terminated or truncated).

Returns:

Type Description
Dict[str, float]

Dictionary of scalar metrics (values should be JSON/log friendly).
Source code in masa/common/constraints/base.py 
120
121
122
123
124
125
126
127
128
def episode_metric(self) -> Dict[str, float]:
    """Return end-of-episode logging metrics.

    This is intended to summarize what matters for evaluation/logging at
    episode termination (terminated or truncated).

    Returns:
        Dictionary of scalar metrics (values should be JSON/log friendly).
    """

masa.common.constraints.base.BaseConstraintEnv 

BaseConstraintEnv(env: Env, constraint: Constraint, **kw)

Bases: Wrapper, Constraint

Common base wrapper for constraint-aware environments.

This wrapper enforces the MASA convention that the wrapped environment is a LabelledEnv and provides info["labels"] as a set (or frozenset) of atomic propositions at each step.

The wrapper:

  1. Delegates reset/step to the underlying environment.
  2. Extracts labels = info.get("labels", set()).
  3. Validates that labels is a set-like container of strings.
  4. Calls self._constraint.update(labels).

Attributes:

Name Type Description
env

The wrapped Gymnasium environment (must be a LabelledEnv).
_constraint

The underlying constraint monitor.

Raises:

Type Description
TypeError

If env is not an instance of LabelledEnv.
ValueError

If info["labels"] exists but is not a set/frozenset.
Notes

The properties label_fn and cost_fn are convenience accessors for downstream algorithms. Depending on how wrappers are composed, these may be None.

Initialize the wrapper.

Parameters:

Name Type Description Default
env Env

Base environment. Must already be wrapped as a LabelledEnv so that step/reset provide label sets in info["labels"].

 required
constraint Constraint

A constraint monitor implementing Constraint.

 required
**kw

Unused extra keyword arguments (kept for wrapper compatibility).

 {}

Raises:

Type Description
TypeError

If env is not a LabelledEnv.
Source code in masa/common/constraints/base.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def __init__(self, env: gym.Env, constraint: Constraint, **kw):
    """Initialize the wrapper.

    Args:
        env: Base environment. Must already be wrapped as a
            :class:`~masa.common.labelled_env.LabelledEnv` so that step/reset
            provide label sets in ``info["labels"]``.
        constraint: A constraint monitor implementing :class:`Constraint`.
        **kw: Unused extra keyword arguments (kept for wrapper compatibility).

    Raises:
        TypeError: If ``env`` is not a :class:`LabelledEnv`.
    """
    if not isinstance(env, LabelledEnv):
        raise TypeError(
            f"{self.__class__.__name__} must wrap a LabelledEnv, "
            f"but got {type(env).__name__}. "
            "Please wrap your environment with LabelledEnv before applying a constraint wrapper."
        )

    super().__init__(env)
    self._constraint = constraint

_constraint instance-attribute 

_constraint = constraint

cost_fn property 

cost_fn

Expose the cost function if available.

Returns:

Type Description

The underlying cost function if present on the wrapped stack, else

None.

label_fn property 

label_fn

Expose the labelling function if available.

Returns:

Type Description

The environment labelling function if present, else None.

constraint_type property 

constraint_type: str

Constraint identifier forwarded from the underlying monitor.

reset 

reset(*, seed: int | None = None, options: Dict[str, Any] | None = None)

Reset environment and constraint state.

This calls env.reset(...) and then resets and updates the constraint using the initial label set in info["labels"].

Parameters:

Name Type Description Default
seed int | None

Optional RNG seed forwarded to the base environment.

 None
options Dict[str, Any] | None

Optional reset options forwarded to the base environment.

 None

Returns:

Type Description

A tuple (obs, info) following the Gymnasium API.

Raises:

Type Description
ValueError

If info["labels"] is present but not a set/frozenset.
Source code in masa/common/constraints/base.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def reset(self, *, seed: int | None = None, options: Dict[str, Any] | None = None):
    """Reset environment and constraint state.

    This calls ``env.reset(...)`` and then resets and updates the constraint
    using the initial label set in ``info["labels"]``.

    Args:
        seed: Optional RNG seed forwarded to the base environment.
        options: Optional reset options forwarded to the base environment.

    Returns:
        A tuple ``(obs, info)`` following the Gymnasium API.

    Raises:
        ValueError: If ``info["labels"]`` is present but not a set/frozenset.
    """
    obs, info = self.env.reset(seed=seed, options=options)
    self._constraint.reset()

    labels = info.get("labels", set())
    if not isinstance(labels, (set, frozenset)):
        raise ValueError(
            f"Expected 'labels' in info to be a set of atomic propositions, got {type(labels).__name__}"
        )

    self._constraint.update(labels)
    return obs, info

step 

step(action: Any)

Step environment and update constraint from labels.

Parameters:

Name Type Description Default
action Any

Action to pass to the underlying environment.

 required

Returns:

Type Description

A 5-tuple (obs, reward, terminated, truncated, info) following

the Gymnasium API.

Raises:

Type Description
ValueError

If info["labels"] is present but not a set/frozenset.
Source code in masa/common/constraints/base.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def step(self, action: Any):
    """Step environment and update constraint from labels.

    Args:
        action: Action to pass to the underlying environment.

    Returns:
        A 5-tuple ``(obs, reward, terminated, truncated, info)`` following
        the Gymnasium API.

    Raises:
        ValueError: If ``info["labels"]`` is present but not a set/frozenset.
    """

    obs, reward, terminated, truncated, info = self.env.step(action)

    labels = info.get("labels", set())
    if not isinstance(labels, (set, frozenset)):
        raise ValueError(
            f"Expected 'labels' in info to be a set of atomic propositions, got {type(labels).__name__}"
        )

    self._constraint.update(labels)
    return obs, reward, terminated, truncated, info

constraint_step_metrics 

constraint_step_metrics() -> Dict[str, float]

Return per-step metrics from the underlying constraint.

Returns:

Type Description
Dict[str, float]

Dictionary of scalar metrics.
Source code in masa/common/constraints/base.py 
261
262
263
264
265
266
267
def constraint_step_metrics(self) -> Dict[str, float]:
    """Return per-step metrics from the underlying constraint.

    Returns:
        Dictionary of scalar metrics.
    """
    return self._constraint.step_metric()

constraint_episode_metrics 

constraint_episode_metrics() -> Dict[str, float]

Return end-of-episode metrics from the underlying constraint.

Returns:

Type Description
Dict[str, float]

Dictionary of scalar metrics.
Source code in masa/common/constraints/base.py 
269
270
271
272
273
274
275
def constraint_episode_metrics(self) -> Dict[str, float]:
    """Return end-of-episode metrics from the underlying constraint.

    Returns:
        Dictionary of scalar metrics.
    """
    return self._constraint.episode_metric()

Next Steps

  • CMDP - Budgeted Constrained MDP.
  • LTL Safety - Safety fragment of LTL as a monitor and constraint.
  • PCTL - A simple Probabilistic Computation Tree Logic constraint.
  • Step-wise Probabilistic - Undiscounted probabilistic step-wise safety constraint.
  • Reach Avoid - A simple reach-avoid constraint.
  • ATL (Multi Agent) - Alternating Time Logic for Multi Agent Systems.