Python context 模块，Context() 实例源码

def _construct_backtest_environment(self):
        '''
        ????????
        '''


        self.data = BacktestDataInitilizer(
                self.universe,
                self.start_date,
                self.end_date)

        self.blotter = Blotter(self.data)
        self.context = Context(self.cash,
                               self.universe,
                               self.blotter,
                               self.data) 
        self.recorder = Recorder()

def ts1_static(gamma: Context, typ: Typ, n):
    ret = []
    for ctx_declaration in gamma.ctx.values():
        f = fresh(ctx_declaration.typ, typ, n)
        mu = mgu(typ, f.typ)
        if not mu.is_failed():
            sigma = mu.restrict(typ)
            ret.append(PreTs1Res(ctx_declaration.sym, sigma))
    return ret

def __init__(self,user_id):
        import SimpleReactor.logger, context, rpc_codec, rpc_invoker, io_client
        self.logger = SimpleReactor.logger.Logger()  # ??
        self.user_id = user_id
        self.context = context.Context(self.user_id)  # ???,????,????
        self.rpc_codec = rpc_codec.RPCCodec(self.logger)  # ???
        self.io_client = io_client.IOClient(0.01, self.logger, self.rpc_codec, self.context)  # ???io

        self.io_client.start()  # ??

        self.rpc_invoker = rpc_invoker.RPCInvoker(self.context, self.logger, self.rpc_codec,
                                                  self.io_client)  # rpc invoker

def run_file():
    with open(sys.argv[1], "r") as f:
        content = f.read()
        ctx = c.Context()
        import_prelude(ctx)
        execute(content, False, ctx)

def process_message(intent_classifier, bow_vectorizer, user_context: Context, text: str):

    annotated_text = nlp.AnnotatedText(text)

    entities = annotated_text.entities

    cur_topic = user_context.current_topic

    if cur_topic:

        # searching for the expected entity
        for entity in entities:
            if entity.type in cur_topic.expecting_entity:
                # if expected entity found is found in user input,
                # removes the expected entity from the InteractionTopic
                cur_topic.remove_expected_entity(entity.type)

                # we're guessing here that what the user input is
                # related to the last topic
                cur_topic.add_conversation(annotated_text)

                # calls the requested slot
                fs_tuple = intent_to_slot(cur_topic.intent, cur_topic.frame)

                update_interaction(fs_tuple.slot, user_context)
                return

        # user probably did not respond with expected entity
        print("I'm confused (have current topic with no expected entity)")

    else:
        # There is no previous topic
        intent = nlp.predict(intent_classifier, bow_vectorizer, annotated_text.root_token)
        fs_tuple = create_frame(intent, user_context)
        user_context.current_topic = InteractionTopic(intent, fs_tuple.frame, annotated_text)
        update_interaction(fs_tuple.slot, user_context)


# TODO: need a better name for this function

def update_interaction(slot: Frame, context: Context):
    message = slot.user_message()
    if message:
        # TODO: need a better way to signal current slot cannot generate response
        print(message)

        # when a message is sent, we can assume that the current
        # topic is done, and can be deleted.
        del context.current_topic

def create_frame(intent: Intent, context: Context) -> FS:
    """
    This function serves to link Intents to their respective frames
    """
    fs_tuple = mapping[intent]

    # Instantiates frame and binds it to the context
    frame_instance = fs_tuple.frame(context)
    frame_slot = frame_instance.__getattribute__(fs_tuple.slot)

    return FS(frame_instance, frame_slot)

def main():
    parser = argparse.ArgumentParser(description="The interpreter for Pluto")

    parser.add_argument("-f", "--file", action="store", dest="file", type=str, help="the file to execute")
    parser.add_argument("-p", "--parse", action="store_true", default=False, help="just parse the file - don't execute it")
    parser.add_argument("-t", "--tree", action="store_true", default=False, help="print the parse tree")
    parser.add_argument("-i", "--interactive", action="store_true", default=False, help="enter interactive mode after the file has been run")
    parser.add_argument("-n", "--no-prelude", action="store_true", dest="no_prelude", help="don't load the prelude")
    parser.add_argument("-v", "--version", action="version", version="Pluto, early beta version")

    args = parser.parse_args()

    if args.file == None:
        ctx = c.Context()

        if not args.no_prelude:
            import_prelude(ctx)

        repl(ctx)
    else:
        try:
            text = open(args.file).read()

            if args.parse or args.tree:
                tokens = l.lex(text)
                parse = p.Parser(tokens)
                program = parse.parse_program()

                if len(parse.errors) > 0:
                    parse.print_errors()
                elif args.tree:
                    print(program)

                return

            ctx = c.Context()

            if not args.no_prelude:
                import_prelude(ctx)

            execute(text, False, ctx)

            if args.interactive:
                print()
                repl(ctx)

        except FileNotFoundError:
            print("File not found: %s" % args.file)
            return