source: exampleparties/randomparty/randomparty/RandomParty.py@ 59

import logging
from random import randint
import traceback
from typing import cast, Dict, List

from geniusweb.actions.Accept import Accept
from geniusweb.actions.Action import Action
from geniusweb.actions.LearningDone import LearningDone
from geniusweb.actions.Offer import Offer
from geniusweb.actions.PartyId import PartyId
from geniusweb.inform.ActionDone import ActionDone
from geniusweb.inform.Finished import Finished
from geniusweb.inform.Inform import Inform
from geniusweb.inform.Settings import Settings
from geniusweb.inform.YourTurn import YourTurn
from geniusweb.issuevalue.Bid import Bid
from geniusweb.issuevalue.Domain import Domain
from geniusweb.issuevalue.Value import Value
from geniusweb.issuevalue.ValueSet import ValueSet
from geniusweb.party.Capabilities import Capabilities
from geniusweb.party.DefaultParty import DefaultParty
from geniusweb.profile.utilityspace.UtilitySpace import UtilitySpace
from geniusweb.profileconnection.ProfileConnectionFactory import ProfileConnectionFactory
from geniusweb.progress.ProgressRounds import ProgressRounds


class RandomParty (DefaultParty):
    """
    Offers random bids until a bid with sufficient utility is offered.
    """
    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) :
            settings:Settings=cast(Settings,info)    
            self._me = settings.getID()
            self._protocol:str = str(settings.getProtocol().getURI())
            self._progress = settings.getProgress()
            if "Learn" ==  self._protocol:
                self.getConnection().send(LearningDone(self._me))
            else:
                self._profile = ProfileConnectionFactory.create(info.getProfile().getURI(), self.getReporter())
        elif isinstance(info, ActionDone):
            action:Action=cast( ActionDone,info).getAction()
            if isinstance(action, Offer):
                self._lastReceivedBid = cast(Offer, action).getBid()
        elif isinstance(info, YourTurn):
            self._myTurn()
            if isinstance(self._progress, ProgressRounds) :
                self._progress = self._progress.advance()
        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", "Learn"]), set(['geniusweb.profile.utilityspace.LinearAdditive']))

    # Override
    def getDescription(self) -> str:
        return "Offers random bids until a bid with sufficient utility is offered"

    # 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 _myTurn(self):
        if self._lastReceivedBid != None and \
            self._profile.getProfile().getUtility(self._lastReceivedBid) > 0.6:
            action = Accept(self._me, self._lastReceivedBid)
        else:
            for _attempt in range(20):
                bid = self._getRandomBid(self._profile.getProfile().getDomain())
                if self._isGood(bid):
                    break
            action = Offer(self._me, bid);
        self.getConnection().send(action)

    def _isGood(self, bid:Bid)->bool:
        if bid == None:
            return False
        profile = self._profile.getProfile()
        if isinstance(profile, UtilitySpace):
            return profile.getUtility(bid) > 0.6
        raise Exception("Can not handle this type of profile")

    def _getRandomBid(self, domain:Domain) -> Bid:
        values:Dict[str, Value]={iss: self._randValue(domain.getValues(iss)) \
                for iss in domain.getIssues()}
        return Bid(values)
        
    def _randValue(self, vals:ValueSet) -> Value:
        values= list(vals)
        return values[randint(0, len(values)-1)]