Управление группой мобильных объектов

            if alpha < 0:

                if abs(alpha) > pi/9:

                    move.left_track_power = -1

                    move.right_track_power = me.engine_rear_power_factor

                if abs(alpha) < pi/18:

                    move.left_track_power = me.engine_rear_power_factor

                    move.right_track_power = -1

            else:

                if abs(alpha) > pi/9:

                    move.left_track_power = me.engine_rear_power_factor

                    move.right_track_power = -1

                if abs(alpha) < pi/18:

                    move.left_track_power = -1

                    move.right_track_power = me.engine_rear_power_factor

#        if frontSideAttacked:

#            print 'front'

 

    #can we dodge moving directly forward/backward

    def canBeDodged(self, power, me, shell, time):

#        ax = a * cos(me.angle)

#        ay = a * sin(me.angle)

        calculation = Calculation()

        newMe = Unit(6666, me.width, me.height, me.x, me.y, me.speedX, me.speedY, me.angle, me.angular_speed)

        newMe.x, newMe.y = calculation.get_new_pos(me, time, power)

#        newMe.x = newMe.x + newMe.speedX*time + ax*time*time/2

#        newMe.y = newMe.y + newMe.speedY*time + ay*time*time/2

 

        MinD, pt = self.hits(shell, newMe)

        if MinD == 6666 and pt.x == -1 and pt.y == -1:

            return True

        else:

            return False

 

    #can we dodge at all. Returns capability to dodge; left track power; right track power

    def tryDodge(self, me, shell, time, move):

 

        forward_works = False

        backward_works = False

 

        if self.canBeDodged(1, me, shell, time):

            forward_works = True

        if self.canBeDodged(-me.engine_rear_power_factor, me, shell, time):

            backward_works = True

 

        if not forward_works and not backward_works:

            return False, move.left_track_power, move.right_track_power

        if forward_works and not backward_works:

            return True, 1, 1

        if not forward_works and backward_works:

            return True, -1, -1

        if forward_works and backward_works:

            if me.speedX == 0:

                angleV = pi/2

            else:

                angleV = atan(me.speedY / me.speedX)

            if abs(angleV - me.angle) <= pi / (1 + me.engine_rear_power_factor):

                return True, 1, 1

            else:

                return True, -1, -1

 

 

    def moveToDodge(self, me, shell, move):

 

        Dist, pt = self.hits(shell, me)

 

        distance = shell.get_distance_to(pt.x, pt.y)

        if shell.type == FireType.PREMIUM:

            time = bisect(Factor.SHELL_PREMIUM_DISTANCE, distance)

        else:

            time = bisect(Factor.SHELL_REGULAR_DISTANCE, distance)

 

Модуль Head.py

# -*- coding: utf-8 -*-

from bisect import bisect

 

__author__ = 'yokre_000'

import copy

from Calculation import Calculation

from model.FireType import FireType

from Factor import Factor

class Head():

 

    def __init__(self):

        self.locks = {}

 

    def move(self, me, world, move):

        targets = self.get_targets(world)

        obstacles = self.get_obstacles(world, me)

        if me.remaining_reloading_time == 0:

            if self.can_fire_any_aimed(me, targets, obstacles):

                move.fire_type = FireType.PREMIUM_PREFERRED

        selected_target = self.select_target(me, targets, obstacles)

        if selected_target:

            move.turret_turn = Calculation.turret_angle_to_moving_unit(me, selected_target)

 

    def can_fire_any_aimed(self, me, targets, obstacles):

 

        bad_obstacles = [] # содержит дистанции до плохих целей

 

        for obstacle in obstacles:

            if Calculation.can_hit_unmoving_unit(me, obstacle):

                bad_obstacles.append(me.get_distance_to_unit(obstacle))

 

        for target in targets:

            if Calculation.aimed_to_moving_unit(me, target):

                distance = me.get_distance_to_unit(target)

                can_hit = True

                for d in bad_obstacles:

                    if d < distance:

                        can_hit = False

                        break

                return can_hit

 

        return False

 

 

    def get_targets(self, world):

        """

        Возвращает  список возможных целей

        """

        good_targets = []

        for tank in world.tanks:

            if not tank.teammate and tank.hull_durability > 0 and tank.crew_health > 0:

                good_targets.append(tank)

        return good_targets

 

    def get_obstacles(self, world, me):

        """

        Возвращает  список препятствий

        """

        obstacles = []

        if len(world.tanks) > 0:

            for tank in world.tanks:

                if tank.hull_durability <= 0 or tank.crew_health <= 0 or (tank.teammate and tank.id != me.id):

                    obstacles.append(tank)

        [obstacles.append(b) for b in world.bonuses]

        [obstacles.append(o) for o in world.obstacles]

        return obstacles

 

    def select_target(self, me, targets, obstacles):

        rotate_time_for_targets = []

        new_targets = copy.deepcopy(targets)

        for key in self.locks:

            if self.locks[key] > 0:

                self.locks[key] -= 1

                bad_target = None

                for target in targets:

                    if target.id == key:

                        bad_target = target

                        break

                if bad_target:

                    targets.remove(bad_target)

        if not targets:

            targets = new_targets

 

        # просчет  времени поворота

        for target in targets:

            time = self.get_time_to_rotate(me, target)

            if time < me.remaining_reloading_time:

                time = -1

            rotate_time_for_targets.append((target, time))

 

            # удалим цели, в которые не cможем попасть

        #        for target in rotate_time_for_targets:

        #            if not Calculation.can_hit_tank_in_future(me, self.me_previous_tick, target[0], self.get_previous_tank(target[0]), target[1]):

        #                rotate_time_for_targets.remove(target)

 

        if not rotate_time_for_targets:

            return targets[0]

            # сортировка по времени

        rotate_time_for_targets.sort(key = lambda x: x[1])

        min_distance = 2000 # не может быть выше

        selected_target = None

        for target in rotate_time_for_targets:

            # выбрать только те цели, до которых успеем повернуться

            if target[1] != -1: break

            # из них выбрать приоритетные

            distance = me.get_distance_to_unit(target[0])

            if distance < min_distance:

                min_distance = distance

                selected_target = target[0]

            # если таких целей нет, то  выбираем ту, до которой меньше  крутиться

        if not selected_target:

            selected_target = rotate_time_for_targets[0][0]

        return selected_target

 

    def get_time_to_rotate(self, me, target):

        """

        Возвращает  время, необходимое для поворота  башни

        Не  учитывает угловую скорость танка

        Не  учитывает новое положение цели

        """

        angle = me.get_turret_angle_to_unit(target)

        time = abs(angle) / me.turret_turn_speed

        return int(time)

 

Модуль RemoteProcessClient.py

import socket

import struct

from model.Bonus import Bonus

from model.BonusType import BonusType

from model.Obstacle import Obstacle

from model.Player import Player

from model.PlayerContext import PlayerContext

from model.Shell import Shell

from model.ShellType import ShellType

from model.Tank import Tank

from model.TankType import TankType

from model.World import World

 

class RemoteProcessClient:

    LITTLE_ENDIAN_BYTE_ORDER = True

 

    SIGNED_BYTE_FORMAT_STRING = '<b' if LITTLE_ENDIAN_BYTE_ORDER else '>b'

    INTEGER_FORMAT_STRING = '<i' if LITTLE_ENDIAN_BYTE_ORDER else '>i'

    LONG_FORMAT_STRING = '<q' if LITTLE_ENDIAN_BYTE_ORDER else '>q'

    DOUBLE_FORMAT_STRING = '<d' if LITTLE_ENDIAN_BYTE_ORDER else '>d'

 

    SIGNED_BYTE_SIZE_BYTES = 1

    INTEGER_SIZE_BYTES = 4

    LONG_SIZE_BYTES = 8

    DOUBLE_SIZE_BYTES = 8

 

    def __init__(self, host, port):

        self.socket = socket.socket()

        self.socket.connect((host, port))

 

    def write_token(self, token):

        self.write_enum(RemoteProcessClient.MessageType.AUTHENTICATION_TOKEN)

        self.write_string(token)

 

    def read_team_size(self):

        message_type = self.read_enum(RemoteProcessClient.MessageType)

        self.ensure_message_type(message_type, RemoteProcessClient.MessageType.TEAM_SIZE)

        return self.read_int()

 

    def write_selected_tanks(self, tank_types):

        self.write_enum(RemoteProcessClient.MessageType.TANK_TYPES)

 

        if tank_types is None:

            self.write_int(-1)

        else:

            self.write_int(tank_types.__len__())

 

            for type in tank_types:

                self.write_enum(type)

 

    def read_player_context(self):

        message_type = self.read_enum(RemoteProcessClient.MessageType)

        if message_type == RemoteProcessClient.MessageType.GAME_OVER:

            return None

 

        self.ensure_message_type(message_type, RemoteProcessClient.MessageType.PLAYER_CONTEXT)

        return PlayerContext(self.read_tanks(), self.read_world()) if self.read_boolean() else None

 

    def write_moves(self, moves):

        self.write_enum(RemoteProcessClient.MessageType.MOVES)

 

        if moves is None:

            self.write_int(-1)

        else:

            self.write_int(moves.__len__())

 

            for move in moves:

                if move is None:

                    self.write_boolean(False)

                else:

                    self.write_boolean(True)

 

                    self.write_double(move.left_track_power)

                    self.write_double(move.right_track_power)

                    self.write_double(move.turret_turn)

                    self.write_enum(move.fire_type)

 

    def close(self):

        self.socket.close()

 

    def read_world(self):

        if not self.read_boolean():

            return None

 

        return World(

            self.read_int(), self.read_double(), self.read_double(), self.read_players(),

            self.read_obstacles(), self.read_tanks(), self.read_shells(), self.read_bonuses()

        )

 

    def read_players(self):

        player_count = self.read_int()

        if player_count < 0:

            return None

 

        players = []

 

        for player_index in xrange(player_count):

            if self.read_boolean():

                player = Player(self.read_string(), self.read_int(), self.read_boolean())

                players.append(player)

            else:

                players.append(None)

 

        return players

 

    def read_obstacles(self):

        obstacle_count = self.read_int()

        if obstacle_count < 0:

            return None

 

        obstacles = []

 

        for obstacle_index in xrange(obstacle_count):

            if self.read_boolean():

                obstacle = Obstacle(

                    self.read_long(), self.read_double(), self.read_double(),

                    self.read_double(), self.read_double()

                )

                obstacles.append(obstacle)

            else:

                obstacles.append(None)

 

        return obstacles

 

    def read_tanks(self):

        tank_count = self.read_int()

        if tank_count < 0:

            return None

 

        tanks = []

 

        for tank_index in xrange(tank_count):

            if self.read_boolean():

                tank = Tank(

                    self.read_long(), self.read_string(), self.read_int(),

                    self.read_double(), self.read_double(), self.read_double(), self.read_double(),

                    self.read_double(), self.read_double(), self.read_double(),

                    self.read_int(), self.read_int(), self.read_int(), self.read_int(),

                    self.read_int(), self.read_boolean(), self.read_enum(TankType)

                )

                tanks.append(tank)

            else:

                tanks.append(None)

 

        return tanks

 

    def read_shells(self):

        shell_count = self.read_int()

        if shell_count < 0:

            return None

 

        shells = []

 

        for shell_index in xrange(shell_count):

            if self.read_boolean():

                shell = Shell(

                    self.read_long(), self.read_string(), self.read_double(), self.read_double(),

                    self.read_double(), self.read_double(), self.read_double(), self.read_double(),

                    self.read_double(), self.read_double(), self.read_enum(ShellType)

                )

                shells.append(shell)

            else:

                shells.append(None)

 

        return shells

 

    def read_bonuses(self):

        bonus_count = self.read_int()

        if bonus_count < 0:

            return None

 

        bonuses = []

 

        for bonus_index in xrange(bonus_count):

            if self.read_boolean():

                bonus = Bonus(

                    self.read_long(), self.read_double(), self.read_double(),

                    self.read_double(), self.read_double(), self.read_enum(BonusType)

                )

                bonuses.append(bonus)

            else:

                bonuses.append(None)

 

        return bonuses

 

    def ensure_message_type(self, actual_type, expected_type):

        if actual_type != expected_type:

            raise ValueError("Received wrong message [actual=%s, expected=%s]" % (actual_type, expected_type))

 

    def read_enum(self, enum_class):

        bytes = self.read_bytes(RemoteProcessClient.SIGNED_BYTE_SIZE_BYTES)

        value = struct.unpack(RemoteProcessClient.SIGNED_BYTE_FORMAT_STRING, bytes)[0]

 

        for enum_key, enum_value in enum_class.__dict__.iteritems():

            if not str(enum_key).startswith("__") and value == enum_value:

                return enum_value

 

        return None

 

    def write_enum(self, value):

        self.write_bytes(struct.pack(RemoteProcessClient.SIGNED_BYTE_FORMAT_STRING, -1 if value is None else value))

 

    def read_string(self):

        length = self.read_int()

        if length == -1:

            return None

 

        bytes = self.read_bytes(length)

        return bytes.decode("utf-8")

 

    def write_string(self, value):

        if value is None:

            self.write_int(-1)

            return

 

        bytes = value.encode("utf-8")

 

        self.write_int(len(bytes))

        self.write_bytes(bytes)

 

    def read_boolean(self):

        bytes = self.read_bytes(RemoteProcessClient.SIGNED_BYTE_SIZE_BYTES)

        return struct.unpack(RemoteProcessClient.SIGNED_BYTE_FORMAT_STRING, bytes)[0] == 1

 

    def write_boolean(self, value):

        self.write_bytes(struct.pack(RemoteProcessClient.SIGNED_BYTE_FORMAT_STRING, 1 if value else 0))

 

    def read_int(self):

        bytes = self.read_bytes(RemoteProcessClient.INTEGER_SIZE_BYTES)

        return struct.unpack(RemoteProcessClient.INTEGER_FORMAT_STRING, bytes)[0]

 

    def write_int(self, value):

        self.write_bytes(struct.pack(RemoteProcessClient.INTEGER_FORMAT_STRING, value))

 

    def read_long(self):

        bytes = self.read_bytes(RemoteProcessClient.LONG_SIZE_BYTES)

        return struct.unpack(RemoteProcessClient.LONG_FORMAT_STRING, bytes)[0]

 

    def write_long(self, value):

        self.write_bytes(struct.pack(RemoteProcessClient.LONG_FORMAT_STRING, value))

 

    def read_double(self):

        bytes = self.read_bytes(RemoteProcessClient.DOUBLE_SIZE_BYTES)

        return struct.unpack(RemoteProcessClient.DOUBLE_FORMAT_STRING, bytes)[0]

 

    def write_double(self, value):

        self.write_bytes(struct.pack(RemoteProcessClient.DOUBLE_FORMAT_STRING, value))

 

    def read_bytes(self, byte_count):

        bytes = ''

 

        while len(bytes) < byte_count:

            chunk = self.socket.recv(byte_count - len(bytes))

 

            if not len(chunk):

                raise IOError("Can't read %s bytes from input stream." % str(byte_count))

 

            bytes += chunk

 

        return bytes

 

    def write_bytes(self, bytes):

        self.socket.sendall(bytes)

 

    class MessageType:

        UNKNOWN = 0

        GAME_OVER = 1

        AUTHENTICATION_TOKEN = 2

        TEAM_SIZE = 3

        TANK_TYPES = 4