source: geniuswebcore/test/testparty/TestParty.py@ 100

import json
import logging
import sys
from typing import Dict, Any, Set
from typing import cast

from geniusweb.actions.Accept import Accept
from geniusweb.actions.Action import Action
from geniusweb.actions.Offer import Offer
from geniusweb.actions.PartyId import PartyId
from geniusweb.actions.Vote import Vote
from geniusweb.actions.Votes import Votes
from geniusweb.inform.ActionDone import ActionDone
from geniusweb.inform.Finished import Finished
from geniusweb.inform.Inform import Inform
from geniusweb.inform.OptIn import OptIn
from geniusweb.inform.Settings import Settings
from geniusweb.inform.Voting import Voting
from geniusweb.inform.YourTurn import YourTurn
from geniusweb.issuevalue.Bid import Bid
from geniusweb.party.Capabilities import Capabilities
from geniusweb.party.DefaultParty import DefaultParty
from geniusweb.utils import val


class TestParty (DefaultParty):
    """
    A Stupid party that places empty bids because it can't download the profile, 
    and accepts the first incoming offer.
    """
    def __init__(self):
        super().__init__()
        self.getReporter().log(logging.INFO,"party is initialized")
        self._profile = None
        self._lastReceivedBid:Bid = None

    # Override
    def notifyChange(self, info: Inform):
        #self.getReporter().log(logging.INFO,"received info:"+str(info))
        if isinstance(info,Settings) :
            self._settings:Settings=cast(Settings,info)    
            self._me = self._settings.getID()
            self._protocol = self._settings.getProtocol()
        elif isinstance(info, ActionDone):
            action:Action=cast( ActionDone,info).getAction()
            if isinstance(action, Offer):
                self._lastReceivedBid = cast(Offer, action).getBid()
        elif isinstance(info, YourTurn):
            # This is a stupid party
            if self._lastReceivedBid!=None:
                self.getReporter().log(logging.INFO,"sending accept:")
                accept =  Accept(self._me, self._lastReceivedBid)
                val(self.getConnection()).send(accept)
            else:
                # We have no clue about our profile
                offer:Offer = Offer(self._me, Bid({}))
                self.getReporter().log(logging.INFO,"sending empty offer:")
                val(self.getConnection()).send(offer)
                self.getReporter().log(logging.INFO,"sent empty offer:")
        elif isinstance(info, Voting):
            # MOPAC protocol
            self._lastvotes = self._vote(cast(Voting, info));
            val(self.getConnection()).send(self._lastvotes)
        elif isinstance(info, OptIn):
            val(self.getConnection()).send(self._lastvotes)

        elif isinstance(info, Finished):
            self.terminate()
        else:
            self.getReporter().log(logging.WARNING, "Ignoring unknown info "+str(info))


    # Override
    def getCapabilities(self): # -> Capabilities
        return Capabilities( set([ "SAOP"]), set(['geniusweb.profile.utilityspace.LinearAdditive']))

    # Override
    def getDescription(self):
        return "Offers null bids and accept other party's first bid"

    # Override
    def terminate(self):
        self.getReporter().log(logging.INFO,"party is terminating:")
        super().terminate()
        if self._profile != None:
            self._profile.close()
            self._profile = None

    def _vote(self, voting:Voting) ->Votes :
        '''
        @param voting the {@link Voting} object containing the options
        
        @return our next Votes. Stupid: vote for all received offers.
        '''
        votes:Set[Vote]  = set([Vote(self._me, offer.getBid(), 2, 5)\
                for offer in voting.getOffers() ])
        return Votes(self._me, votes)