1#!/usr/bin/env python3
2"""
3Provdes the Main ArgParser_p Protocol,
4and the ParseMachineBase StateMachine.
5
6ParseMachineBase descibes the state progression to parse arguments,
7while jgdv.cli.arg_parser.CLIParser adds the specific logic to states and transitions
8"""
9# Imports:
10from __future__ import annotations
11
12# ##-- stdlib imports
13import datetime
14import enum
15import functools as ftz
16import itertools as itz
17import logging as logmod
18import pathlib as pl
19import re
20import time
21import types
22import weakref
23from collections import ChainMap
24from uuid import UUID, uuid1
25
26# ##-- end stdlib imports
27
28# ##-- 3rd party imports
29from statemachine import State, StateMachine, Event
30from statemachine.states import States
31from statemachine.exceptions import TransitionNotAllowed
32
33# ##-- end 3rd party imports
34
35# ##-- 1st party imports
36from jgdv import Maybe
37from jgdv.structs.chainguard import ChainGuard
38
39# ##-- end 1st party imports
40
41from . import _interface as API # noqa: N812
42from . import errors
43
44# ##-- types
45# isort: off
46import abc
47import collections.abc
48from typing import TYPE_CHECKING, cast, assert_type, assert_never
49from typing import Generic, NewType
50# Protocols:
51from typing import Protocol, runtime_checkable
52# Typing Decorators:
53from typing import no_type_check, final, override, overload
54
55if TYPE_CHECKING:
56 from ._interface import ParamSource_p, ArgParserModel_p
57 from jgdv import Maybe
58 from typing import Final
59 from typing import ClassVar, Any, LiteralString
60 from typing import Never, Self, Literal
61 from typing import TypeGuard
62 from collections.abc import Iterable, Iterator, Callable, Generator
63 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
64
65##--|
66# isort: on
67# ##-- end types
68
69logging = logmod.getLogger(__name__)
70
71MAX_STAGES : Final[int] = 200
72##--|
[docs]
73class ParseMachine(StateMachine):
74 """
75 Implemented Parse State Machine
76
77 __call__ with:
78 args : list[str] -- the cli args to parse (ie: from sys.argv)
79 prog : list[ParamSpec_i] -- specs of the top level program
80 cmds : list[ParamSource_p] -- commands that can provide their parameters
81 subs : dict[str, list[ParamSource_p]] -- a mapping from commands -> subcommands that can provide parameters
82
83 A cli call will be of the form:
84 {proghead} {prog [kw]args} {cmd} {cmd[kw]args}* [{subs} {subs[kw]args} [-- {subs} {subargs}]* ]? (--help)?
85
86 eg:
87 doot -v list -by-group a b c --help
88 doot run basic::task -quick --value=2 --help
89
90 Will raise a jgdv.cli.errors.ParseError on failure
91 """
92
93 ## States
94 # startup states
95 Start = State(initial=True)
96 Prepare = State(enter="prepare_for_parse")
97 # Parsing states
98 Help = State(enter="set_force_help")
99 Head = State()
100 Section = State(enter="initialise_section")
101 Prog = State(enter="select_prog_spec")
102 Cmd = State(enter="select_cmd_spec")
103 Sub = State(enter="select_sub_spec")
104 Kwargs = State(enter="parse_kwarg")
105 Posargs = State(enter="parse_posarg")
106 Separator = State(enter="parse_separator")
107 Section_end = State(enter="clear_section")
108 # teardown states
109 Cleanup = State(enter="cleanup")
110 Report = State(enter="report")
111 End = State(final=True)
112
113 # Event Transitions
114 setup = (
115 Start.to(Prepare)
116 | Prepare.to(End, cond="not _has_more_args")
117 | Prepare.to(End, cond="_has_no_specs")
118 | Prepare.to(Help, cond="_has_help_flag_at_tail")
119 | Head.from_(Prepare, Help)
120 )
121
122 parse = (
123 # program / cmd / subs
124 Head.to(Prog, cond="_prog_at_front")
125 | Head.to(Cmd, cond="_cmd_at_front")
126 | Head.to(Cmd, cond="_no_cmd", on="_insert_implicit_cmd")
127 | Head.to(Sub, cond="_sub_at_front")
128 | Head.to(Sub, cond="_no_sub", on="_insert_implicit_sub")
129 | Head.to(Report)
130 | Section.from_(Prog, Cmd, Sub)
131 # args
132 | Kwargs.from_(Section, Kwargs, cond="_kwarg_at_front")
133 | Posargs.from_(Section, Kwargs, Posargs, cond="_posarg_at_front")
134 | Separator.from_(Section, Kwargs, Posargs, cond="_separator_at_front")
135 # separator
136
137 # loop
138 | Section_end.from_(Section, Kwargs, Posargs, Separator)
139 | Section_end.to(Report, cond="not _has_more_args")
140 | Section_end.to(Head)
141 )
142
143 finish = (
144 Report.to(Cleanup)
145 | Cleanup.to(End)
146 )
147
148 ##--| composite
149 progress = (setup | parse | finish)
150
151 def __init__(self, parser:Maybe[ArgParserModel_p]=None, max:int=MAX_STAGES) -> None: # noqa: A002
152 match parser:
153 case API.ArgParserModel_p():
154 pass
155 case x:
156 raise TypeError(type(x))
157 super().__init__(parser)
158 self.count = 0
159 self.max_attempts = max
160
161 def __call__(self, args:list[str], *, prog:ParamSource_p, cmds:list[ParamSource_p], subs:list[tuple[tuple[str, ...], ParamSource_p]], implicits:Maybe[dict[str,list[str]]]=None) -> Maybe[dict]:
162 assert(self.current_state == self.Start) # type: ignore[has-type]
163 while self.current_state != self.End:
164 self.progress(prog=prog, cmds=cmds, subs=subs, raw_args=args, implicits=implicits)
165 else:
166 return self.model._report
167
[docs]
168 def on_enter_state(self, source:State, event:Event, target:State) -> None:
169 logging.debug("Parse(%s, R:%s/%s): %s -> %s -> %s",
170 getattr(self, "count", 0),
171 len(self.model.args_remaining), len(self.model.args_initial),
172 source, event, target)
173
[docs]
174 def on_exit_state(self) -> None:
175 self.count += 1
176 if self.max_attempts < self.count:
177 logging.debug("Max Parse Stages Occurred")
178 raise StopIteration