Coverage for quality/utils.py: 100%

1import contextlib 

2from typing import TypedDict 

3 

4import pendulum 

5import requests 

6from django.conf import settings 

7 

8 

9class Filter(TypedDict): 

10 year: int | None 

11 triggered: bool | None 

12 network: list[str] | None 

13 station: list[str] | None 

14 channel: list[str] | None 

15 operator: list[str] | None 

16 

17 

18def get_rtserved_stations(network: str) -> list[str]: 

19 url = f"{settings.RT_SERVER_URL}/streamids" 

20 with contextlib.suppress(requests.ConnectionError): 

21 response = requests.get(url, timeout=0.5, params={"level": 2}) 

22 if response.ok: 

23 return [ 

24 line.split("_")[1] 

25 for line in response.text.splitlines() 

26 if line.split("_")[0] == network 

27 ] 

28 return [] 

29 

30 

31def get_station_last_rt_data(network: str, station: str) -> dict[str, object]: 

32 url = f"{settings.RT_SERVER_URL}/streams" 

33 with contextlib.suppress(requests.ConnectionError): 

34 response = requests.get(url, timeout=0.5) 

35 if response.ok: 

36 return { 

37 line.split(" ")[0].split("/")[0].replace("_", "."): pendulum.parse( 

38 line.split(" ")[-1] 

39 ) 

40 for line in response.text.splitlines() 

41 if line.split("_")[0:2] == [network, station] 

42 } 

43 return {}