16262675714

Committed 14 Jul 2025 04:49AM UTC coverage: 46.419% (-48.0%) from 94.444%

Build # 16262675714

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #3 from sisl/feat/core

Initial implementation of core AST algorithm.

Run Details

160 of 361 new or added lines in 18 files covered. (44.32%)

175 of 377 relevant lines covered (46.42%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.91

/src/astra_rl/core/environment.py

"""
environment.py
Roll out a problem, and specify how its environment behaves.
"""

from abc import abstractmethod, ABC
from dataclasses import dataclass
from typing import Sequence, Generic, Self, Optional

from astra_rl.core.common import StateT, ActionT
from astra_rl.core.problem import Problem


@dataclass
class Node(Generic[StateT, ActionT]):
    """A node in the rollout graph.

    Represents a single leaf in the rollout process, containing the context,
    the action taken, the response received, the reward for that action,
    and any children nodes that follow this action in this rollout.

    Attributes:
        context (StateT): The initial state before the action.
        attack (ActionT): The action taken in this node.
        response (StateT): The resulting state after the action.
        reward (float): The reward received for taking the action.
        children (Sequence[Node[StateT, ActionT]]): Subsequent nodes that follow this action.

    Generics:
        StateT (type): The type of the state in the environment.
        ActionT (type): The type of the action in the environment.
    """

    context: StateT
    attack: ActionT
    response: StateT
    reward: float

    children: Sequence[Self]


@dataclass
class Graph(Generic[StateT, ActionT]):
    """A graph representing the rollout (history + actions) of a problem.

    Attributes:
        context (StateT): The initial state of the environment.
        children (Sequence[Node[StateT, ActionT]]): The sequence of nodes representing actions and responses.
    """

    context: StateT
    children: Sequence[Node[StateT, ActionT]]


class Environment(ABC, Generic[StateT, ActionT]):
    """An Environment used for rolling out a problem.

    The primary point of this class is to make a `Graph` of the problem
    by calling the `rollout` method. The environment can keep/sample
    initial state, but should not have global state that persists
    across rollouts.

    Attributes:
        problem (Problem[StateT, ActionT]): The problem instance that defines the environment and actions.

    Generics:
        StateT (type): The type of the state in the environment.
        ActionT (type): The type of the action in the environment.
    """

    def __init__(self, problem: Problem[StateT, ActionT]):
        self.problem = problem

    @abstractmethod
    def rollout(self, seed: Optional[int] = None) -> Graph[StateT, ActionT]:
        """Roll out a problem and return a graph of the actions taken.

        Args:
            seed (Optional[int]): An optional seed; the same seed should produce the same graph.

        Returns:
            Graph[StateT, ActionT]: A graph representing the rollout of the problem.
        """

        pass

1	"""
2	environment.py
3	Roll out a problem, and specify how its environment behaves.
4	"""
5
6	from abc import abstractmethod, ABC	2✔
7	from dataclasses import dataclass	2✔
8	from typing import Sequence, Generic, Self, Optional	2✔
9
10	from astra_rl.core.common import StateT, ActionT	2✔
11	from astra_rl.core.problem import Problem	2✔
12
13
14	@dataclass	2✔
15	class Node(Generic[StateT, ActionT]):	2✔
16	"""A node in the rollout graph.
17
18	Represents a single leaf in the rollout process, containing the context,
19	the action taken, the response received, the reward for that action,
20	and any children nodes that follow this action in this rollout.
21
22	Attributes:
23	context (StateT): The initial state before the action.
24	attack (ActionT): The action taken in this node.
25	response (StateT): The resulting state after the action.
26	reward (float): The reward received for taking the action.
27	children (Sequence[Node[StateT, ActionT]]): Subsequent nodes that follow this action.
28
29	Generics:
30	StateT (type): The type of the state in the environment.
31	ActionT (type): The type of the action in the environment.
32	"""
33
34	context: StateT	2✔
35	attack: ActionT	2✔
36	response: StateT	2✔
37	reward: float	2✔
38
39	children: Sequence[Self]	2✔
40
41
42	@dataclass	2✔
43	class Graph(Generic[StateT, ActionT]):	2✔
44	"""A graph representing the rollout (history + actions) of a problem.
45
46	Attributes:
47	context (StateT): The initial state of the environment.
48	children (Sequence[Node[StateT, ActionT]]): The sequence of nodes representing actions and responses.
49	"""
50
51	context: StateT	2✔
52	children: Sequence[Node[StateT, ActionT]]	2✔
53
54
55	class Environment(ABC, Generic[StateT, ActionT]):	2✔
56	"""An Environment used for rolling out a problem.
57
58	The primary point of this class is to make a `Graph` of the problem
59	by calling the `rollout` method. The environment can keep/sample
60	initial state, but should not have global state that persists
61	across rollouts.
62
63	Attributes:
64	problem (Problem[StateT, ActionT]): The problem instance that defines the environment and actions.
65
66	Generics:
67	StateT (type): The type of the state in the environment.
68	ActionT (type): The type of the action in the environment.
69	"""
70
71	def __init__(self, problem: Problem[StateT, ActionT]):	2✔
NEW 72	self.problem = problem	×
73
74	@abstractmethod	2✔
75	def rollout(self, seed: Optional[int] = None) -> Graph[StateT, ActionT]:	2✔
76	"""Roll out a problem and return a graph of the actions taken.
77
78	Args:
79	seed (Optional[int]): An optional seed; the same seed should produce the same graph.
80
81	Returns:
82	Graph[StateT, ActionT]: A graph representing the rollout of the problem.
83	"""
84
NEW 85	pass	×

sisl / astra-rl / 16262675714

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous