1212
1313LOG = logging .getLogger (__name__ )
1414
15- CLIENT_ID = os .getenv ("SWISSCOM_CLIENT_ID" , "" ) # customer key in the Swisscom digital marketplace
16- CLIENT_SECRET = os .getenv ("SWISSCOM_CLIENT_SECRET" , "" ) # customer secret in the Swisscom digital marketplace
15+ CLIENT_ID = os .getenv (
16+ "SWISSCOM_CLIENT_ID" , ""
17+ ) # customer key in the Swisscom digital marketplace
18+ CLIENT_SECRET = os .getenv (
19+ "SWISSCOM_CLIENT_SECRET" , ""
20+ ) # customer secret in the Swisscom digital marketplace
1721MIN_DATE = os .getenv ("MIN_DATE" , "03.10.2022" )
1822MAX_DATE = os .getenv ("MAX_DATE" , "16.10.2022" )
1923MAX_NB_TILES_REQUEST = int (os .getenv ("MAX_NB_TILES_REQUEST" , "100" ))
@@ -47,18 +51,24 @@ def auth(self) -> OAuth2Session:
4751 # Fetch an access token
4852 client = BackendApplicationClient (client_id = CLIENT_ID )
4953 oauth = OAuth2Session (client = client )
50- oauth .fetch_token (token_url = TKN_URL , client_id = CLIENT_ID , client_secret = CLIENT_SECRET )
54+ oauth .fetch_token (
55+ token_url = TKN_URL , client_id = CLIENT_ID , client_secret = CLIENT_SECRET
56+ )
5157 return oauth
5258
5359 def get_tiles_ids (self , oauth : OAuth2Session , postal_code : int ) -> list [int ]:
5460 # For muni/district id, see https://www.atlas.bfs.admin.ch/maps/13/fr/17804_229_228_227/27579.html
5561 # Municipalities and Districts doesn't work well probably because of the free plan
5662 # Get all the first MAX_NB_TILES_REQUEST tile ids associated with the postal code of interest
57- muni_tiles_json = oauth .get (BASE_URL + f"/grids/postal-code-areas/{ postal_code } " , headers = HEADERS )
63+ muni_tiles_json = oauth .get (
64+ BASE_URL + f"/grids/postal-code-areas/{ postal_code } " , headers = HEADERS
65+ )
5866 self .check_api_error (muni_tiles_json )
5967 tiles = muni_tiles_json .json ()["tiles" ]
6068 LOG .info ("Nb tiles received: %s" , len (tiles ))
61- return [t ["tileId" ] for t in muni_tiles_json .json ()["tiles" ]][:MAX_NB_TILES_REQUEST ]
69+ return [t ["tileId" ] for t in muni_tiles_json .json ()["tiles" ]][
70+ :MAX_NB_TILES_REQUEST
71+ ]
6272
6373 def query_api_generic (
6474 self , oauth : OAuth2Session , path : str , postal_code : int , date_time : datetime
@@ -79,24 +89,32 @@ def response_to_geojson_result(self, data: dict[str, Any]) -> FeatureCollection:
7989 features .append (Feature (geometry = Point (coordinate ), properties = element ))
8090 return FeatureCollection (features )
8191
82- def get_dwell_density (self , postal_code : int , date_time : datetime ) -> FeatureCollection | Response :
92+ def get_dwell_density (
93+ self , postal_code : int , date_time : datetime
94+ ) -> FeatureCollection | Response :
8395 self .error = None
8496 try :
8597 self .limit_query ()
8698 oauth = self .auth ()
87- api_request = self .query_api_generic (oauth , "/dwell-density/hourly" , postal_code , date_time )
99+ api_request = self .query_api_generic (
100+ oauth , "/dwell-density/hourly" , postal_code , date_time
101+ )
88102 response = oauth .get (api_request , headers = HEADERS )
89103 self .check_api_error (response )
90104 except (ExternalAPIError , APIUsageExceededError ):
91105 return self .error
92106 return self .response_to_geojson_result (response .json ())
93107
94- def get_dwell_demographics (self , postal_code : int , date_time : datetime ) -> FeatureCollection | Response :
108+ def get_dwell_demographics (
109+ self , postal_code : int , date_time : datetime
110+ ) -> FeatureCollection | Response :
95111 self .error = None
96112 try :
97113 self .limit_query ()
98114 oauth = self .auth ()
99- api_request = self .query_api_generic (oauth , "/dwell-demographics/hourly" , postal_code , date_time )
115+ api_request = self .query_api_generic (
116+ oauth , "/dwell-demographics/hourly" , postal_code , date_time
117+ )
100118 response = oauth .get (api_request , headers = HEADERS )
101119 self .check_api_error (response )
102120 except (ExternalAPIError , APIUsageExceededError ):
0 commit comments