12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849 |
- from operator import attrgetter
- from typing import Dict, List
- from httpx import AsyncClient
- from loguru import logger
- from app.schemas.bluetooth import IBeaconBase
- from app.services.platform import DataPlatformService
- class SpacePositioningController:
- def __init__(self, realtime_ibeacon_list: List[IBeaconBase], ibeacon_space_dict: Dict):
- super(SpacePositioningController, self).__init__()
- self.realtime_ibeacon_list = realtime_ibeacon_list
- self.ibeacon_space_dict = ibeacon_space_dict
- def run(self) -> str:
- ibeacon_list = sorted(self.realtime_ibeacon_list, key=attrgetter('rssi'), reverse=True)
- strongest_id = ''
- for ibeacon in ibeacon_list:
- if ibeacon.rssi != 0:
- strongest = ibeacon
- strongest_id = str(strongest.major) + '-' + str(strongest.minor)
- break
- return self.ibeacon_space_dict.get(strongest_id)
- @logger.catch()
- async def get_space_location(project_id: str, realtime_ibeacon_list: List[IBeaconBase]) -> str:
- async with AsyncClient() as client:
- platform = DataPlatformService(client, project_id)
- spaces = await platform.get_items_by_category('Sp')
- ibeacon_space_dict = dict()
- for sp in spaces:
- try:
- major = str(sp['infos']['ctm-iBeacon-Major'])
- minor = str(sp['infos']['ctm-iBeacon-Minor'])
- if len(major) > 0 and len(minor) > 0:
- k = major + '-' + minor
- v = sp['id']
- ibeacon_space_dict.update({k: v})
- except KeyError:
- pass
- controller = SpacePositioningController(realtime_ibeacon_list, ibeacon_space_dict)
- return controller.run()
|