Skip to content

Commit

Permalink
Some cleanup in Parser
Browse files Browse the repository at this point in the history
  • Loading branch information
@sezanzeb
sezanzeb committed Jan 2, 2025
1 parent c9de7ef commit 4016f89
Show file tree
Hide file tree
Showing 2 changed files with 135 additions and 110 deletions.
12 changes: 10 additions & 2 deletions inputremapper/injection/macros/argument.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -304,9 +304,17 @@ def _is_numeric_string(self, value: str) -> bool:
return False

def _type_error_factory(self, value: Any) -> MacroError:
formatted_types: List[str] = []

for type_ in self.types:
if type_ is None:
formatted_types.append("None")
else:
formatted_types.append(type_.__name__)

return MacroError(
msg=(
f'Expected "{self.name}" to be one of {self.types}, but got '
f"{type(value)} {value}"
f'Expected "{self.name}" to be one of {formatted_types}, but got '
f'{type(value).__name__} "{value}"'
)
)
233 changes: 125 additions & 108 deletions inputremapper/injection/macros/parse.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@ def _split_keyword_arg(param):
return None, param

@staticmethod
def check_for_unknown_keyword_arguments(
def _validate_keyword_argument_names(
keyword_args: Dict[str, Any],
task_class: Type[Task],
) -> None:
Expand Down Expand Up @@ -231,121 +231,138 @@ def debug(*args, **kwargs):
code = code.strip()

# is it another macro?
call_match = re.match(r"^(\w+)\(", code)
call = call_match[1] if call_match else None
if call is not None:
if macro_instance is None:
# start a new chain
macro_instance = Macro(code, context, mapping)
task_call_match = re.match(r"^(\w+)\(", code)
task_name = task_call_match[1] if task_call_match else None

if task_name is None:
# It is probably either a key name like KEY_A or a variable name as in `set(var,1)`,
# both won't contain special characters that can break macro syntax so they don't
# have to be wrapped in quotes. The argument configuration of the tasks will
# detemrine how to parse it.
debug("%svalue %s", space, code)
return RawValue(value=code)

if macro_instance is None:
# start a new chain
macro_instance = Macro(code, context, mapping)
else:
# chain this call to the existing instance
assert isinstance(macro_instance, Macro)

task_class = Parser.TASK_CLASSES.get(task_name)
if task_class is None:
raise MacroError(code, f"Unknown function {task_name}")

# get all the stuff inbetween
closing_bracket_position = Parser._count_brackets(code) - 1
inner = code[code.index("(") + 1 : closing_bracket_position]
debug("%scalls %s with %s", space, task_name, inner)

# split "3, foo=a(2, k(a).w(10))" into arguments
raw_string_args = Parser._extract_args(inner)

# parse and sort the params
positional_args: List[RawValue] = []
keyword_args: Dict[str, RawValue] = {}
for param in raw_string_args:
key, value = Parser._split_keyword_arg(param)
parsed = Parser._parse_recurse(
value.strip(),
context,
mapping,
verbose,
None,
depth + 1,
)
if key is None:
if len(keyword_args) > 0:
msg = f'Positional argument "{key}" follows keyword argument'
raise MacroError(code, msg)
positional_args.append(parsed)
else:
# chain this call to the existing instance
assert isinstance(macro_instance, Macro)

task_class = Parser.TASK_CLASSES.get(call)
if task_class is None:
raise MacroError(code, f"Unknown function {call}")

# get all the stuff inbetween
closing_bracket_position = Parser._count_brackets(code) - 1
inner = code[code.index("(") + 1 : closing_bracket_position]
debug("%scalls %s with %s", space, call, inner)

# split "3, foo=a(2, k(a).w(10))" into arguments
raw_string_args = Parser._extract_args(inner)

# parse and sort the params
positional_args: List[RawValue] = []
keyword_args: Dict[str, RawValue] = {}
for param in raw_string_args:
key, value = Parser._split_keyword_arg(param)
parsed = Parser._parse_recurse(
value.strip(),
if key in keyword_args:
raise MacroError(code, f'The "{key}" argument was specified twice')
keyword_args[key] = parsed

debug(
"%sadd call to %s with %s, %s",
space,
task_name,
positional_args,
keyword_args,
)

Parser._validate_keyword_argument_names(
keyword_args,
task_class,
)
Parser._validate_num_args(
code,
task_name,
task_class,
raw_string_args,
)

try:
task = task_class(
positional_args,
keyword_args,
context,
mapping,
)
macro_instance.add_task(task)
except TypeError as exception:
raise MacroError(msg=str(exception)) from exception

# is after this another call? Chain it to the macro_instance
more_code_exists = len(code) > closing_bracket_position + 1
if more_code_exists:
next_char = code[closing_bracket_position + 1]
statement_closed = next_char == "."

if statement_closed:
# skip over the ")."
chain = code[closing_bracket_position + 2 :]
debug("%sfollowed by %s", space, chain)
Parser._parse_recurse(
chain,
context,
mapping,
verbose,
None,
depth + 1,
macro_instance,
depth,
)
elif re.match(r"[a-zA-Z_]", next_char):
# something like foo()bar
raise MacroError(
code,
f'Expected a "." to follow after '
f"{code[:closing_bracket_position + 1]}",
)
if key is None:
if len(keyword_args) > 0:
msg = f'Positional argument "{key}" follows keyword argument'
raise MacroError(code, msg)
positional_args.append(parsed)
else:
if key in keyword_args:
raise MacroError(
code, f'The "{key}" argument was specified twice'
)
keyword_args[key] = parsed

Parser.check_for_unknown_keyword_arguments(keyword_args, task_class)

debug(
"%sadd call to %s with %s, %s",
space,
call,
positional_args,
keyword_args,
)

min_args, max_args = task_class.get_num_parameters()
num_provided_args = len(raw_string_args)
if num_provided_args < min_args or num_provided_args > max_args:
if min_args != max_args:
msg = (
f"{call} takes between {min_args} and {max_args}, "
f"not {num_provided_args} parameters"
)
else:
msg = f"{call} takes {min_args}, not {num_provided_args} parameters"

raise MacroError(code, msg)
return RawValue(value=macro_instance)

try:
task = task_class(
positional_args,
keyword_args,
context,
mapping,
@staticmethod
def _validate_num_args(
code: str,
task_name: str,
task_class: Type[Task],
raw_string_args: List[str],
) -> None:
min_args, max_args = task_class.get_num_parameters()
num_provided_args = len(raw_string_args)
if num_provided_args < min_args or num_provided_args > max_args:
if min_args != max_args:
msg = (
f"{task_name} takes between {min_args} and {max_args}, "
f"not {num_provided_args} parameters"
)
macro_instance.add_task(task)
except TypeError as exception:
raise MacroError(msg=str(exception)) from exception

# is after this another call? Chain it to the macro_instance
more_code_exists = len(code) > closing_bracket_position + 1
if more_code_exists:
next_char = code[closing_bracket_position + 1]
statement_closed = next_char == "."

if statement_closed:
# skip over the ")."
chain = code[closing_bracket_position + 2 :]
debug("%sfollowed by %s", space, chain)
Parser._parse_recurse(
chain,
context,
mapping,
verbose,
macro_instance,
depth,
)
elif re.match(r"[a-zA-Z_]", next_char):
# something like foo()bar
raise MacroError(
code,
f'Expected a "." to follow after '
f"{code[:closing_bracket_position + 1]}",
)

return RawValue(value=macro_instance)

# It is probably either a key name like KEY_A or a variable name as in `set(var,1)`,
# both won't contain special characters that can break macro syntax so they don't
# have to be wrapped in quotes. The argument configuration of the tasks will
# detemrine how to parse it.
debug("%svalue %s", space, code)
return RawValue(value=code)
else:
msg = (
f"{task_name} takes {min_args}, not {num_provided_args} parameters"
)

raise MacroError(code, msg)

@staticmethod
def handle_plus_syntax(macro):
Expand Down

0 comments on commit 4016f89

Please sign in to comment.