space.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
  1. from operator import attrgetter
  2. from typing import Dict, List
  3. from httpx import AsyncClient
  4. from loguru import logger
  5. from app.schemas.bluetooth import IBeaconBase
  6. from app.services.platform import DataPlatformService
  7. class SpacePositioningController:
  8. def __init__(self, realtime_ibeacon_list: List[IBeaconBase], ibeacon_space_dict: Dict):
  9. super(SpacePositioningController, self).__init__()
  10. self.realtime_ibeacon_list = realtime_ibeacon_list
  11. self.ibeacon_space_dict = ibeacon_space_dict
  12. def run(self) -> str:
  13. ibeacon_list = sorted(self.realtime_ibeacon_list, key=attrgetter('rssi'), reverse=True)
  14. strongest_id = ''
  15. for ibeacon in ibeacon_list:
  16. if ibeacon.rssi != 0:
  17. strongest = ibeacon
  18. strongest_id = str(strongest.major) + '-' + str(strongest.minor)
  19. break
  20. return self.ibeacon_space_dict.get(strongest_id)
  21. @logger.catch()
  22. async def get_space_location(project_id: str, realtime_ibeacon_list: List[IBeaconBase]) -> str:
  23. async with AsyncClient() as client:
  24. platform = DataPlatformService(client, project_id)
  25. spaces = await platform.get_items_by_category('Sp')
  26. ibeacon_space_dict = dict()
  27. for sp in spaces:
  28. try:
  29. major = str(sp['infos']['ctm-iBeacon-Major'])
  30. minor = str(sp['infos']['ctm-iBeacon-Minor'])
  31. if len(major) > 0 and len(minor) > 0:
  32. k = major + '-' + minor
  33. v = sp['id']
  34. ibeacon_space_dict.update({k: v})
  35. except KeyError:
  36. pass
  37. controller = SpacePositioningController(realtime_ibeacon_list, ibeacon_space_dict)
  38. return controller.run()