Skip to content

DSL

dsl

TypeCheckTransformer 

TypeCheckTransformer(factor_schema: dict[str, DslType], function_schema: dict[str, list[dict[str, Any]]])

Bases: Transformer[Any, DslType]

Static analysis transformer to perform type checking on the AST.

Initialize the type checker with provided schemas.

Args: factor_schema: A dictionary mapping factor names to their expected DslType. function_schema: A dictionary mapping function names to a list of overloads, each {"args": list[Arg], "return": DslType}. Order matters: first matching overload wins. Optional args (those carrying a default) must be trailing within each overload.

Source code in ctalearn/dsl/analyzer.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    factor_schema: dict[str, DslType],
    function_schema: dict[str, list[dict[str, Any]]],
) -> None:
    """Initialize the type checker with provided schemas.

    Args:
        factor_schema: A dictionary mapping factor names to their expected DslType.
        function_schema: A dictionary mapping function names to a list of
        overloads, each `{"args": list[Arg], "return": DslType}`. Order matters:
        first matching overload wins. Optional args (those carrying a default)
        must be trailing within each overload.
    """
    super().__init__()
    self.factor_schema = factor_schema
    self.function_schema = function_schema
    self.local_env: dict[str, DslType] = {}

number 

number(token: Token) -> DslType

Infer the type of a numeric token.

Source code in ctalearn/dsl/analyzer.py 
32
33
34
def number(self, token: Token) -> DslType:
    """Infer the type of a numeric token."""
    return DslType.FLOAT if "." in token.value else DslType.INT

variable 

variable(token: Token) -> DslType

Infer the type of a variable token.

Checks local assignment cache first, then falls back to the global factor schema.

Args: token: The parsed variable token.

Returns: The resolved DslType.

Raises: DslTypeError: If the variable is not found in either the environment or schema.

Source code in ctalearn/dsl/analyzer.py 
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
def variable(self, token: Token) -> DslType:
    """Infer the type of a variable token.

    Checks local assignment cache first, then falls back to the global factor
    schema.

    Args:
        token: The parsed variable token.

    Returns:
        The resolved DslType.

    Raises:
        DslTypeError: If the variable is not found in either the environment or
        schema.
    """
    var_name = token.value

    # 1. Check if it's a defined local variable
    if var_name in self.local_env:
        return self.local_env[var_name]

    # 2. Check predefined factor schema (global data sources)
    if var_name in self.factor_schema:
        return self.factor_schema[var_name]

    raise DslTypeError(f"Unknown variable: '{var_name}'")

add 

add(left: DslType, right: DslType) -> DslType

Infer the return type for an addition operation.

Source code in ctalearn/dsl/analyzer.py 
68
69
70
def add(self, left: DslType, right: DslType) -> DslType:
    """Infer the return type for an addition operation."""
    return self._binop(left, right)

sub 

sub(left: DslType, right: DslType) -> DslType

Infer the return type for a subtraction operation.

Source code in ctalearn/dsl/analyzer.py 
72
73
74
def sub(self, left: DslType, right: DslType) -> DslType:
    """Infer the return type for a subtraction operation."""
    return self._binop(left, right)

mul 

mul(left: DslType, right: DslType) -> DslType

Infer the return type for a multiplication operation.

Source code in ctalearn/dsl/analyzer.py 
76
77
78
def mul(self, left: DslType, right: DslType) -> DslType:
    """Infer the return type for a multiplication operation."""
    return self._binop(left, right)

div 

div(left: DslType, right: DslType) -> DslType

Infer the return type for a division operation.

Source code in ctalearn/dsl/analyzer.py 
80
81
82
def div(self, left: DslType, right: DslType) -> DslType:
    """Infer the return type for a division operation."""
    return self._binop(left, right)

neg 

neg(val_type: DslType) -> DslType

Infer the return type for a negation operation.

Source code in ctalearn/dsl/analyzer.py 
84
85
86
def neg(self, val_type: DslType) -> DslType:
    """Infer the return type for a negation operation."""
    return val_type

func_call 

func_call(func_name_token: Token, *args_types: DslType) -> DslType

Resolve overloads and infer the return type of a function call.

Args: func_name_token: The parsed function name token. *args_types: The inferred types of the arguments passed to the function.

Returns: The return type of the first overload whose signature accepts args_types.

Raises: DslTypeError: If the function is unknown or no overload accepts the given arg types.

Source code in ctalearn/dsl/analyzer.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def func_call(self, func_name_token: Token, *args_types: DslType) -> DslType:
    """Resolve overloads and infer the return type of a function call.

    Args:
        func_name_token: The parsed function name token.
        *args_types: The inferred types of the arguments passed to the function.

    Returns:
        The return type of the first overload whose signature accepts
        `args_types`.

    Raises:
        DslTypeError: If the function is unknown or no overload accepts the
        given arg types.
    """
    func_name = func_name_token.value
    if func_name not in self.function_schema:
        raise DslTypeError(f"Unknown function: '{func_name}'")

    actual = list(args_types)
    for overload in self.function_schema[func_name]:
        if match_signature(overload["args"], actual):
            return_type: DslType = overload["return"]
            return return_type

    raise DslTypeError(f"No matching overload for '{func_name}'")

statement 

statement(var_name_token: Token, expr_type: DslType) -> None

Register a variable assignment type into the local environment.

Args: var_name_token: The variable name token. expr_type: The inferred type of the assigned expression.

Source code in ctalearn/dsl/analyzer.py 
115
116
117
118
119
120
121
122
123
def statement(self, var_name_token: Token, expr_type: DslType) -> None:
    """Register a variable assignment type into the local environment.

    Args:
        var_name_token: The variable name token.
        expr_type: The inferred type of the assigned expression.
    """
    self.local_env[var_name_token.value] = expr_type
    return None

return_stmt 

return_stmt(expr_type: DslType) -> DslType

Process and pass through the final return statement type.

Source code in ctalearn/dsl/analyzer.py 
125
126
127
def return_stmt(self, expr_type: DslType) -> DslType:
    """Process and pass through the final return statement type."""
    return expr_type

start 

start(*statements_and_return: DslType | None) -> DslType

Evaluate the AST entry point and validate the final return type.

Args: *statements_and_return: The evaluated types of all statements and the final return.

Returns: The final output type of the script.

Raises: DslTypeError: If the final returned type is not a DataFrame.

Source code in ctalearn/dsl/analyzer.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def start(self, *statements_and_return: DslType | None) -> DslType:
    """Evaluate the AST entry point and validate the final return type.

    Args:
        *statements_and_return: The evaluated types of all statements
        and the final return.

    Returns:
        The final output type of the script.

    Raises:
        DslTypeError: If the final returned type is not a DataFrame.
    """
    final_type = statements_and_return[-1]

    if final_type is None or final_type != DslType.DATAFRAME:
        got = "None" if final_type is None else final_type.value
        raise DslTypeError(f"Return type should be DataFrame, got {got}")

    return final_type

DslError

Bases: Exception

Base exception class for all DSL-related errors.

DslRuntimeError

Bases: DslError

Raised during AST execution due to failed data fetching or function crashes.

DslSyntaxError

Bases: DslError

Raised when the parser encounters invalid syntax in the DSL script.

DslTypeError

Bases: DslError

Raised during static analysis for type mismatches or unknown variables.

ExecutionTransformer 

ExecutionTransformer(functions: dict[str, list[tuple[Callable[..., Any], list[Arg]]]], data_loaders: dict[str, Callable[[], Any]])

Bases: Transformer[Any, Any]

Runtime transformer to execute the validated AST.

Initialize the execution transformer.

Args: functions: A dictionary mapping function names to a list of overloads, each (callable, params). First overload whose params accept the actual arg types is dispatched. data_loaders: A dictionary of lazy-loading callables to fetch data.

Source code in ctalearn/dsl/interpreter.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    functions: dict[str, list[tuple[Callable[..., Any], list[Arg]]]],
    data_loaders: dict[str, Callable[[], Any]],
) -> None:
    """Initialize the execution transformer.

    Args:
        functions: A dictionary mapping function names to a list of overloads,
        each `(callable, params)`. First overload whose params accept the
        actual arg types is dispatched.
        data_loaders: A dictionary of lazy-loading callables to fetch data.
    """
    super().__init__()
    self.functions = functions
    self.data_loaders = data_loaders
    self.local_env: dict[str, Any] = {}

number 

number(token: Token) -> float | int

Parse a numeric token into a float or integer.

Source code in ctalearn/dsl/interpreter.py 
52
53
54
def number(self, token: Token) -> float | int:
    """Parse a numeric token into a float or integer."""
    return float(token.value) if "." in token.value else int(token.value)

variable 

variable(token: Token) -> Any

Resolve a variable by fetching from local cache or triggering data loaders.

Args: token: The parsed variable token.

Returns: The underlying data payload (e.g., a DataFrame or Series).

Raises: DslRuntimeError: If data loading fails or returns None.

Source code in ctalearn/dsl/interpreter.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def variable(self, token: Token) -> Any:
    """Resolve a variable by fetching from local cache or triggering data loaders.

    Args:
        token: The parsed variable token.

    Returns:
        The underlying data payload (e.g., a DataFrame or Series).

    Raises:
        DslRuntimeError: If data loading fails or returns None.
    """
    var_name = token.value

    # 1. Fetch from local cache (calculated variables)
    if var_name in self.local_env:
        return self.local_env[var_name]

    # 2. Raise error if data loader is not defined
    if var_name not in self.data_loaders:
        raise DslRuntimeError(f"Unknown variable '{var_name}'")

    # 3. Lazy Loading: Execute the callable to fetch data
    try:
        data = self.data_loaders[var_name]()
    except Exception as e:
        raise DslRuntimeError(f"Failed to fetch '{var_name}': {e}")

    if data is None:
        raise DslRuntimeError(f"'{var_name}' is None after fetching.")

    self.local_env[var_name] = data
    return data

add 

add(left: Any, right: Any) -> Any

Execute an addition operation.

Source code in ctalearn/dsl/interpreter.py 
90
91
92
def add(self, left: Any, right: Any) -> Any:
    """Execute an addition operation."""
    return operator.add(left, right)

sub 

sub(left: Any, right: Any) -> Any

Execute a subtraction operation.

Source code in ctalearn/dsl/interpreter.py 
94
95
96
def sub(self, left: Any, right: Any) -> Any:
    """Execute a subtraction operation."""
    return operator.sub(left, right)

mul 

mul(left: Any, right: Any) -> Any

Execute a multiplication operation.

Source code in ctalearn/dsl/interpreter.py 
 98
 99
100
def mul(self, left: Any, right: Any) -> Any:
    """Execute a multiplication operation."""
    return operator.mul(left, right)

div 

div(left: Any, right: Any) -> Any

Execute a division operation.

Source code in ctalearn/dsl/interpreter.py 
102
103
104
def div(self, left: Any, right: Any) -> Any:
    """Execute a division operation."""
    return operator.truediv(left, right)

neg 

neg(val: Any) -> Any

Execute a negation operation.

Source code in ctalearn/dsl/interpreter.py 
106
107
108
def neg(self, val: Any) -> Any:
    """Execute a negation operation."""
    return operator.neg(val)

func_call 

func_call(func_name_token: Token, *args: Any) -> Any

Execute a registered function with the provided arguments.

Args: func_name_token: The token containing the function name. *args: The evaluated arguments to pass into the function.

Returns: The result of the underlying function execution.

Raises: DslRuntimeError: If the underlying function crashes during execution.

Source code in ctalearn/dsl/interpreter.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def func_call(self, func_name_token: Token, *args: Any) -> Any:
    """Execute a registered function with the provided arguments.

    Args:
        func_name_token: The token containing the function name.
        *args: The evaluated arguments to pass into the function.

    Returns:
        The result of the underlying function execution.

    Raises:
        DslRuntimeError: If the underlying function crashes during execution.
    """
    func_name = func_name_token.value
    overloads = self.functions.get(func_name)
    if overloads is None:
        raise DslRuntimeError(f"Unknown function '{func_name}'")

    # Single overload: analyzer already validated types; dispatch is trivial.
    # Multi-overload: inspect runtime types to pick the right callable.
    if len(overloads) == 1:
        fn: Callable[..., Any] = overloads[0][0]
    else:
        actual = [_py_to_dsl(a) for a in args]
        for cand, params in overloads:
            if match_signature(params, actual):
                fn = cand
                break
        else:
            # Analyzer guarantees a match; reachable only if host bypassed it.
            raise DslRuntimeError(f"No matching overload for '{func_name}'")

    try:
        return fn(*args)
    except Exception as e:
        raise DslRuntimeError(f"Failed to execution function '{func_name}': {e}")

statement 

statement(var_name_token: Token, expr_result: Any) -> None

Store the result of an assignment expression into the local cache.

Source code in ctalearn/dsl/interpreter.py 
147
148
149
150
def statement(self, var_name_token: Token, expr_result: Any) -> None:
    """Store the result of an assignment expression into the local cache."""
    self.local_env[var_name_token.value] = expr_result
    return None

return_stmt 

return_stmt(expr_result: Any) -> Any

Process and return the final execution result.

Source code in ctalearn/dsl/interpreter.py 
152
153
154
def return_stmt(self, expr_result: Any) -> Any:
    """Process and return the final execution result."""
    return expr_result

start 

start(*statements_and_return: Any) -> Any

Process the AST entry point and return the final executed result.

Source code in ctalearn/dsl/interpreter.py 
156
157
158
def start(self, *statements_and_return: Any) -> Any:
    """Process the AST entry point and return the final executed result."""
    return statements_and_return[-1]

Arg dataclass 

Arg(type: DslType, default: Any = _MISSING)

Type signature of a single DSL function argument.

Attributes: type: The expected DslType of the argument. default: The default value. Left as the _MISSING sentinel for required arguments; any other value (including None) marks the argument optional.

required property 

required: bool

Whether the argument must be supplied by the caller.

DslType

Bases: Enum

Enumeration of supported virtual data types in the DSL.