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

Last change on this file since 93 was 90, checked in by Bart Vastenhouw, 3 years ago

Refactor to help reusing partiesserver.

File size: 3.7 KB
Line 
1import json
2import logging
3import sys
4from typing import Dict, Any, Set
5from typing import cast
6
7from geniusweb.actions.Accept import Accept
8from geniusweb.actions.Action import Action
9from geniusweb.actions.Offer import Offer
10from geniusweb.actions.PartyId import PartyId
11from geniusweb.actions.Vote import Vote
12from geniusweb.actions.Votes import Votes
13from geniusweb.inform.ActionDone import ActionDone
14from geniusweb.inform.Finished import Finished
15from geniusweb.inform.Inform import Inform
16from geniusweb.inform.OptIn import OptIn
17from geniusweb.inform.Settings import Settings
18from geniusweb.inform.Voting import Voting
19from geniusweb.inform.YourTurn import YourTurn
20from geniusweb.issuevalue.Bid import Bid
21from geniusweb.party.Capabilities import Capabilities
22from geniusweb.party.DefaultParty import DefaultParty
23from geniusweb.utils import val
24
25
26class TestParty (DefaultParty):
27 """
28 A Stupid party that places empty bids because it can't download the profile,
29 and accepts the first incoming offer.
30 """
31 def __init__(self):
32 super().__init__()
33 self.getReporter().log(logging.INFO,"party is initialized")
34 self._profile = None
35 self._lastReceivedBid:Bid = None
36
37 # Override
38 def notifyChange(self, info: Inform):
39 #self.getReporter().log(logging.INFO,"received info:"+str(info))
40 if isinstance(info,Settings) :
41 self._settings:Settings=cast(Settings,info)
42 self._me = self._settings.getID()
43 self._protocol = self._settings.getProtocol()
44 elif isinstance(info, ActionDone):
45 action:Action=cast( ActionDone,info).getAction()
46 if isinstance(action, Offer):
47 self._lastReceivedBid = cast(Offer, action).getBid()
48 elif isinstance(info, YourTurn):
49 # This is a stupid party
50 if self._lastReceivedBid!=None:
51 self.getReporter().log(logging.INFO,"sending accept:")
52 accept = Accept(self._me, self._lastReceivedBid)
53 val(self.getConnection()).send(accept)
54 else:
55 # We have no clue about our profile
56 offer:Offer = Offer(self._me, Bid({}))
57 self.getReporter().log(logging.INFO,"sending empty offer:")
58 val(self.getConnection()).send(offer)
59 self.getReporter().log(logging.INFO,"sent empty offer:")
60 elif isinstance(info, Voting):
61 # MOPAC protocol
62 self._lastvotes = self._vote(cast(Voting, info));
63 val(self.getConnection()).send(self._lastvotes)
64 elif isinstance(info, OptIn):
65 val(self.getConnection()).send(self._lastvotes)
66
67 elif isinstance(info, Finished):
68 self.terminate()
69 else:
70 self.getReporter().log(logging.WARNING, "Ignoring unknown info "+str(info))
71
72
73 # Override
74 def getCapabilities(self): # -> Capabilities
75 return Capabilities( set([ "SAOP"]), set(['geniusweb.profile.utilityspace.LinearAdditive']))
76
77 # Override
78 def getDescription(self):
79 return "Offers null bids and accept other party's first bid"
80
81 # Override
82 def terminate(self):
83 self.getReporter().log(logging.INFO,"party is terminating:")
84 super().terminate()
85 if self._profile != None:
86 self._profile.close()
87 self._profile = None
88
89 def _vote(self, voting:Voting) ->Votes :
90 '''
91 @param voting the {@link Voting} object containing the options
92
93 @return our next Votes. Stupid: vote for all received offers.
94 '''
95 votes:Set[Vote] = set([Vote(self._me, offer.getBid(), 2, 5)\
96 for offer in voting.getOffers() ])
97 return Votes(self._me, votes)
Note: See TracBrowser for help on using the repository browser.