WIP: Adjust ReceiveProtosEndpoint.py and SerializedMitmDataProcessor.…

…py for raw processing

mapadroid/data_handler/mitm_data/PlayerData.py

@@ -8,6 +8,7 @@
 from mapadroid.utils.ProtoIdentifier import ProtoIdentifier
 from mapadroid.utils.collections import Location
 from mapadroid.utils.logging import LoggerEnums, get_logger
+import mapadroid.mitm_receiver.protos.Rpc_pb2 as pogoprotos
 
 logger = get_logger(LoggerEnums.mitm_mapper)
 
@@ -62,26 +63,28 @@ def get_specific_latest_data(self, key: Union[int, str],
     def get_full_latest_data(self) -> Dict[Union[int, str], LatestMitmDataEntry]:
         return self._latest_data_holder.get_all()
 
-    def update_latest(self, key: str, value: Optional[Union[List, Dict]],
+    def update_latest(self, key: str, value: Union[List, Dict, bytes],
                       timestamp_received: Optional[int] = None,
                       timestamp_of_data_retrieval: Optional[int] = None,
                       location: Optional[Location] = None) -> None:
         self._latest_data_holder.update(key, value, timestamp_received, timestamp_of_data_retrieval, location)
-        if key == str(ProtoIdentifier.GMO.value):
-            self.__parse_gmo_for_location(value, timestamp_received, location)
+        if key == str(ProtoIdentifier.GMO.value) and isinstance(value, bytes):
+            gmo: pogoprotos.GetMapObjectsOutProto = pogoprotos.GetMapObjectsOutProto.ParseFromString(
+                value)
+            self.__parse_gmo_for_location(gmo, timestamp_received, location)
             self._injected = True
+        else:
+            logger.warning("update_latest not of GMO type")
 
     # Async since we may move it to DB for persistence, same for above methods like level and
     # pokestops visited (today/week/total/whatever)
     async def get_last_possibly_moved(self) -> int:
         return self.__last_possibly_moved
 
-    # TODO: Call it from within update_latest accordingly rather than externally...
-    def __parse_gmo_for_location(self, gmo_payload: Dict, timestamp: int, location: Optional[Location]):
-        cells = gmo_payload.get("cells", None)
-        if not cells:
+    def __parse_gmo_for_location(self, gmo_payload: pogoprotos.GetMapObjectsOutProto, timestamp: int, location: Optional[Location]):
+        if not gmo_payload.map_cell:
             return
-        cell_ids: List[int] = [cell['id'] for cell in cells]
+        cell_ids: List[int] = [cell.s2_cell_id for cell in gmo_payload.map_cell]
         if not bool(set(cell_ids).intersection(self.__last_cell_ids)):
             self.__last_cell_ids = cell_ids
             self.__last_possibly_moved = timestamp

mapadroid/data_handler/mitm_data/RedisMitmMapper.py

@@ -68,22 +68,15 @@ async def update_latest(self, worker: str, key: str, value: Union[List, Dict, by
                 await self.__cache.set(RedisMitmMapper.LATEST_DATA_KEY.format(worker, key), json_data)
             except Exception as e:
                 logger.exception(e)
-        if key == str(ProtoIdentifier.GMO.value):
-            await self.__parse_gmo_for_location(worker, value, timestamp_received_raw, location)
+        if key == str(ProtoIdentifier.GMO.value) and isinstance(value, bytes):
+            gmo: pogoprotos.GetMapObjectsOutProto = pogoprotos.GetMapObjectsOutProto.ParseFromString(
+                value)
+            await self.__parse_gmo_for_location(worker, gmo, timestamp_received_raw, location)
             await self.__cache.set(RedisMitmMapper.IS_INJECTED_KEY.format(worker), 1)
 
-    async def __parse_gmo_for_location(self, worker: str, gmo_payload: Union[Dict, bytes], timestamp: int,
+    async def __parse_gmo_for_location(self, worker: str, gmo_proto: pogoprotos.GetMapObjectsOutProto, timestamp: int,
                                        location: Optional[Location]):
-        if isinstance(gmo_payload, dict):
-            cells = gmo_payload.get("cells", None)
-            if not cells:
-                return
-            cell_ids: List[int] = [cell['id'] for cell in cells]
-        else:
-            # Raw proto...
-            gmo_proto: pogoprotos.GetMapObjectsOutProto = pogoprotos.GetMapObjectsOutProto.ParseFromString(
-                gmo_payload)
-            cell_ids: List[int] = [cell.s2_cell_id for cell in gmo_proto.map_cell]
+        cell_ids: List[int] = [cell.s2_cell_id for cell in gmo_proto.map_cell]
         last_cell_ids_raw: Optional[str] = await self.__cache.get(RedisMitmMapper.LAST_CELL_IDS_KEY.format(worker))
         last_cell_ids: List[int] = []
         if last_cell_ids_raw:

mapadroid/db/DbWrapper.py

@@ -6,7 +6,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from mapadroid.db.DbAccessor import DbAccessor
-from mapadroid.db.DbPogoProtoSubmit import DbPogoProtoSubmit
+from mapadroid.db.DbPogoProtoSubmitRaw import DbPogoProtoSubmitRaw
 from mapadroid.db.helper.MadminInstanceHelper import MadminInstanceHelper
 from mapadroid.db.helper.SettingsAreaHelper import SettingsAreaHelper
 from mapadroid.db.helper.SettingsAreaIdleHelper import SettingsAreaIdleHelper
@@ -41,7 +41,7 @@ def __init__(self, db_exec: PooledQueryExecutor, args):
         # self.sanity_check.check_all()
         # self.supports_apks = self.sanity_check.supports_apks
 
-        self.proto_submit: DbPogoProtoSubmit = DbPogoProtoSubmit(db_exec, args)
+        self.proto_submit: DbPogoProtoSubmitRaw = DbPogoProtoSubmitRaw(db_exec)
         self.__instance_id: Optional[int] = None
 
     async def setup(self) -> None:

mapadroid/db/helper/TrsS2CellHelper.py

@@ -8,6 +8,7 @@
 from mapadroid.db.model import TrsS2Cell
 from mapadroid.utils.collections import Location
 from mapadroid.utils.s2Helper import S2Helper
+import mapadroid.mitm_receiver.protos.Rpc_pb2 as pogoprotos
 
 
 class TrsS2CellHelper:
@@ -18,17 +19,16 @@ async def get(session: AsyncSession, cell_id: int) -> Optional[TrsS2Cell]:
         return result.scalars().first()
 
     @staticmethod
-    async def insert_update_cell(session: AsyncSession, cell: dict) -> None:
-        cell_id = cell["id"]
-        if cell_id < 0:
-            cell_id = cell_id + 2 ** 64
-        lat, lng, _ = S2Helper.get_position_from_cell(cell_id)
+    async def insert_update_cell(session: AsyncSession, cell: pogoprotos.ClientMapCellProto) -> None:
+        if cell.s2_cell_id < 0:
+            cell.s2_cell_id = cell.s2_cell_id + 2 ** 64
+        lat, lng, _ = S2Helper.get_position_from_cell(cell.s2_cell_id)
         insert_stmt = insert(TrsS2Cell).values(
-            id=str(cell_id),
+            id=str(cell.s2_cell_id),
             level=15,
             center_latitude=lat,
             center_longitude=lng,
-            updated=int(cell["current_timestamp"] / 1000)
+            updated=int(cell.as_of_time_ms / 1000)
         )
         on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
             updated=insert_stmt.inserted.updated