Spaces:
Sleeping
Sleeping
File size: 2,469 Bytes
81e328b 1aaa98d 81e328b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """
Pydantic models for the PLL Cyberattack Detection OpenEnv.
Defines Observation, Action, Reward, and State schemas.
"""
import numpy as np
from typing import Annotated, Any, Dict, List, Optional
from pydantic import BaseModel, Field, model_validator
# Exactly 20 floats — enforced at validation time, not just documented.
WindowList = Annotated[List[float], Field(min_length=20, max_length=20)]
# Exactly 3 floats for [va, vb, vc].
VoltageList = Annotated[List[float], Field(min_length=3, max_length=3)]
class Observation(BaseModel):
vq_window: WindowList
vd_window: WindowList
omega_window: WindowList
omega_deviation_window: WindowList
raw_voltages: VoltageList
task_id: int = Field(ge=0, le=2)
step: int = Field(ge=0)
class Action(BaseModel):
attack_detected: bool
attack_type: int = Field(ge=0, le=4)
confidence: float = Field(ge=0.0, le=1.0)
protective_action: int = Field(ge=0, le=3)
class Reward(BaseModel):
total: float
detection_reward: float
classification_bonus: float
early_detection_bonus: float
false_alarm_penalty: float
lock_loss_penalty: float
class State(BaseModel):
theta_true: float
theta_hat: float
omega_hat: float
vq_integral: float
attack_active: bool
attack_type: int # Integer ID of the current attack: 0=none, 1=sinusoidal, 2=ramp, 3=pulse, 4=stealthy.
attack_params: Dict[str, Any]
attack_start_step: int
lock_lost: bool # Whether the PLL has lost lock (|theta_err| > 5°). Task 2 only.
step: int = Field(ge=0)
episode_id: str
task_id: int = Field(ge=0, le=2)
@model_validator(mode="before")
@classmethod
def coerce_attack_params(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""
This validator ensures JSON serialization never fails due
to np.float32 / np.int64 / np.bool_ leaking into the params dict.
"""
params = values.get("attack_params", {})
if isinstance(params, dict):
coerced = {}
for k, v in params.items():
if isinstance(v, np.floating):
coerced[k] = float(v)
elif isinstance(v, np.integer):
coerced[k] = int(v)
elif isinstance(v, np.bool_):
coerced[k] = bool(v)
else:
coerced[k] = v
values["attack_params"] = coerced
return values
|