move out of syntax-tree to script'

This commit is contained in:
Jesse Bannon 2023-11-22 12:06:50 -08:00
parent 26eca6668c
commit 8dc695bdd1
2 changed files with 59 additions and 80 deletions

View file

@ -1,10 +1,12 @@
from typing import Dict from typing import Dict
from typing import List
from typing import Optional from typing import Optional
from ytdl_sub.script.parser import parse from ytdl_sub.script.parser import parse
from ytdl_sub.script.types.resolvable import Resolvable from ytdl_sub.script.types.resolvable import Resolvable
from ytdl_sub.script.types.syntax_tree import SyntaxTree from ytdl_sub.script.types.syntax_tree import SyntaxTree
from ytdl_sub.script.types.variable import Variable from ytdl_sub.script.types.variable import Variable
from ytdl_sub.script.utils.exceptions import CycleDetected
class Script: class Script:
@ -26,6 +28,32 @@ class Script:
""" """
return function_key[1:] return function_key[1:]
def _traverse_custom_function_dependencies(
self,
custom_function_name: str,
custom_function_dependency: SyntaxTree,
deps: List[str],
) -> None:
for dep in custom_function_dependency.custom_functions:
if custom_function_name in deps + [dep.name]:
cycle_deps = [custom_function_name] + deps + [dep.name]
cycle_deps_str = " -> ".join([f"%{name}" for name in cycle_deps])
raise CycleDetected(f"Custom functions contain a cycle: {cycle_deps_str}")
self._traverse_custom_function_dependencies(
custom_function_name=custom_function_name,
custom_function_dependency=self._functions[dep.name],
deps=deps + [dep.name],
)
def _ensure_no_custom_function_cycles(self):
for custom_function_name, custom_function in self._functions.items():
self._traverse_custom_function_dependencies(
custom_function_name=custom_function_name,
custom_function_dependency=custom_function,
deps=[],
)
def __init__(self, overrides: Dict[str, str]): def __init__(self, overrides: Dict[str, str]):
self._functions: Dict[str, SyntaxTree] = { self._functions: Dict[str, SyntaxTree] = {
# custom_function_name must be passed to properly type custom function # custom_function_name must be passed to properly type custom function
@ -43,6 +71,8 @@ class Script:
if not self._is_function(override_name) if not self._is_function(override_name)
} }
self._ensure_no_custom_function_cycles()
def resolve( def resolve(
self, pre_resolved_variables: Optional[Dict[Variable, Resolvable]] = None self, pre_resolved_variables: Optional[Dict[Variable, Resolvable]] = None
) -> Dict[str, Resolvable]: ) -> Dict[str, Resolvable]:
@ -56,8 +86,32 @@ class Script:
------- -------
Dict of resolved values Dict of resolved values
""" """
return SyntaxTree.resolve_overrides( overrides: Dict[Variable, SyntaxTree] = {
parsed_overrides=self._variables, Variable(name): ast for name, ast in self._variables.items()
custom_functions=self._functions, }
pre_resolved_variables=pre_resolved_variables,
unresolved_variables: List[Variable] = list(overrides.keys())
resolved_variables: Dict[Variable, Resolvable] = (
pre_resolved_variables if pre_resolved_variables else {}
) )
while unresolved_variables:
unresolved_count: int = len(unresolved_variables)
for variable in unresolved_variables:
if not overrides[variable].has_variable_dependency(
resolved_variables=resolved_variables
):
resolved_variables[variable] = overrides[variable].resolve(
resolved_variables=resolved_variables,
custom_functions=self._functions,
)
unresolved_variables.remove(variable)
if len(unresolved_variables) == unresolved_count:
raise CycleDetected(
f"Cycle detected within these variables: "
f"{', '.join(sorted([var.name for var in unresolved_variables]))}"
)
return {variable.name: resolvable for variable, resolvable in resolved_variables.items()}

View file

@ -1,17 +1,12 @@
import copy
from dataclasses import dataclass from dataclasses import dataclass
from typing import Dict from typing import Dict
from typing import List from typing import List
from typing import Optional
from ytdl_sub.script.types.resolvable import ArgumentType from ytdl_sub.script.types.resolvable import ArgumentType
from ytdl_sub.script.types.resolvable import NamedCustomFunction
from ytdl_sub.script.types.resolvable import Resolvable from ytdl_sub.script.types.resolvable import Resolvable
from ytdl_sub.script.types.resolvable import String from ytdl_sub.script.types.resolvable import String
from ytdl_sub.script.types.variable import Variable from ytdl_sub.script.types.variable import Variable
from ytdl_sub.script.types.variable_dependency import VariableDependency from ytdl_sub.script.types.variable_dependency import VariableDependency
from ytdl_sub.script.utils.exceptions import CycleDetected
from ytdl_sub.utils.exceptions import StringFormattingException
@dataclass(frozen=True) @dataclass(frozen=True)
@ -25,7 +20,7 @@ class SyntaxTree(VariableDependency):
def resolve( def resolve(
self, self,
resolved_variables: Dict[Variable, Resolvable], resolved_variables: Dict[Variable, Resolvable],
custom_functions: Dict[str, "VariableDependency"], custom_functions: Dict[str, VariableDependency],
) -> Resolvable: ) -> Resolvable:
resolved: List[Resolvable] = [] resolved: List[Resolvable] = []
for token in self.ast: for token in self.ast:
@ -43,73 +38,3 @@ class SyntaxTree(VariableDependency):
# Otherwise, to concat multiple resolved outputs, we must concat as strings # Otherwise, to concat multiple resolved outputs, we must concat as strings
return String("".join([str(res) for res in resolved])) return String("".join([str(res) for res in resolved]))
@classmethod
def _traverse_custom_function_dependencies(
cls,
custom_function_name: str,
custom_function_dependency: "SyntaxTree",
custom_functions: Dict[str, "SyntaxTree"],
deps: List[str],
) -> None:
for dep in custom_function_dependency.custom_functions:
if custom_function_name in deps + [dep.name]:
cycle_deps = [custom_function_name] + deps + [dep.name]
cycle_deps_str = " -> ".join([f"%{name}" for name in cycle_deps])
raise CycleDetected(f"Custom functions contain a cycle: {cycle_deps_str}")
cls._traverse_custom_function_dependencies(
custom_function_name=custom_function_name,
custom_function_dependency=custom_functions[dep.name],
custom_functions=custom_functions,
deps=deps + [dep.name],
)
@classmethod
def _ensure_no_custom_function_cycles(cls, custom_functions: Dict[str, "SyntaxTree"]):
for custom_function_name, custom_function in custom_functions.items():
cls._traverse_custom_function_dependencies(
custom_function_name=custom_function_name,
custom_function_dependency=custom_function,
custom_functions=custom_functions,
deps=[],
)
@classmethod
def resolve_overrides(
cls,
parsed_overrides: Dict[str, "SyntaxTree"],
custom_functions: Dict[str, "SyntaxTree"],
pre_resolved_variables: Optional[Dict[Variable, Resolvable]],
) -> Dict[str, Resolvable]:
overrides: Dict[Variable, "SyntaxTree"] = {
Variable(name): ast for name, ast in parsed_overrides.items()
}
unresolved_variables: List[Variable] = list(overrides.keys())
resolved_variables: Dict[Variable, Resolvable] = (
pre_resolved_variables if pre_resolved_variables else {}
)
cls._ensure_no_custom_function_cycles(custom_functions=custom_functions)
while unresolved_variables:
unresolved_count: int = len(unresolved_variables)
for variable in unresolved_variables:
if not overrides[variable].has_variable_dependency(
resolved_variables=resolved_variables
):
resolved_variables[variable] = overrides[variable].resolve(
resolved_variables=resolved_variables,
custom_functions=custom_functions,
)
unresolved_variables.remove(variable)
if len(unresolved_variables) == unresolved_count:
raise StringFormattingException(
f"Cycle detected within these variables: "
f"{', '.join(sorted([var.name for var in unresolved_variables]))}"
)
return {variable.name: resolvable for variable, resolvable in resolved_variables.items()}