Feature: fps & speed control

src/interpolation/Interpolation.py

@@ -0,0 +1,61 @@
+from renderer.data import ReplayData, Events
+from interpolation.interpolator.liner import LinerInterpolationer
+import math
+import copy
+import logging
+import collections
+from tqdm import tqdm
+
+class Interpolator():
+    def interpolate(self, 
+            replay_info: ReplayData,
+            fpsTarget=60,
+            speedScale=6.0,
+            method="liner"
+            ):
+        logging.info("started interpolating")
+        evtList=list(replay_info.events.values())
+        # add more frames in replays to support higher fps
+        # by default, one frame in game represents 1 second
+        origTotLen=len(evtList)
+        # number of events in one sec
+        framePerEvt=fpsTarget/speedScale   
+
+
+        # total events 
+        afterLen=int((origTotLen - 1)/framePerEvt) + 1   
+        newRepInfo=[]
+        for i in tqdm(range(origTotLen)):
+            currentLenTarget=framePerEvt*(i+1)
+            # in case speed scale is faster than 60 events per sec
+            if currentLenTarget <= len(newRepInfo):
+                continue
+
+            originEvt=evtList[i]
+            newRepInfo.append(originEvt)
+            if originEvt.last_frame or i+1 >= origTotLen:
+                break
+            nextEvt=evtList[i+1] 
+
+            eventsToAdd=math.floor(currentLenTarget - len(newRepInfo))
+
+            match method:
+                case "liner":
+                    # liner implementation
+                    for item in (LinerInterpolationer().doLinerInterpolation(originEvt, nextEvt, eventsToAdd)):
+                        newRepInfo.append(item)
+                case "bezier":
+                    # todo
+                    raise NotImplementedError("not implemented")
+        logging.info("interpolation complete")  
+        tmp = replay_info._asdict()
+        tmp['events']=dict(zip(range(len(newRepInfo)), newRepInfo))
+        return toNamedTuple('ReplayData', tmp)
+
+def toNamedTuple(name: str, evt: dict):
+        keyList=evt.keys()
+        valueList=evt.values()
+        tmpTuple=collections.namedtuple(name, keyList)
+        return tmpTuple._make(valueList)
+
+

src/interpolation/interpolator/liner.py

@@ -0,0 +1,291 @@
+from typing import NamedTuple, Optional
+from renderer.data import (
+    PlayerInfo,
+    Vehicle,
+    ReplayData,
+    Events,
+    Smoke,
+    Shot,
+    Torpedo,
+    Consumable,
+    Plane,
+    Ward,
+    ControlPoint,
+    Score,
+    Frag,
+    Message,
+    BattleResult,
+    BuildingInfo,
+    Building,
+    Units,
+    Skills,
+    AcousticTorpedo,
+)
+import math
+import copy
+import logging
+import collections
+
+class VehicleDiff(NamedTuple):
+    x: float
+    y: float
+    yaw: float
+    regenerated_health: float
+    regen_crew_hp_limit: float
+    regeneration_health: float
+
+class PlaneDiff(NamedTuple):
+    x: float
+    y: float
+
+class ShotDiff(NamedTuple):
+    origX: float
+    origY: float
+    destX: float
+    destY: float
+
+class TorpedoDiff(NamedTuple):
+    x: float
+    y: float
+    yaw: float
+
+class AcousticTorpedoDiff(NamedTuple):
+    x: float
+    y: float
+    yaw: float      
+
+# dict key is id of elements（vehicle_id, plane_id, etc.).
+class LinerDiff(NamedTuple):
+       vehicleDiff: dict[int, VehicleDiff]
+       planeDiff: dict[int, PlaneDiff]
+       shotDiff: dict[int, ShotDiff]
+       torpedoDiff: dict[int, TorpedoDiff]
+       acousticTorpedoDiff: dict[int, AcousticTorpedoDiff]
+
+class LinerInterpolationer:
+
+    def __init__(self) -> None:
+        pass
+
+    def doLinerInterpolation(self, originEvt: Events, nextEvt: Events, eventsToAdd: int):
+        result = []
+        if eventsToAdd <= 0 :
+            return result
+        diffData=self.calcDiff(originEvt, nextEvt, eventsToAdd)
+        currentEvt = self.initCurrentEvt(originEvt)
+        for i in range(eventsToAdd): 
+            tmpEvt = currentEvt._asdict()
+            self.applyDiffData(diffData, tmpEvt)
+            currentEvt = toNamedTuple('Events', tmpEvt)
+            result.append(currentEvt)
+        return result
+
+    def initCurrentEvt(self, originEvt: Events):
+        currentEvt = Events(
+            time_left=originEvt.time_left,
+            evt_vehicle=originEvt.evt_vehicle,
+            evt_plane=originEvt.evt_plane,
+            evt_shot=[],
+            evt_torpedo=originEvt.evt_torpedo,
+            evt_acoustic_torpedo=originEvt.evt_acoustic_torpedo,
+            evt_building = dict(),
+            evt_ward = originEvt.evt_ward,
+            evt_smoke = originEvt.evt_smoke,
+            evt_hits = [],
+            evt_consumable = dict(),
+            evt_control = originEvt.evt_control,
+            evt_score = originEvt.evt_score,
+            evt_damage_maps = originEvt.evt_damage_maps,
+            evt_frag = [],
+            evt_ribbon = originEvt.evt_ribbon,
+            evt_achievement = originEvt.evt_achievement,
+            evt_times_to_win = originEvt.evt_times_to_win,
+            evt_chat = [],
+        )
+        return currentEvt
+
+    def calcDiff(self, originEvt: Events, nextEvt: Events, eventsToAdd: int):
+        result=LinerDiff(
+            vehicleDiff=self.calcVehicle(originEvt, nextEvt, eventsToAdd),
+            planeDiff=self.calcPlaneDiff(originEvt, nextEvt, eventsToAdd),
+            shotDiff=self.calcShotDiff(originEvt, nextEvt, eventsToAdd),
+            torpedoDiff=self.calcTorpedoDiff(originEvt, nextEvt, eventsToAdd),
+            acousticTorpedoDiff=self.calcAcousticTorpedoDiff(originEvt, nextEvt, eventsToAdd)
+        )
+        return result
+
+    def calcVehicle(self, originEvt: Events, nextEvt: Events, eventsToAdd: int):
+        result=dict()
+        if eventsToAdd <= 0:
+            return result
+        for id, vehicle in originEvt.evt_vehicle.items():
+            nextVehicle=nextEvt.evt_vehicle.get(id)
+            if nextVehicle is None: 
+                continue
+            result[id]=VehicleDiff(
+                float(nextVehicle.x - vehicle.x)/ float(eventsToAdd),
+                float(nextVehicle.y - vehicle.y)/ float(eventsToAdd),
+                (nextVehicle.yaw - vehicle.yaw )/ eventsToAdd,
+                (nextVehicle.regenerated_health - vehicle.regenerated_health)/ eventsToAdd,
+                (nextVehicle.regen_crew_hp_limit - vehicle.regen_crew_hp_limit)/ eventsToAdd,
+                (nextVehicle.regeneration_health - vehicle.regeneration_health)/ eventsToAdd
+            )
+
+        return result
+
+    def calcPlaneDiff(self, originEvt: Events, nextEvt: Events, eventsToAdd: int):
+        result=dict()
+        if eventsToAdd <= 0:
+            return result
+        for id, plane in originEvt.evt_plane.items():
+            nextPlane=nextEvt.evt_plane.get(id)
+            if nextPlane is None: 
+                continue
+            result[id]=PlaneDiff(
+                float(nextPlane.position[0] - plane.position[0])/ float(eventsToAdd),
+                float(nextPlane.position[1] - plane.position[1])/ float(eventsToAdd)
+            )
+
+        return result
+
+    def calcShotDiff(self, originEvt: Events, nextEvt: Events, eventsToAdd: int):
+        result=dict()
+        if eventsToAdd <= 0:
+            return result
+        for shot in originEvt.evt_shot:
+            id=shot.shot_id
+
+            nextShot=None
+            for tmp in nextEvt.evt_shot:
+                if tmp.shot_id==id:
+                    nextShot=tmp
+                    break
+            if nextShot is None: 
+                continue
+            result[id]=ShotDiff(
+                float(nextShot.origin[0] - shot.origin[0])/ float(eventsToAdd),
+                float(nextShot.origin[1] - shot.origin[1])/ float(eventsToAdd),
+                float(nextShot.destination[0] - shot.destination[0])/ float(eventsToAdd),
+                float(nextShot.destination[1] - shot.destination[1])/ float(eventsToAdd),
+            )
+
+        return result
+
+    def calcTorpedoDiff(self, originEvt: Events, nextEvt: Events, eventsToAdd: int):
+        result=dict()
+        if eventsToAdd <= 0:
+            return result
+        for id, torpedo in originEvt.evt_torpedo.items():
+            nextTorpedo=nextEvt.evt_torpedo.get(id)
+            if nextTorpedo is None: 
+                continue
+            result[id]=TorpedoDiff(
+                float(nextTorpedo.x - torpedo.x)/ float(eventsToAdd),
+                float(nextTorpedo.y - torpedo.y)/ float(eventsToAdd),
+                (nextTorpedo.yaw - torpedo.yaw )/ eventsToAdd,
+           )
+
+        return result
+
+    def calcAcousticTorpedoDiff(self, originEvt: Events, nextEvt: Events, eventsToAdd: int):
+        result=dict()
+        if eventsToAdd <= 0:
+            return result
+        for id, acousticTorpedo in originEvt.evt_acoustic_torpedo.items():
+            nextAcousticTorpedo=nextEvt.evt_acoustic_torpedo.get(id)
+            if nextAcousticTorpedo is None: 
+                continue
+            result[id]=TorpedoDiff(
+                float(nextAcousticTorpedo.x - acousticTorpedo.x)/ float(eventsToAdd),
+                float(nextAcousticTorpedo.y - acousticTorpedo.y)/ float(eventsToAdd),
+                (nextAcousticTorpedo.yaw - acousticTorpedo.yaw )/ eventsToAdd,
+           )
+
+        return result
+
+
+    def applyDiffData(self, diffData: LinerDiff, currentEvt: dict):
+        tmp=[]
+        for id, vehicle in currentEvt['evt_vehicle'].items():
+            vehicle: Vehicle
+            diffVehicle = diffData.vehicleDiff.get(id)
+            diffVehicle: VehicleDiff
+
+            if diffVehicle is None:
+                logging.debug("diffVehicle: id %s is none.", vehicle.vehicle_id)
+                continue
+            dic = vehicle._asdict()
+            dic['x']=dic['x'] + diffVehicle.x
+            dic['y']=dic['y'] + diffVehicle.y
+            dic['yaw']=dic['yaw'] + diffVehicle.yaw
+            dic['regenerated_health']=dic['regenerated_health']+diffVehicle.regenerated_health
+            dic['regen_crew_hp_limit']=dic['regen_crew_hp_limit']+diffVehicle.regen_crew_hp_limit
+            dic['regeneration_health']=dic['regeneration_health']+diffVehicle.regeneration_health
+
+            currentEvt['evt_vehicle'][id]=toNamedTuple('Vehicle', dic)
+
+        for id, plane in currentEvt['evt_plane'].items():
+            plane: Plane
+            diffPlane = diffData.planeDiff.get(id)
+            diffPlane: PlaneDiff
+
+            if diffPlane is None:
+                logging.debug("diffPlane: id %s is none.", id)
+                continue
+            dic=plane._asdict()
+            dic['position']=tuple([
+                dic['position'][0]+diffPlane.x,
+                dic['position'][1]+diffPlane.y]
+            )
+
+            currentEvt['evt_plane'][id]=toNamedTuple('Plane', dic)
+        for i in range(len(currentEvt['evt_shot'])):
+            shot = currentEvt['evt_shot'][i]
+            id = shot.shot_id
+            shot: Shot
+            diffShot = diffData.shotDiff.get(id)
+            diffShot: ShotDiff
+
+            if diffShot is None:
+                logging.debug("diffShot: id %s is none.", id)
+                continue 
+            dic=shot._asdict()
+            origOrigin=dic['origin']
+            origDest=dic['destnation']
+            dic['origin']=tuple([origOrigin[0]+diffShot.origX,origOrigin[1]+diffShot.origY])
+            dic['destnation']=tuple([origDest[0]+diffShot.destX,origDest[1]+diffShot.destY])
+            currentEvt['evt_shot'][i]=toNamedTuple('Shot', dic)
+        for id, torpedo in currentEvt['evt_torpedo'].items():
+            torpedo: Torpedo
+            diffTorpedo = diffData.torpedoDiff.get(id)
+            diffTorpedo: TorpedoDiff
+
+            if diffTorpedo is None:
+                logging.debug("diffTorpedo: id %s is none.", id)
+                continue 
+            dic=torpedo._asdict()
+            origOrigin=dic['origin']
+            dic['origin']=tuple(origOrigin[0]+diffTorpedo.x,origOrigin[1]+diffTorpedo.y)
+            dic['yaw']=dic['yaw']+diffTorpedo.yaw
+            currentEvt['evt_torpedo'][id]=toNamedTuple('Torpedo', dic)
+
+        for id, acoustic_torpedo in currentEvt['evt_acoustic_torpedo'].items():
+            acoustic_torpedo: Torpedo
+            diffAcousticTorpedo = diffData.acousticTorpedoDiff.get(id)
+            diffAcousticTorpedo: AcousticTorpedoDiff
+
+            if diffAcousticTorpedo is None:
+                logging.debug("diffAcousticTorpedo: id %s is none.", id)
+                continue 
+            dic=acoustic_torpedo._asdict()
+            dic['x']=dic['x']+diffAcousticTorpedo.x
+            dic['y']=dic['y']+diffAcousticTorpedo.y
+            dic['yaw']=dic['yaw']+diffAcousticTorpedo.yaw
+            currentEvt['evt_acoustic_torpedo'][id]=toNamedTuple('AcousticTorpedo', dic)
+
+
+def toNamedTuple(name: str, evt: dict):
+        keyList=evt.keys()
+        valueList=evt.values()
+        tmpTuple=collections.namedtuple(name, keyList)
+        return tmpTuple._make(valueList)