exploration.analysis

  1"""
  2- Authors: Peter Mawhorter
  3- Consulted:
  4- Date: 2022-10-24
  5- Purpose: Analysis functions for decision graphs an explorations.
  6"""
  7
  8from typing import (
  9    List, Dict, Tuple, Optional, TypeVar, Callable, Union, Any,
 10    ParamSpec, Concatenate, Set, cast
 11)
 12
 13from . import base, core, parsing
 14
 15import textwrap
 16
 17
 18#-------------------#
 19# Text descriptions #
 20#-------------------#
 21
 22def describeConsequence(consequence: base.Consequence) -> str:
 23    """
 24    Returns a string which concisely describes a consequence list.
 25    Returns an empty string if given an empty consequence. Examples:
 26
 27    >>> describeConsequence([])
 28    ''
 29    >>> describeConsequence([
 30    ...     base.effect(gain=('gold', 5), delay=2, charges=3),
 31    ...     base.effect(lose='flight')
 32    ... ])
 33    'gain gold*5 ,2 =3; lose flight'
 34    >>> from . import commands
 35    >>> d = describeConsequence([
 36    ...     base.effect(edit=[
 37    ...         [
 38    ...             commands.command('val', '5'),
 39    ...             commands.command('empty', 'list'),
 40    ...             commands.command('append')
 41    ...         ],
 42    ...         [
 43    ...             commands.command('val', '11'),
 44    ...             commands.command('assign', 'var'),
 45    ...             commands.command('op', '+', '$var', '$var')
 46    ...         ],
 47    ...     ])
 48    ... ])
 49    >>> d
 50    'with consequences:\
 51\\n    edit {\
 52\\n      val 5;\
 53\\n      empty list;\
 54\\n      append $_;\
 55\\n    } {\
 56\\n      val 11;\
 57\\n      assign var $_;\
 58\\n      op + $var $var;\
 59\\n    }\
 60\\n'
 61    >>> for line in d.splitlines():
 62    ...     print(line)
 63    with consequences:
 64        edit {
 65          val 5;
 66          empty list;
 67          append $_;
 68        } {
 69          val 11;
 70          assign var $_;
 71          op + $var $var;
 72        }
 73    """
 74    edesc = ''
 75    pf = parsing.ParseFormat()
 76    if consequence:
 77        parts = []
 78        for item in consequence:
 79            # TODO: Challenges and Conditions here!
 80            if 'skills' in item:  # a Challenge
 81                item = cast(base.Challenge, item)
 82                parts.append(pf.unparseChallenge(item))
 83            elif 'value' in item:  # an Effect
 84                item = cast(base.Effect, item)
 85                parts.append(pf.unparseEffect(item))
 86            elif 'condition' in item:  # a Condition
 87                item = cast(base.Condition, item)
 88                parts.append(pf.unparseCondition(item))
 89            else:
 90                raise TypeError(
 91                    f"Invalid consequence item (no 'skills', 'value', or"
 92                    f" 'condition' key found):\n{repr(item)}"
 93                )
 94        edesc = '; '.join(parts)
 95        if len(edesc) > 60 or '\n' in edesc:
 96            edesc = 'with consequences:\n' + ';\n'.join(
 97                textwrap.indent(part, '    ')
 98                for part in parts
 99            ) + '\n'
100
101    return edesc
102
103
104def describeProgress(exploration: core.DiscreteExploration) -> str:
105    """
106    Describes the progress of an exploration by noting each room/zone
107    visited and explaining the options visible at each point plus which
108    option was taken. Notes powers/tokens gained/lost along the way.
109    Returns a string.
110
111    Example:
112    >>> from exploration import journal
113    >>> e = journal.convertJournal('''\\
114    ... S Start::pit
115    ... A gain jump
116    ... A gain attack
117    ... n button check
118    ... zz Wilds
119    ... o up
120    ...   q _flight
121    ... o left
122    ... x left left_nook right
123    ... a geo_rock
124    ...   At gain geo*15
125    ...   At deactivate
126    ... o up
127    ...   q _tall_narrow
128    ... t right
129    ... o right
130    ...   q attack
131    ... ''')
132    >>> for line in describeProgress(e).splitlines():
133    ...    print(line)
134    Start of the exploration
135    Start exploring domain main at 0 (Start::pit)
136      Gained capability 'attack'
137      Gained capability 'jump'
138    At decision 0 (Start::pit)
139      In zone Start
140      In region Wilds
141      There are transitions:
142        left to unconfirmed
143        up to unconfirmed; requires _flight
144      1 note(s) at this step
145    Explore left from decision 0 (Start::pit) to 2 (now Start::left_nook)
146    At decision 2 (Start::left_nook)
147      There are transitions:
148        right to 0 (Start::pit)
149      There are actions:
150        geo_rock
151    Do action geo_rock
152      Gained 15 geo(s)
153    Take right from decision 2 (Start::left_nook) to 0 (Start::pit)
154    At decision 0 (Start::pit)
155      There are transitions:
156        left to 2 (Start::left_nook)
157        right to unconfirmed; requires attack
158        up to unconfirmed; requires _flight
159    Waiting for another action...
160    End of the exploration.
161    """
162    result = ''
163
164    regions: Set[base.Zone] = set()
165    zones: Set[base.Zone] = set()
166    last: Union[base.DecisionID, Set[base.DecisionID], None] = None
167    lastState: base.State = base.emptyState()
168    prevCapabilities = base.effectiveCapabilitySet(lastState)
169    prevMechanisms = lastState['mechanisms']
170    oldActiveDecisions: Set[base.DecisionID] = set()
171    for i, situation in enumerate(exploration):
172        if i == 0:
173            result += "Start of the exploration\n"
174
175        # Extract info
176        graph = situation.graph
177        activeDecisions = exploration.getActiveDecisions(i)
178        newActive = activeDecisions - oldActiveDecisions
179        departedFrom = exploration.movementAtStep(i)[0]
180        # TODO: use the other parts of this?
181        nowZones: Set[base.Zone] = set()
182        for active in activeDecisions:
183            nowZones |= graph.zoneAncestors(active)
184        regionsHere = set(
185            z
186            for z in nowZones
187            if graph.zoneHierarchyLevel(z) == 1
188        )
189        zonesHere = set(
190            z
191            for z in nowZones
192            if graph.zoneHierarchyLevel(z) == 0
193        )
194        here = departedFrom
195        state = situation.state
196        capabilities = base.effectiveCapabilitySet(state)
197        mechanisms = state['mechanisms']
198
199        # Describe capabilities gained/lost relative to previous step
200        # (i.e., as a result of the previous action)
201        gained = (
202            capabilities['capabilities']
203          - prevCapabilities['capabilities']
204        )
205        gainedTokens = []
206        for tokenType in capabilities['tokens']:
207            net = (
208                capabilities['tokens'][tokenType]
209              - prevCapabilities['tokens'].get(tokenType, 0)
210            )
211            if net != 0:
212                gainedTokens.append((tokenType, net))
213        changed = [
214            mID
215            for mID in list(mechanisms.keys()) + list(prevMechanisms.keys())
216            if mechanisms.get(mID) != prevMechanisms.get(mID)
217        ]
218
219        for capability in sorted(gained):
220            result += f"  Gained capability '{capability}'\n"
221
222        for tokenType, net in gainedTokens:
223            if net > 0:
224                result += f"  Gained {net} {tokenType}(s)\n"
225            else:
226                result += f"  Lost {-net} {tokenType}(s)\n"
227
228        for mID in changed:
229            oldState = prevMechanisms.get(mID, base.DEFAULT_MECHANISM_STATE)
230            newState = mechanisms.get(mID, base.DEFAULT_MECHANISM_STATE)
231
232            details = graph.mechanismDetails(mID)
233            if details is None:
234                mName = "(unknown)"
235            else:
236                mName = details[1]
237            result += (
238                f"  Set mechanism {mID} ({mName}) to {newState} (was"
239                f" {oldState})"
240            )
241            # TODO: Test this!
242
243        if isinstance(departedFrom, base.DecisionID):
244            # Print location info
245            if here != last:
246                if here is None:
247                    result += "Without a position...\n"
248                elif isinstance(here, set):
249                    result += f"With {len(here)} active decisions\n"
250                    # TODO: List them using namesListing?
251                else:
252                    result += f"At decision {graph.identityOf(here)}\n"
253            newZones = zonesHere - zones
254            for zone in sorted(newZones):
255                result += f"  In zone {zone}\n"
256            newRegions = regionsHere - regions
257            for region in sorted(newRegions):
258                result += f"  In region {region}\n"
259
260        elif isinstance(departedFrom, set):  # active in spreading domain
261            spreadingDomain = graph.domainFor(list(departedFrom)[0])
262            result += (
263                f"  In domain {spreadingDomain} with {len(departedFrom)}"
264                f" active decisions...\n"
265            )
266
267        else:
268            assert departedFrom is None
269
270        # Describe new position/positions at start of this step
271        if len(newActive) > 1:
272            newListing = ', '.join(
273                sorted(graph.identityOf(n) for n in newActive)
274            )
275            result += (
276                f"  There are {len(newActive)} new active decisions:"
277                f"\n  {newListing}"
278            )
279
280        elif len(newActive) == 1:
281            here = list(newActive)[0]
282
283            outgoing = graph.destinationsFrom(here)
284
285            transitions = {t: d for (t, d) in outgoing.items() if d != here}
286            actions = {t: d for (t, d) in outgoing.items() if d == here}
287            if transitions:
288                result += "  There are transitions:\n"
289                for transition in sorted(transitions):
290                    dest = transitions[transition]
291                    if not graph.isConfirmed(dest):
292                        destSpec = 'unconfirmed'
293                    else:
294                        destSpec = graph.identityOf(dest)
295                    req = graph.getTransitionRequirement(here, transition)
296                    rDesc = ''
297                    if req != base.ReqNothing():
298                        rDesc = f"; requires {req.unparse()}"
299                    cDesc = describeConsequence(
300                        graph.getConsequence(here, transition)
301                    )
302                    if cDesc:
303                        cDesc = '; ' + cDesc
304                    result += (
305                        f"    {transition} to {destSpec}{rDesc}{cDesc}\n"
306                    )
307
308            if actions:
309                result += "  There are actions:\n"
310                for action in sorted(actions):
311                    req = graph.getTransitionRequirement(here, action)
312                    rDesc = ''
313                    if req != base.ReqNothing():
314                        rDesc = f"; requires {req.unparse()}"
315                    cDesc = describeConsequence(
316                        graph.getConsequence(here, action)
317                    )
318                    if cDesc:
319                        cDesc = '; ' + cDesc
320                    if rDesc or cDesc:
321                        desc = (rDesc + cDesc)[2:]  # chop '; ' from either
322                        result += f"    {action} ({desc})\n"
323                    else:
324                        result += f"    {action}\n"
325
326        # note annotations
327        if len(situation.annotations) > 0:
328            result += (
329                f"  {len(situation.annotations)} note(s) at this step\n"
330            )
331
332        # Describe action taken
333        if situation.action is None and situation.type == "pending":
334            result += "Waiting for another action...\n"
335        else:
336            desc = base.describeExplorationAction(situation, situation.action)
337            desc = desc[0].capitalize() + desc[1:]
338            result += desc + '\n'
339
340        if i == len(exploration) - 1:
341            result += "End of the exploration.\n"
342
343        # Update state variables
344        oldActiveDecisions = activeDecisions
345        prevCapabilities = capabilities
346        prevMechanisms = mechanisms
347        regions = regionsHere
348        zones = zonesHere
349        if here is not None:
350            last = here
351        lastState = state
352
353    return result
354
355
356#--------------------#
357# Analysis functions #
358#--------------------#
359
360def lastIdentity(
361    exploration: core.DiscreteExploration,
362    decision: base.DecisionID
363):
364    """
365    Returns the `identityOf` result for the specified decision in the
366    last step in which that decision existed.
367    """
368    for i in range(-1, -len(exploration) - 1, -1):
369        situation = exploration.getSituation(i)
370        try:
371            return situation.graph.identityOf(decision)
372        except core.MissingDecisionError:
373            pass
374    raise core.MissingDecisionError(
375        f"Decision {decision!r} never existed."
376    )
377
378
379def unexploredBranches(
380    graph: core.DecisionGraph,
381    context: Optional[base.RequirementContext] = None
382) -> List[Tuple[base.DecisionID, base.Transition]]:
383    """
384    Returns a list of from-decision, transition-at-that-decision pairs
385    which each identify an unexplored branch in the given graph.
386
387    When a `context` is provided it only counts options whose
388    requirements are satisfied in that `RequirementContext`, and the
389    'searchFrom' part of the context will be replaced by both ends of
390    each transition tested. This doesn't perfectly map onto actually
391    reachability since nodes between where the player is and where the
392    option is might force changes in the game state that make it
393    un-takeable.
394
395    TODO: add logic to detect trivially-unblocked edges?
396    """
397    result = []
398    # TODO: Fix networkx type stubs for MultiDiGraph!
399    for (src, dst, transition) in graph.edges(keys=True):  # type:ignore
400        req = graph.getTransitionRequirement(src, transition)
401        localContext: Optional[base.RequirementContext] = None
402        if context is not None:
403            localContext = base.RequirementContext(
404                state=context.state,
405                graph=context.graph,
406                searchFrom=graph.bothEnds(src, transition)
407            )
408        # Check if this edge goes from a confirmed to an unconfirmed node
409        if (
410            graph.isConfirmed(src)
411        and not graph.isConfirmed(dst)
412        and (localContext is None or req.satisfied(localContext))
413        ):
414            result.append((src, transition))
415    return result
416
417
418def currentDecision(situation: base.Situation) -> str:
419    """
420    Returns the `identityOf` string for the current decision in a given
421    situation.
422    """
423    return situation.graph.identityOf(situation.state['primaryDecision'])
424
425
426def countAllUnexploredBranches(situation: base.Situation) -> int:
427    """
428    Counts the number of unexplored branches in the given situation's
429    graph, regardless of traversibility (see `unexploredBranches`).
430    """
431    return len(unexploredBranches(situation.graph))
432
433
434def countTraversableUnexploredBranches(situation: base.Situation) -> int:
435    """
436    Counts the number of traversable unexplored branches (see
437    `unexploredBranches`) in a given situation, using the situation's
438    game state to determine which branches are traversable or not
439    (although this isn't strictly perfect TODO: Fix that).
440    """
441    context = base.genericContextForSituation(
442        situation,
443        base.combinedDecisionSet(situation.state)
444    )
445    return len(unexploredBranches(situation.graph, context))
446
447
448def countActionsAtDecision(
449    graph: core.DecisionGraph,
450    decision: base.DecisionID
451) -> Optional[int]:
452    """
453    Given a graph and a particular decision within that graph, returns
454    the number of actions available at that decision. Returns None if the
455    specified decision does not exist.
456    """
457    if decision not in graph:
458        return None
459    return len(graph.decisionActions(decision))
460
461
462def countBranches(
463    graph: core.DecisionGraph,
464    decision: base.DecisionID
465) -> Optional[int]:
466    """
467    Computes the number of branches at a particular decision, not
468    counting actions. Returns `None` for unvisited and nonexistent
469    decisions so that they aren't counted as part of averages.
470    """
471    if decision not in graph or not graph.isConfirmed(decision):
472        return None
473
474    dests = graph.destinationsFrom(decision)
475    branches = 0
476    for transition, dest in dests.items():
477        if dest != decision:
478            branches += 1
479
480    return branches
481
482
483def countRevisits(
484    exploration: core.DiscreteExploration,
485    decision: base.DecisionID
486) -> int:
487    """
488    Given an `DiscreteExploration` object and a particular `Decision`
489    which exists at some point during that exploration, counts the number
490    of times that decision was activated after its initial discovery (not
491    counting steps where we remain in it due to a wait or action).
492
493    Returns 0 even for decisions that aren't part of the exploration.
494    """
495    result = 0
496    wasActive = False
497    for i in range(len(exploration)):
498        active = exploration.getActiveDecisions(i)
499        if decision in active:
500            if not wasActive:
501                result += 1
502            wasActive = True
503        else:
504            wasActive = False
505
506    # Make sure not to return -1 for decisions that were never visited
507    if result >= 1:
508        return result - 1
509    else:
510        return 0
511
512
513#-----------------------#
514# Generalizer Functions #
515#-----------------------#
516
517# Some type variables to make type annotations work
518T = TypeVar('T')
519P = ParamSpec('P')
520
521
522def analyzeGraph(
523    routine: Callable[Concatenate[core.DecisionGraph, P], T]
524) -> Callable[Concatenate[base.Situation, P], T]:
525    """
526    Wraps a `DecisionGraph` analysis routine (possibly with extra
527    arguments), returning a function which applies that analysis to a
528    `Situation`.
529    """
530    def analyzesGraph(
531        situation: base.Situation,
532        *args: P.args,
533        **kwargs: P.kwargs
534    ) -> T:
535        "Created by `analyzeGraph`."
536        return routine(situation.graph, *args, **kwargs)
537
538    analyzesGraph.__name__ = routine.__name__ + "InSituation"
539    analyzesGraph.__doc__ = f"""
540    Application of a graph analysis routine to a situation.
541
542    The analysis routine applied is: {routine.__name__}
543    """ + (routine.__doc__ or '')
544    return analyzesGraph
545
546
547def perDecision(
548    routine: Callable[[base.Situation, base.DecisionID], T]
549) -> Callable[[base.Situation], Dict[base.DecisionID, T]]:
550    """
551    Returns a wrapped function that applies the given
552    individual-decision analysis routine to each decision in a
553    situation, returning a dictionary mapping decisions to results.
554    """
555    def appliedPerDecision(
556        situation: base.Situation,
557    ) -> Dict[base.DecisionID, T]:
558        'Created by `perDecision`.'
559        result = {}
560        for decision in situation.graph:
561            result[decision] = routine(situation, decision)
562        return result
563    appliedPerDecision.__name__ = routine.__name__ + "PerDecision"
564    appliedPerDecision.__doc__ = f"""
565    Application of an analysis routine to each decision in a situation,
566    returning a dictionary mapping decisions to results. The analysis
567    routine applied is: {routine.__name__}
568    """ + (routine.__doc__ or '')
569    return appliedPerDecision
570
571
572def perExplorationDecision(
573    routine: Callable[[core.DiscreteExploration, base.DecisionID], T],
574    mode: str = "all"
575) -> Callable[[core.DiscreteExploration], Dict[base.DecisionID, T]]:
576    """
577    Returns a wrapped function that applies the given
578    decision-in-exploration analysis routine to each decision in an
579    exploration, returning a dictionary mapping decisions to results.
580
581    The `mode` argument controls what we mean by "each decision:" use
582    "all" to apply it to all decisions which ever existed, "known" to
583    apply it to all decisions which were known at any point, "visited"
584    to apply it to all visited decisions, and "final" to apply it to
585    each decision in the final decision graph.
586    """
587    def appliedPerDecision(
588        exploration: core.DiscreteExploration,
589    ) -> Dict[base.DecisionID, T]:
590        'Created by `perExplorationDecision`.'
591        result = {}
592        now = exploration.getSituation()
593        graph = now.graph
594        if mode == "all":
595            applyTo = exploration.allDecisions()
596        elif mode == "known":
597            applyTo = exploration.allExploredDecisions()
598        elif mode == "visited":
599            applyTo = exploration.allVisitedDecisions()
600        elif mode == "final":
601            applyTo = list(graph)
602
603        for decision in applyTo:
604            result[decision] = routine(exploration, decision)
605
606        return result
607
608    appliedPerDecision.__name__ = routine.__name__ + "PerExplorationDecision"
609    desc = mode + ' '
610    if desc == "all ":
611        desc = ''
612    appliedPerDecision.__doc__ = f"""
613    Application of an analysis routine to each {desc}decision in an
614    exploration, returning a dictionary mapping decisions to results. The
615    analysis routine applied is: {routine.__name__}
616    """ + (routine.__doc__ or '')
617    return appliedPerDecision
618
619
620Base = TypeVar('Base', base.Situation, core.DiscreteExploration)
621"Either a situation or an exploration."
622
623
624def sumOfResults(
625    routine: Callable[
626        [Base],
627        Dict[Any, Union[int, float, complex, None]]
628    ]
629) -> Callable[[Base], Union[int, float, complex]]:
630    """
631    Given an analysis routine that applies to either a situation or an
632    exploration and which returns a dictionary mapping some analysis
633    units to individual numerical results, returns a new analysis
634    routine which applies to the same input and which returns a single
635    number that's the sum of the individual results, ignoring `None`s.
636    Returns 0 if there are no results.
637    """
638    def sumResults(base: Base) -> Union[int, float, complex]:
639        "Created by sumOfResults"
640        results = routine(base)
641        return sum(v for v in results.values() if v is not None)
642
643    sumResults.__name__ = routine.__name__ + "Sum"
644    sumResults.__doc__ = f"""
645    Sum of analysis results over analysis units.
646    The analysis routine applied is: {routine.__name__}
647    """ + (routine.__doc__ or '')
648    return sumResults
649
650
651def meanOfResults(
652    routine: Callable[
653        [Base],
654        Dict[Any, Union[int, float, complex, None]]
655    ]
656) -> Callable[[Base], Union[int, float, complex, None]]:
657    """
658    Works like `sumOfResults` but returns a function which gives the
659    mean, not the sum. The function will return `None` if there are no
660    results.
661    """
662    def meanResult(base: Base) -> Union[int, float, complex, None]:
663        "Created by meanOfResults"
664        results = routine(base)
665        nums = [v for v in results.values() if v is not None]
666        if len(nums) == 0:
667            return None
668        else:
669            return sum(nums) / len(nums)
670
671    meanResult.__name__ = routine.__name__ + "Mean"
672    meanResult.__doc__ = f"""
673    Mean of analysis results over analysis units.
674    The analysis routine applied is: {routine.__name__}
675    """ + (routine.__doc__ or '')
676    return meanResult
677
678
679def medianOfResults(
680    routine: Callable[
681        [Base],
682        Dict[Any, Union[int, float, None]]
683    ]
684) -> Callable[[Base], Union[int, float, None]]:
685    """
686    Works like `sumOfResults` but returns a function which gives the
687    median, not the sum. The function will return `None` if there are no
688    results.
689    """
690    def medianResult(base: Base) -> Union[int, float, None]:
691        "Created by medianOfResults"
692        results = routine(base)
693        nums = sorted(v for v in results.values() if v is not None)
694        half = len(nums) // 2
695        if len(nums) == 0:
696            return None
697        elif len(nums) % 2 == 0:
698            return (nums[half] + nums[half + 1]) / 2
699        else:
700            return nums[half]
701
702    medianResult.__name__ = routine.__name__ + "Mean"
703    medianResult.__doc__ = f"""
704    Mean of analysis results over analysis units.
705    The analysis routine applied is: {routine.__name__}
706    """ + (routine.__doc__ or '')
707    return medianResult
708
709
710def perSituation(
711    routine: Callable[[base.Situation], T]
712) -> Callable[[core.DiscreteExploration], List[T]]:
713    """
714    Returns a function which will apply an analysis routine to each
715    situation in an exploration, returning a list of results.
716    """
717    def appliedPerSituation(
718        exploration: core.DiscreteExploration
719    ) -> List[T]:
720        result = []
721        for situ in exploration:
722            result.append(routine(situ))
723        return result
724
725    appliedPerSituation.__name__ = routine.__name__ + "PerSituation"
726    appliedPerSituation.__doc__ = f"""
727    Analysis routine applied to each situation in an exploration,
728    returning a list of results.
729
730    The analysis routine applied is: {routine.__name__}
731    """ + (routine.__doc__ or '')
732    return appliedPerSituation