Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if isinstance(fn, io.StringIO):
fn.seek(0)
yield fn
elif isinstance(fn, Path):
finf = fn.stat()
if finf.st_size > 100e6:
logging.info(f'opening {finf.st_size/1e6} MByte {fn.name}')
if fn.suffix == '.gz':
with gzip.open(fn, 'rt') as f:
version, is_crinex = rinex_version(first_nonblank_line(f))
f.seek(0)
if is_crinex and not header:
f = io.StringIO(opencrx(f))
yield f
elif fn.suffix == '.zip':
with zipfile.ZipFile(fn, 'r') as z:
flist = z.namelist()
for rinexfn in flist:
with z.open(rinexfn, 'r') as bf:
f = io.StringIO(io.TextIOWrapper(bf, encoding='ascii', errors='ignore').read())
yield f
elif fn.suffix == '.Z':
if unlzw is None:
raise ImportError('pip install unlzw3')
with fn.open('rb') as zu:
with io.StringIO(unlzw(zu.read()).decode('ascii')) as f:
yield f
else: # assume not compressed (or Hatanaka)
with fn.open('r', encoding='ascii', errors='ignore') as f:
with z.open(rinexfn, 'r') as bf:
f = io.StringIO(io.TextIOWrapper(bf, encoding='ascii', errors='ignore').read())
yield f
elif fn.suffix == '.Z':
if unlzw is None:
raise ImportError('pip install unlzw3')
with fn.open('rb') as zu:
with io.StringIO(unlzw(zu.read()).decode('ascii')) as f:
yield f
else: # assume not compressed (or Hatanaka)
with fn.open('r', encoding='ascii', errors='ignore') as f:
version, is_crinex = rinex_version(first_nonblank_line(f))
f.seek(0)
if is_crinex and not header:
f = io.StringIO(opencrx(f))
yield f
else:
raise OSError(f'Unsure what to do with input of type: {type(fn)}')