@@ -27,22 +27,18 @@ def __exit__(self, e_t, e_v, e_tr):
2727
2828
2929# Open a file or a URL and return a File-like object for reading
30- def open_url (url : str , username = None , password = None ) -> io .BufferedReader :
31- if url .split (":" , 1 )[0 ] not in ("http" , "https" ):
32- return open (url , "rb" )
33- else :
30+ def open_url (url_or_filename : str , ** kw ) -> io .BufferedReader :
31+ if url_or_filename .split (":" , 1 )[0 ] in ("http" , "https" ):
3432 import requests
3533
36- if username and password :
37- r = requests .get (url , auth = (username , password ))
38- else :
39- r = requests .get (url )
40-
34+ r = requests .get (url_or_filename , ** kw )
4135 code : int = r .status_code
4236 if code != 200 :
4337 r .close ()
4438 raise ValueError (f"HTTP Error: { code } " )
4539 return SocketWrapper (r .raw ) # type: ignore
40+ else :
41+ return open (url_or_filename , "rb" )
4642
4743
4844# OTA manages a MicroPython firmware update over-the-air. It checks that there
@@ -110,24 +106,25 @@ def from_stream(self, f: io.BufferedReader, sha: str = "", length: int = 0) -> i
110106 # - url: a filename or a http[s] url for the micropython.bin firmware.
111107 # - sha: the sha256sum of the firmware file
112108 # - length: the length (in bytes) of the firmware file
113- def from_firmware_file (self , url : str , sha : str = "" , length : int = 0 , username = None , password = None ) -> int :
109+ def from_firmware_file (self , url : str , sha : str = "" , length : int = 0 , ** kw ) -> int :
114110 if self .verbose :
115111 print (f"Opening firmware file { url } ..." )
116- with open_url (url , username , password ) as f :
112+ with open_url (url , ** kw ) as f :
117113 return self .from_stream (f , sha , length )
118114
119115 # Load a firmware file, the location of which is read from a json file
120116 # containing the url for the firmware file, the sha and length of the file.
121117 # - url: the name of a file or url containing the json.
122- def from_json (self , url : str ) -> int :
118+ # - kw: extra keywords arguments that will be passed to `requests.get()`
119+ def from_json (self , url : str , ** kw ) -> int :
123120 if not url .endswith (".json" ):
124121 raise ValueError ("Url does not end with '.json'" )
125122 if self .verbose :
126123 print (f"Opening json file { url } ..." )
127- with open_url (url ) as f :
128- import json
124+ with open_url (url , ** kw ) as f :
125+ from json import load
129126
130- data : dict = json . load (f )
127+ data : dict = load (f )
131128 try :
132129 firmware : str = data ["firmware" ]
133130 sha : str = data ["sha" ]
@@ -136,20 +133,20 @@ def from_json(self, url: str) -> int:
136133 # If firmware filename is relative, append to base of url of json file
137134 baseurl , * _ = url .rsplit ("/" , 1 )
138135 firmware = f"{ baseurl } /{ firmware } "
139- return self .from_firmware_file (firmware , sha , length )
136+ return self .from_firmware_file (firmware , sha , length , ** kw )
140137 except KeyError as err :
141138 print ('OTA json must include "firmware", "sha" and "length" keys.' )
142139 raise err
143140
144141
145142# Convenience functions which use the OTA class to perform OTA updates.
146143def from_file (
147- url : str , sha = "" , length = 0 , verify = True , verbose = True , reboot = True , username = None , password = None
144+ url : str , sha = "" , length = 0 , verify = True , verbose = True , reboot = True , ** kw
148145) -> None :
149146 with OTA (verify , verbose , reboot ) as ota_update :
150- ota_update .from_firmware_file (url , sha , length , username , password )
147+ ota_update .from_firmware_file (url , sha , length , ** kw )
151148
152149
153- def from_json (url : str , verify = True , verbose = True , reboot = True , username = None , password = None ) :
150+ def from_json (url : str , verify = True , verbose = True , reboot = True , ** kw ) -> None :
154151 with OTA (verify , verbose , reboot ) as ota_update :
155- ota_update .from_json (url , username , password )
152+ ota_update .from_json (url , ** kw )
0 commit comments