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)