mcmas

489class ISPL(Fragment):
490    """
491    Python wrapper for ISPL specifications.
492
493    This permits partials or "fragments", i.e. the specification need not be complete and ready to run.
494
495    See the ISPL reference here: http://mattvonrocketstein.github.io/py-mcmas/isplref
496    """
497
498    PROMPT_HINTS: typing.ClassVar = "Individual proper nouns refer to separate agents."
499
500    def __str__(self):
501        return f"<ISPL: agents={len(self.agents)} vars={len(self.environment.vars)}>"
502
503    def analyze_types(self):
504        types = super().analyze_types()
505        for agent in self.agents.values():
506            types += agent.analyze_types()
507        return list(set(types))
508
509    @classmethod
510    def get_trivial(kls, *args, **kwargs):
511        """
512        ISPL(...) notation.  Unlike ISPL(..)
513
514        This returns the trivial specification, with just enough
515        structure to validate and run, plus any optional
516        overrides.
517        """
518        kwargs.update(check_slice(*args))
519        title = kwargs.pop("title", "Minimal valid ISPL specification")
520        agents = kwargs.pop("agents", {"DefaultAgent": DefaultAgent})
521        environment = kwargs.pop("environment", Environment(...))
522        evaluation = kwargs.pop("evaluation", ["ticking if Environment.ticking=true"])
523        init_states = kwargs.pop("init_states", ["Environment.ticking=true"])
524        formulae = kwargs.pop("formulae", ["ticking"])
525        return kls(
526            title=title,
527            environment=environment,
528            evaluation=evaluation,
529            init_states=init_states,
530            formulae=formulae,
531            agents=agents,
532            **kwargs,
533        )
534
535    # Metadata: typing.ClassVar = fmtk.SpecificationMetadata
536    COMPARATOR: typing.ClassVar = abs
537    REQUIRED: typing.ClassVar = ["agents", "evaluation", "formulae"]
538    parser: typing.ClassVar
539    title: str = Field(
540        default="Untitled Model",
541        description="Optional title.  Used as a comment at the top of the file",
542    )
543    agents: typing.Dict[str, Agent] = Field(
544        default={},
545        description="All agents involved in this specification.  A map of {name: agent_object}",
546    )
547    environment: Environment = Field(
548        default={},
549        description="The environment for this specification",
550    )
551    fairness: typing.Dict[str, typing.List[str]] = Field(
552        default={},
553        description=(
554            "Specifies conditions that must hold infinitely often along "
555            "all execution paths, used to rule out unrealistic behaviors."
556        ),
557    )
558    init_states: typing.InitStateType = Field(
559        default=[],
560        description="Initial global states of the system when verification begins.",
561    )
562    evaluation: typing.EvalType = Field(
563        default=[],
564        description="Calculations, composites, aggregates that can be referenced in formulae",
565    )
566    groups: typing.GroupsType = Field(
567        default={},
568        description=(
569            "A map of {group_name: [member1, .. ]}"
570            "Defines collections of agents for use in group-based verification formulae."
571        ),
572    )
573
574    formulae: typing.FormulaeType = Field(
575        description=(
576            "A list of formulas.\n\n"
577            "These will be partitioned into true/false categories, per the rest of the model"
578        ),
579        default=[],
580    )
581    simulation: SimType = Field(
582        default=None,
583        description=(
584            "The result of simulating this specification.  "
585            "Empty if simulation has never been run"
586        ),
587    )
588    source_code: typing.Union[str, None] = Field(
589        default=None,
590        description=(
591            "Source code for this specification.  "
592            "Only available if the specification was loaded from raw ISPL"
593        ),
594    )
595
596    def __contains__(self, other):
597        """
598        Specification algebra.
599        """
600        if isinstance(other, Agent):
601            return other in self.agents.values()
602        else:
603            raise TypeError(f"__contains__ undefined for {[type(self), type(other)]}")
604
605    def __abs__(self) -> float:
606        """
607        Specification algebra.
608
609        Used in __lt__ and __gt__. For ISPL specifications this
610        is the cumulative sum of formulae complexity
611        """
612        return sum([c.score for c in self.analyze_complexity()])
613
614    def __iadd__(self, other):
615        """
616        Specification algebra.
617
618        This is for in-place addition, i.e. `spec+=Agent(..)`
619        """
620        upd = self + other
621        self.update(upd.model_dump())
622        return self
623
624    def __sub__(self, other):
625        """
626        Specification algebra.
627
628        ISPL - agent => removes agent from this agent list, if present.
629        """
630        # if isinstance(other, (ISPL,)):
631        #     return self.model_copy(update=other.model_dump())
632        # if isinstance(other, (Environment,)):
633        #     return self.model_copy(update=dict(environment=other))
634        if isinstance(other, (Agent,)):
635            agents = {}
636            for a in self.agents.values():
637                if a.name != other.name:
638                    agents[a.name] = a
639            return self.model_copy(update=dict(agents=agents))
640        raise TypeError(f"Cannot add {type(self)} and {type(other)}")
641
642    def __add__(self, other):
643        """
644        Specification algebra.
645
646        ISPL + agent => adds agent to spec ISPL + ISPL =>
647        overrides first spec with values from 2nd
648        """
649        if isinstance(other, (ISPL,)):
650            return self.model_copy(update=other.model_dump())
651        if isinstance(other, (Environment,)):
652            return self.model_copy(update=dict(environment=other))
653        if isinstance(other, (Agent,)):
654            agents = self.agents
655            agents.update(**{other.name: other})
656            return self.model_copy(update=dict(agents=agents))
657        raise TypeError(f"Cannot add {type(self)} and {type(other)}")
658
659    def model_dump_source(self):
660        """
661        Dump the source-code for this piece of the specification.
662        """
663        # from mcmas.engine import dict2ispl
664        return util.dict2ispl(self.model_dump())
665
666    @util.classproperty
667    def parser(self):
668        """
669        Shortcut for `mcmas.parser.parse`
670        """
671        from mcmas import parser
672
673        return parser.parse
674
675    @property
676    def local_advice(self) -> list:
677        """
678        
679        """
680        out = []
681        for agent in self.agents:
682            agentish = self.agents[agent]
683            out += agentish.advice
684        return out
685
686    @classmethod
687    @pydantic.validate_call
688    def load_from_source(
689        kls, txt, strict: bool = False
690    ) -> typing.Dict[str, typing.Self]:
691        """
692        Return ISPL object from given string.
693        """
694        # from mcmas import parser
695        return kls.parser(txt)
696
697    @classmethod
698    @pydantic.validate_call
699    def load_from_ispl_file(
700        kls,
701        file: Optional[str] = None,
702    ):
703        """
704        Return ISPL object from contents of given file.
705        """
706        LOGGER.debug(f"ISPL.load_from_ispl_file: {file}")
707        metadata = dict(file=file)
708        if file:
709            assert os.path.exists(file), f"no such file: {file}"
710            if file.endswith("ispl"):
711                with open(file) as fhandle:
712                    text = fhandle.read()
713                    tmp = kls.parser(text, file=file)
714                    data = tmp.model_dump(exclude="metadata")
715                    return ISPL(metadata=ISPL.Metadata(**metadata), **data)
716            #    return ISPL(metadata=dict(file=file, engine="bonk"), **data)
717            elif file.endswith("json"):
718                raise ValueError("refusing to work with ispl file")
719            elif file in ["-", "/dev/stdin"]:
720                metadata.update(file="<<stream>>")
721                text = sys.stdin.read().strip()
722                data = kls.parser(text).model_dump(exclude="metadata")
723                return ISPL(metadata=ISPL.Metadata(**metadata), **data)
724            else:
725                LOGGER.critical(f"could not create model from {file}")
726                raise Exception(file)
727        if text:
728            LOGGER.critical("NIY")
729            raise Exception(text)
730
731    load_from_file = load_from_ispl_file
732
733    @classmethod
734    @pydantic.validate_call
735    def load_from_json_file(kls, file: Optional[str] = None, text=None):
736        """
737        Return ISPL object from the contents of given file.
738
739        File *must* be JSON encoded.
740        """
741        LOGGER.critical(f"ISPL.load_from_json_file: {file}")
742        metadata = dict(file=file)
743        if file:
744            assert os.path.exists(file), f"no such file: {file}"
745            if file.endswith("json"):
746                with open(file) as fhandle:
747                    data = json.loads(fhandle.read())
748                return ISPL(metadata=ISPL.Metadata(**metadata), **data)
749            elif file.endswith("ispl"):
750                raise ValueError("refusing to work with ispl file")
751            elif file in ["-", "/dev/stdin"]:
752                metadata.update(file="<<stream>>")
753                text = sys.stdin.read().strip()
754                data = json.loads(text)
755                return ISPL(
756                    metadata=ISPL.Metadata(**{**data.pop("metadata", {}), **metadata}),
757                    **data,
758                )
759            else:
760                LOGGER.critical(f"could not create model from {file}")
761                raise Exception(file)
762
763        if text:
764            LOGGER.critical("NIY")
765            raise Exception(text)
766
767    @property
768    @pydantic.validate_call
769    def validates(self) -> bool:
770        """
771        Asks the engine directly whether this spec validates.
772
773        Note that this is ground-truth and not heuristic like the `valid` property elsewhere!
774        """
775        return mcmas.engine.validate(model=self)
776
777    @property
778    @pydantic.validate_call
779    def analysis(self) -> spec.Analysis:
780        """
781        Static-analysis for this ISPL specification.
782
783        Returns details about symbols and logical operators.
784        """
785        ents = list(self.agents.values())
786        ents += [self.environment]
787        vars = []
788        for agent in ents:
789            if isinstance(agent, (Agent,)):
790                vars += [k.strip() for k in (agent.lobsvars or []) if k.strip()]
791            vars += [k.strip() for k in (agent.vars or []) if k.strip()]
792        vars = list(set(vars))
793        meta = dict(
794            symbols=spec.SymbolMetadata(
795                agents=[agent for agent in self.agents],
796                vars=vars,
797                actions=list(
798                    set(
799                        functools.reduce(
800                            operator.add,
801                            [self.agents[agent].actions for agent in self.agents],
802                        )
803                    )
804                ),
805            ),
806            types=self.analyze_types(),
807            complexity=self.analyze_complexity(),
808        )
809        meta = spec.Analysis(**meta)
810        return meta
811
812    def analyze_complexity(self) -> typing.List:
813        from mcmas.logic import complexity
814
815        return [
816            complexity.analyzer.analyze(f.lstrip().rstrip(), index=i)
817            for i, f in enumerate(self.formulae)
818        ]
819
820    def exec(self, strict: bool = False, **kwargs) -> typing.Self:
821        """
822        Execute this ISPL specification.
823        """
824        required = ["init_states"]
825        self.logger.debug(f"validating: {self.source_code or self.model_dump_source()}")
826
827        # check advice before execution
828        if self.advice:
829            msg = "exec: Model has non-empty advice!"
830            LOGGER.critical(msg)
831            if strict:
832                raise RuntimeError(msg)
833
834        for k in required:
835            if not getattr(self, k):
836                err = f"Validation failed.  Required key `{k}` is missing."
837                return self.model_copy(
838                    update={
839                        # "source_code": src,
840                        "simulation": Simulation(
841                            error=err,
842                            metadata=Simulation.Metadata(parsed=False, validates=False),
843                        ),
844                    }
845                )
846
847        self.logger.debug("starting..")
848        sim = mcmas.engine(text=self.model_dump_source(), output_format="model")
849        out = self.model_copy(
850            update=dict(
851                source_code=self.source_code or self.model_dump_source(),
852                simulation=sim,
853                metadata=self.metadata.model_dump(),
854            )
855        )
856        self.logger.debug("done")
857        return out
858
859    run_sim = run_simulation = exec
860
861    def repl(self):
862        """
863        Start a REPL shell with this object available as `spec`.
864        """
865        result_model = self.exec()
866        return util.repl(spec=result_model)

288class Agent(Fragment):
289    """
290    Python wrapper for ISPL Agents.
291
292    See also the relevant [ISPL reference](http://mattvonrocketstein.github.io/py-mcmas/isplref/#agent)
293    """
294
295    COMPARATOR: typing.ClassVar = len
296    DefaultAgent: typing.ClassVar
297    REQUIRED: typing.ClassVar = ["protocol", "evolution", "actions"]
298    parser: typing.ClassVar
299
300    name: str = Field(
301        default="player",
302        description=("Name of this agent"),
303    )
304    actions: typing.ActionsType = ActionsField
305    evolution: typing.EvolType = EvolutionField
306    obsvars: typing.ObsVarsType = ObsvarsField
307    protocol: typing.ProtocolType = ProtocolField
308    vars: typing.VarsType = VarsField
309
310    lobsvars: typing.LobsvarsType = Field(
311        description="",
312        default=[],
313    )
314    red_states: typing.List[str] = Field(
315        description="",
316        default=[],
317    )
318
319    __len__ = Environment.__len__
320
321    @util.classproperty
322    def DefaultAgent(kls):
323        """
324        Returns the trivial agent.
325        """
326        return kls(...)
327
328    @classmethod
329    def _load_from_pydantic_ai_agent(kls, pagent, **extra) -> typing.Self:
330        """
331        Create ISPL agent from the given pydantic agent.
332        """
333        kls.logger.info(f"_load_from_pydantic_ai_agent: {pagent}")
334        from mcmas import util
335
336        name = pagent.name or f"Agent_{id(pagent)}"
337        actions = list(pagent._function_toolset.tools.keys())
338        tools = list(pagent._function_toolset.tools.values())
339        tool = tools[0]
340        if len(tools) > 1:
341            kls.logger.warning(f"pydantic agent `{name}` has multiple tools!")
342            kls.logger.warning(f"using just the first one: {tool}")
343        # naive conversion from function signature to ISPL types.
344        vars = util.fxn_sig.as_ispl_types(tool.function)
345        vars.pop("ctx", None)
346        return Agent(
347            name=name,
348            actions=actions,
349            vars=vars,
350            metadata=dict(
351                parser=f"{kls.__module__}.{kls.__name__}._load_from_pydantic_ai_agent",
352                file=inspect.getfile(pagent.__class__),
353            ),
354            **extra,
355        ).model_completion()
356
357    def analyze_symbols(self) -> typing.List[str]:
358        """
359        Returns symbol-related metadata including agent-names,
360        actions, variables, and type info.
361        """
362        return spec.SymbolMetadata(
363            agents=[self.name],
364            actions=self.actions,
365            vars=[k.strip() for k in list(self.lobsvars) + list(self.vars)],
366            types=self.analyze_types(),
367        )
368
369    @property
370    @pydantic.validate_call
371    def analysis(self) -> spec.Analysis:
372        """
373        Static-analysis for this Agent specification.
374
375        Returns details about symbols and logical operators.
376        """
377        return spec.Analysis(
378            symbols=self.analyze_symbols(),
379            types=self.analyze_types(),
380            complexity=self.analyze_complexity(),
381        )
382        # out.actions = [getattr(symbols, x) for x in sorted(list(set(out.actions)))]
383        # out.vars = [getattr(symbols, x) for x in sorted(list(set(out.vars)))]
384        # out.agents = [getattr(symbols, x) for x in sorted(list(set(out.agents)))]
385        # return meta
386
387    def __pow__(self, other: float = 0.1) -> typing.Self:
388        """
389        
390        """
391        from mcmas import ai
392
393        return ai.model_mutation(obj=self, model_settings=dict(top_p=other))
394
395    def __invert__(self) -> typing.Self:
396        """
397        Trigger completion for this Agent.
398
399        This is NOT backed by an LLM; see instead `mcmas.ai.agent_completion`.
400        """
401        if not self.advice:
402            err = f"{self} is already valid, returning it instead of completing"
403            LOGGER.warning(err)
404            return self
405        else:
406            tmp = self
407            trivial = DefaultAgent
408            defaults = dict(
409                protocol=trivial.protocol,
410                vars=trivial.vars,
411                evolution=trivial.evolution,
412            )
413            for x in ["protocol", "vars", "evolution"]:
414                if getattr(self, x, None):
415                    pass
416                else:
417                    LOGGER.warning(f"could not find required {x}")
418                    tmp = tmp.model_copy(update={x: defaults[x]})
419            return tmp.model_copy(
420                update=dict(actions=list(set(tmp.actions + trivial.actions)))
421            )
422
423    model_completion = __invert__
424
425    @classmethod
426    def get_trivial(kls, *args, **kwargs):
427        """
428        Smallest legal agent.
429        """
430        kwargs.update(check_slice(*args))
431        return kls(
432            name="trivial",
433            vars=dict(ticking="boolean"),
434            actions=["tick"],
435            protocol=["Other : {tick}"],
436            evolution=["ticking=true if Action=tick;"],
437        )
438
439    def model_dump_source(self):
440        """
441        Dump the source-code for this piece of the specification.
442        """
443        from mcmas import rendering
444
445        return rendering.get_template("Agent.j2").render(
446            name=self.name, agent=self.model_dump()
447        )
448
449    @util.classproperty
450    def parser(self) -> typing.Callable:
451        """
452        Return an appropriate parser for this spec-fragment.
453        """
454        from mcmas import parser
455
456        return parser.extract_agents
457
458    @classmethod
459    @pydantic.validate_call
460    def load_from_source(kls, txt, strict: bool = False) -> typing.Self:
461        """
462        Load ISPL agent from string.
463        """
464        agents = kls.parser(txt)
465        agents.pop("Environment", None)
466        if len(agents) != 1:
467            LOGGER.critical("load_from_source: more than 1 agent! returning first..")
468        return Agent(**list(agents.values())[0])
469
470    @property
471    def local_advice(self) -> list:
472        return []