getloc stringio

testindex

test if read

georinex/utils.py

@@ -64,20 +64,24 @@ def gettime(fn: Union[TextIO, str, Path]) -> xarray.DataArray:
     return times
 
 
-def getlocations(flist: Sequence[Path]) -> pandas.DataFrame:
+def getlocations(flist: Union[TextIO, Sequence[Path]]) -> pandas.DataFrame:
     """
     retrieve locations of GNSS receivers
 
     Requires pymap3d.ecef2geodetic
     """
-    if isinstance(flist, Path):
+    if isinstance(flist, (Path, io.StringIO)):
         flist = [flist]
 
-    locs = pandas.DataFrame(index=[f.name for f in flist],
-                            columns=['lat', 'lon', 'interval'])
+    if isinstance(flist[0], io.StringIO):
+        locs = pandas.DataFrame(index=['0'],
+                                columns=['lat', 'lon', 'interval'])
+    else:
+        locs = pandas.DataFrame(index=[f.name for f in flist],
+                                columns=['lat', 'lon', 'interval'])
 
     for f in flist:
-        if f.suffix == '.nc':
+        if isinstance(f, Path) and f.suffix == '.nc':
             dat = xarray.open_dataset(f, group='OBS')
             hdr = dat.attrs
         else:
@@ -86,13 +90,18 @@ def getlocations(flist: Sequence[Path]) -> pandas.DataFrame:
             except ValueError:
                 continue
 
+        if isinstance(f, Path):
+            key = f.name
+        else:
+            key = '0'
+
         if 'position_geodetic' not in hdr:
             continue
 
-        locs.loc[f.name, 'lat'] = hdr['position_geodetic'][0]
-        locs.loc[f.name, 'lon'] = hdr['position_geodetic'][1]
+        locs.loc[key, 'lat'] = hdr['position_geodetic'][0]
+        locs.loc[key, 'lon'] = hdr['position_geodetic'][1]
         if 'interval' in hdr and hdr['interval'] is not None:
-            locs.loc[f.name, 'interval'] = hdr['interval']
+            locs.loc[key, 'interval'] = hdr['interval']
 
     locs = locs.loc[locs.loc[:, ['lat', 'lon']].notna().all(axis=1), :]
 
@@ -114,11 +123,12 @@ def rinextype(fn: Union[TextIO, Path]) -> str:
         return info
 
 
-def rinexheader(fn: Path) -> Dict[str, Any]:
+def rinexheader(fn: Union[TextIO, str, Path]) -> Dict[str, Any]:
     """
     retrieve RINEX 2/3 header as unparsed dict()
     """
-    fn = Path(fn).expanduser()
+    if isinstance(fn, (str, Path)):
+        fn = Path(fn).expanduser()
 
     info = rinexinfo(fn)
     rtype = rinextype(fn)

tests/test_stringio.py

@@ -86,9 +86,16 @@ def test_obs3():
 
 
 def test_locs():
-    locs = gr.getlocations(R / 'demo.10o')
+    fn = R / 'demo.10o'
 
-    assert locs.loc['demo.10o'].values == approx([41.3887, 2.112, 30])
+    with fn.open('r') as f:
+        txt = f.read()
+
+    with io.StringIO(txt) as f:
+        locs = gr.getlocations(f)
+
+    if locs.size > 0:
+        assert locs.iloc[0].values == approx([41.3887, 2.112, 30])
 
 
 if __name__ == '__main__':