Coverage for mobius_adapter/utils.py: 80%

1from datetime import datetime 

2 

3import requests 

4from beanie import PydanticObjectId 

5from pydantic import BaseModel, Field 

6 

7from config import config 

8 

9 

10class Reading(BaseModel): 

11 id: PydanticObjectId 

12 node: PydanticObjectId 

13 canID: int = Field(default=..., lt=5, gt=0) 

14 sensorNumber: int = Field(default=..., lt=3, gt=0) 

15 readingID: int 

16 sessionID: int 

17 dangerLevel: int = 0 

18 window1: int = 0 

19 window2: int = 0 

20 window3: int = 0 

21 publishedAt: datetime 

22 

23 

24# Conversione reading per mobius 

25def to_mobius_payload(reading: Reading, sensorId: str) -> dict: 

26 return { 

27 "m2m:cin": { 

28 "con": { 

29 "metadata": { 

30 "sensorId": sensorId, 

31 "readingTimestamp": datetime.fromtimestamp( 

32 reading.readingID 

33 ).isoformat(), 

34 # "latitude": <latitudine del sensore>, // opzionale 

35 # "longitude": <longitudine del sensore>, // opzionale 

36 # "heading": <orientazione del sensore>, // opzionale 

37 } 

38 }, 

39 "sensorData": { 

40 "canID": reading.canID, 

41 "sensorNumber": reading.sensorNumber, 

42 "dangerLevel": reading.dangerLevel, 

43 "window1Count": reading.window1, 

44 "window2Count": reading.window2, 

45 "window3Count": reading.window3, 

46 }, 

47 } 

48 } 

49 

50 

51def insert(reading: Reading, nodeID: int): 

52 sensorId = config.nodeID_to_sensorId(nodeID) 

53 sensorPath = config.nodeID_to_sensorPath(nodeID) 

54 originator = config.originator 

55 

56 mobius_payload: dict = to_mobius_payload(reading, sensorId) 

57 

58 requests.post( 

59 f"{config.host}:{config.port}/{sensorPath}", 

60 headers={ 

61 "X-M2M-Origin": originator, 

62 "Content-Type": "application/vnd.onem2m-res+json;ty=4", 

63 "X-M2M-RI": str(int(datetime.now().timestamp() * 1000)), 

64 }, 

65 json=mobius_payload, 

66 )