space.py 1.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  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__(
  9. self, realtime_ibeacon_list: List[IBeaconBase], ibeacon_space_dict: Dict
  10. ):
  11. super(SpacePositioningController, self).__init__()
  12. self.realtime_ibeacon_list = realtime_ibeacon_list
  13. self.ibeacon_space_dict = ibeacon_space_dict
  14. def run(self) -> str:
  15. ibeacon_list = sorted(
  16. self.realtime_ibeacon_list, key=attrgetter("rssi"), reverse=True
  17. )
  18. space = ""
  19. for ibeacon in ibeacon_list:
  20. if ibeacon.rssi != 0:
  21. strongest = ibeacon
  22. logger.debug(strongest)
  23. strongest_id = str(strongest.major) + "-" + str(strongest.minor)
  24. space = self.ibeacon_space_dict.get(strongest_id)
  25. if space:
  26. logger.debug(strongest_id)
  27. break
  28. else:
  29. space = ""
  30. return space
  31. @logger.catch()
  32. async def get_space_location(
  33. project_id: str, realtime_ibeacon_list: List[IBeaconBase]
  34. ) -> str:
  35. async with AsyncClient() as client:
  36. platform = DataPlatformService(client, project_id)
  37. spaces = await platform.get_items_by_category("Sp")
  38. ibeacon_space_dict = dict()
  39. for sp in spaces:
  40. try:
  41. major = str(sp["infos"]["ctm-iBeacon-Major"])
  42. minor = str(sp["infos"]["ctm-iBeacon-Minor"])
  43. if len(major) > 0 and len(minor) > 0:
  44. k = major + "-" + minor
  45. v = sp["id"]
  46. ibeacon_space_dict.update({k: v})
  47. except KeyError:
  48. pass
  49. logger.debug(ibeacon_space_dict)
  50. controller = SpacePositioningController(realtime_ibeacon_list, ibeacon_space_dict)
  51. return controller.run()