source: CSE3210/agent14/agent14.py

import logging
import time
from decimal import Decimal
from typing import cast

from geniusweb.actions.Accept import Accept
from geniusweb.actions.Action import Action
from geniusweb.actions.Offer import Offer
from geniusweb.bidspace.BidsWithUtility import BidsWithUtility
from geniusweb.bidspace.Interval import Interval
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.opponentmodel.FrequencyOpponentModel import FrequencyOpponentModel
from geniusweb.party.Capabilities import Capabilities
from geniusweb.party.DefaultParty import DefaultParty
from geniusweb.profileconnection.ProfileConnectionFactory import (
    ProfileConnectionFactory,
)
from geniusweb.progress.ProgressRounds import ProgressRounds
from tudelft_utilities_logging.Reporter import Reporter


class Agent14(DefaultParty):

    def __init__(self, reporter: Reporter = None):
        super().__init__(reporter)
        self.getReporter().log(logging.INFO, "party is initialized")
        self._last_received_bid = None
        self.last_my_utility = 1
        self.last_bids = []

    def notifyChange(self, info: Inform):
        """This is the entry point of all interaction with your agent after is has been initialised.

        Args:
            info (Inform): Contains either a request for action or information.
        """

        # a Settings message is the first message that will be send to your
        # agent containing all the information about the negotiation session.
        if isinstance(info, Settings):
            self._settings: Settings = cast(Settings, info)
            self._me = self._settings.getID()

            # progress towards the deadline has to be tracked manually through the use of the Progress object
            self._progress = self._settings.getProgress()
            # the profile contains the preferences of the agent over the domain
            self._profile = ProfileConnectionFactory.create(
                info.getProfile().getURI(), self.getReporter()
            )
            # initialize FrequencyOpponentModel
            self._opponent_model = FrequencyOpponentModel.create().With(
                newDomain=self._profile.getProfile().getDomain(),
                newResBid=self._profile.getProfile().getReservationBid())

            # sort all issues by relevance
            profile = self._profile.getProfile()
            issue_infos = BidsWithUtility._getInfo(profile, 6)
            weights = self._profile.getProfile().getWeights()
            issue_infos.sort(key=lambda x: weights[x.getName()])
            issue_infos.reverse()

            # initialize BidsWithUtility with sorted issues
            self._all_bids = BidsWithUtility(issue_infos, 6)

        # ActionDone is an action send by an opponent (an offer or an accept)
        elif isinstance(info, ActionDone):
            action: Action = cast(ActionDone, info).getAction()

            # if it is an offer, set the last received bid
            if isinstance(action, Offer):
                self._last_received_bid = cast(Offer, action).getBid()

                if self._last_received_bid:
                    # Adjust FrequencyOpponentModel with new bid
                    self._opponent_model = self._opponent_model.WithAction(action=action, progress=self._progress)
        # YourTurn notifies you that it is your turn to act
        elif isinstance(info, YourTurn):
            action = self._myTurn()
            if isinstance(self._progress, ProgressRounds):
                self._progress = self._progress.advance()
            self.getConnection().send(action)

        # Finished will be send if the negotiation has ended (through agreement or deadline)
        elif isinstance(info, Finished):
            # terminate the agent MUST BE CALLED
            self.terminate()
        else:
            self.getReporter().log(
                logging.WARNING, "Ignoring unknown info " + str(info)
            )

    # lets the geniusweb system know what settings this agent can handle
    # leave it as it is for this competition
    def getCapabilities(self) -> Capabilities:
        return Capabilities(
            set(["SAOP"]),
            set(["geniusweb.profile.utilityspace.LinearAdditive"]),
        )

    # terminates the agent and its connections
    # leave it as it is for this competition
    def terminate(self):
        self.getReporter().log(logging.INFO, "party is terminating:")
        super().terminate()
        if self._profile is not None:
            self._profile.close()
            self._profile = None

    # give a description of your agent
    def getDescription(self) -> str:
        return "Group 14 Negotiation agent"

    # execute a turn
    def _myTurn(self):
        # check if the last received offer if the opponent is good enough
        if self._isGood(self._last_received_bid):
            # if so, accept the offer
            action = Accept(self._me, self._last_received_bid)
        else:
            # if not, find a bid to propose as counter offer
            bid = self._findBid()
            action = Offer(self._me, bid)

        # send the action
        return action

    # method that checks if we would agree with an offer.
    # our agent accepts the offer only if utility received is higher than
    # last utility calculated by _getNextUtility
    def _isGood(self, bid: Bid) -> bool:
        if bid is None:
            return False
        return self._profile.getProfile().getUtility(bid) >= self.last_my_utility

    # This calculates utility of the next bid agent should make
    # e in range [1, inf] - exponential coefficient
    # k in range [1, inf] - linear coefficient
    def _getNextUtility(self, e, k):
        util = 1 - self._progress.get(time.time() * 1000) ** e - 1/k * self._progress.get(time.time() * 1000)
        return util

    def _findBid(self) -> Bid:
        # Calculate next utility
        # Optimal values are k=4 and e=10
        utility = Decimal(self._getNextUtility(4, 10))
        eps = Decimal(0.01)

        # Looks for bids in BidsUtilitySpace in interval(utility-eps, utility+eps)
        # In case no bids found, extends interval
        best_bid = None
        while best_bid is None:
            bids_interval = self._all_bids.getBids(Interval(min=utility - eps, max=utility + eps))

            # Finds the maximum utility bid for the opponent in interval
            max_util = Decimal(0)
            for bid in bids_interval:
                # Making sure that bid was not proposed in last n turns
                # Optimal parameter for n is 5
                if bid not in self.last_bids[-5:]:
                    bid_util = self._opponent_model.getUtility(bid)
                    if bid_util >= max_util:
                        best_bid = bid
                        max_util = bid_util
            eps = eps + Decimal(0.01)

        self.last_my_utility = utility
        self.last_best_bid = best_bid
        self.last_bids.append(best_bid)
        return best_bid