Skip to content

Probabilistic Computation Tree Logic (PCTL) Constraint

masa.common.constraints.pctl.PCTL 

PCTL(cost_fn: CostFn, alpha: float)

Bases: Constraint

Simplified PCTL-named monitor tracking whether any unsafe step occurred.

Parameters:

Name Type Description Default
cost_fn CostFn

Mapping from label sets to scalar cost.

 required
alpha float

Threshold parameter stored for downstream use (not currently used in the logic in this file).

 required

Attributes:

Name Type Description
cost_fn

Cost function labels -> float.
alpha

User-specified parameter (reserved for probabilistic variants).
safe

True until an unsafe cost is observed.
step_cost

Most recent cost value.
total_unsafe

Count of unsafe steps (as floats).
Source code in masa/common/constraints/pctl.py 
48
49
50
def __init__(self, cost_fn: CostFn, alpha: float):
    self.cost_fn = cost_fn
    self.alpha = alpha

cost_fn instance-attribute 

cost_fn = cost_fn

alpha instance-attribute 

alpha = alpha

constraint_type property 

constraint_type: str

Stable identifier string: "PCTL".

reset 

reset()

Reset episode counters.

Source code in masa/common/constraints/pctl.py 
52
53
54
55
56
def reset(self):
    """Reset episode counters."""
    self.safe = True
    self.step_cost = 0.0
    self.total_unsafe = 0.0

update 

update(labels: Iterable[str])

Update safety flags from the current label set.

Parameters:

Name Type Description Default
labels Iterable[str]

Iterable of atomic propositions.

 required
Source code in masa/common/constraints/pctl.py 
58
59
60
61
62
63
64
65
66
def update(self, labels: Iterable[str]):
    """Update safety flags from the current label set.

    Args:
        labels: Iterable of atomic propositions.
    """
    self.step_cost = self.cost_fn(labels)
    self.total_unsafe += float(self.step_cost >= 0.5)
    self.safe = self.safe and (not self.total_unsafe)

satisfied 

satisfied() -> bool

Whether the episode remains safe so far.

Source code in masa/common/constraints/pctl.py 
68
69
70
def satisfied(self) -> bool:
    """Whether the episode remains safe so far."""
    return self.safe

episode_metric 

episode_metric() -> Dict[str, float]

End-of-episode metrics.

Returns:

Type Description
Dict[str, float]

Dict containing:
Dict[str, float]
  • "cum_unsafe": count of unsafe steps,
Dict[str, float]
  • "satisfied": 1.0 if safe else 0.0.
Source code in masa/common/constraints/pctl.py 
72
73
74
75
76
77
78
79
80
81
def episode_metric(self) -> Dict[str, float]:
    """End-of-episode metrics.

    Returns:
        Dict containing:

        - ``"cum_unsafe"``: count of unsafe steps,
        - ``"satisfied"``: 1.0 if safe else 0.0.
    """
    return {"cum_unsafe": float(self.total_unsafe), "satisfied": float(self.satisfied())}

step_metric 

step_metric() -> Dict[str, float]

Per-step metrics.

Returns:

Type Description
Dict[str, float]

Dict containing:
Dict[str, float]
  • "cost": current step cost,
Dict[str, float]
  • "violation": 1.0 if cost >= 0.5 else 0.0.
Source code in masa/common/constraints/pctl.py 
83
84
85
86
87
88
89
90
91
92
def step_metric(self) -> Dict[str, float]:
    """Per-step metrics.

    Returns:
        Dict containing:

        - ``"cost"``: current step cost,
        - ``"violation"``: 1.0 if ``cost >= 0.5`` else 0.0.
    """
    return {"cost": self.step_cost, "violation": float(self.step_cost >= 0.5), "cost_done": float(self.step_cost >= 0.5)}

masa.common.constraints.pctl.PCTLEnv 

PCTLEnv(env: Env, cost_fn: CostFn = dummy_cost_fn, alpha: float = 0.01, **kw)

Bases: BaseConstraintEnv

Gymnasium wrapper for the PCTL monitor.

Parameters:

Name Type Description Default
env Env

Base environment (must be a LabelledEnv).

 required
cost_fn CostFn

Cost function mapping labels to a scalar.

 cost_fn
alpha float

Threshold parameter stored on the monitor (see PCTL).

 0.01
**kw

Extra keyword arguments forwarded to BaseConstraintEnv.

 {}
Source code in masa/common/constraints/pctl.py 
110
111
def __init__(self, env: gym.Env, cost_fn: CostFn = dummy_cost_fn, alpha: float = 0.01, **kw):
    super().__init__(env, PCTL(cost_fn=cost_fn, alpha=alpha), **kw)