We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG-REPORT] groupby AttributeError: 'str' object has no attribute 'kind'
**Description**
Have a `numpy.ndarray` of shape `(25, 24)` that I'm adding as columns to a `df` with:
```
for i in range(len(hours_col) - 1):
    df.add_column("hour_%d" % i, hours_col.T[i], dtype='uint16')
```

It gets added fine & I see:
```
Columns:
column	type	unit	description	expression
IDN_NUM int64		
dow	virtual column	                astype(dt_dayofweek(__START_DT),` 'uint8')
hour_0	uint16		
hour_1	uint16		
hour_2	uint16		
hour_3	uint16		
hour_4	uint16		
hour_5	uint16		
hour_6	uint16		
hour_7	uint16		
hour_8	uint16		
hour_9	uint16		
hour_10	uint16		
hour_11	uint16		
hour_12	uint16		
hour_13	uint16		
hour_14	uint16		
hour_15	uint16		
hour_16	uint16		
hour_17	uint16		
hour_18	uint16		
hour_19	uint16		
hour_20	uint16		
hour_21	uint16		
hour_22	uint16		
hour_23	uint16		
```

However, when I do `type(df.hour_0.dtype)`, it says `str`.
Whereas, `type(df.dow.dtype)` says `numpy.dtype`.

`df.hour_0` says:
```
Expression = hour_0
Length: 25 dtype: uint16 (column)
---------------------------------
 0     0
 1  1777
 2     0
 3     0
 4     0
  ...   
20     0
21   122
22     0
23     0
24     0
```

So, the following `groupby` errors out:
```
df = df.groupby([df.IDN_NUM]).agg('sum')
```


Is this because `hour_%` are `expressions`, or maybe because `dow` is a `virtual column` that needs to be materialized somehow?
**EDIT:** I've just tried `df = df.materialize('dow')` and the error still remains.


**Error**
```
c:\program files\python38\lib\site-packages\vaex\agg.py in _prepare_types(self, df)
     95                 self.dtype_out = np.dtype('int64')
     96             if self.short_name in ['sum', 'summoment']:
---> 97                 self.dtype_out = vaex.utils.upcast(self.dtype_in)
     98 
     99     def add_operations(self, agg_task, **kwargs):

c:\program files\python38\lib\site-packages\vaex\utils.py in upcast(dtype)
    910 
    911 def upcast(dtype):
--> 912     if dtype.kind == "b":
    913         return np.dtype('int64')
    914     if dtype.kind == "i":

AttributeError: 'str' object has no attribute 'kind'
```


**Software information**
 - Vaex version: 
 ```{'vaex': '3.0.0',
 'vaex-core': '2.0.3',
 'vaex-viz': '0.4.0',
 'vaex-hdf5': '0.6.0',
 'vaex-server': '0.3.1',
 'vaex-astro': '0.7.0',
 'vaex-jupyter': '0.5.2',
 'vaex-ml': '0.9.0',
 'vaex-arrow': '0.5.1'
``` 
 - Vaex was installed via: pip
 - OS: WIn10


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `packages/vaex-core/vaex/convert.py`
Content:
```
1 import re
2 import logging
3 import os
4 
5 import vaex
6 import vaex.file
7 import vaex.cache
8 
9 log = logging.getLogger('vaex.cache')
10 
11 
12 def _convert_name(filenames, shuffle=False, suffix=None):
13     '''Convert a filename (or list of) to a filename with .hdf5 and optionally a -shuffle or other suffix'''
14     if not isinstance(filenames, (list, tuple)):
15         filenames = [filenames]
16     filenames = [vaex.file.stringyfy(f) for f in filenames]
17     base = filenames[0]
18     if shuffle:
19         base += '-shuffle'
20     if suffix:
21         base += suffix
22     if len(filenames) > 1:
23         return base + "_and_{}_more.hdf5".format(len(filenames)-1)
24     else:
25         return base + ".hdf5"
26 
27 
28 def convert(path_input, fs_options_input, fs_input, path_output, fs_options_output, fs_output, *args, **kwargs):
29     @vaex.cache.output_file(
30         path_input=path_input, fs_options_input=fs_options_input, fs_input=fs_input,
31         path_output=path_output, fs_options_output=fs_options_output, fs_output=fs_output)
32     def cached_output(*args, **kwargs):
33         ds = vaex.dataset.open(path_input, fs_options=fs_options_input, fs=fs_input, *args, **kwargs)
34         if ds is not None:
35             df = vaex.from_dataset(ds)
36             df.export(path_output, fs_options=fs_options_output, fs=fs_output)
37     cached_output(*args, **kwargs)
38 
39 
40 def convert_csv(path_input, fs_options_input, fs_input, path_output, fs_options_output, fs_output, *args, **kwargs):
41     @vaex.cache.output_file(
42         path_input=path_input, fs_options_input=fs_options_input, fs_input=fs_input,
43         path_output=path_output, fs_options_output=fs_options_output, fs_output=fs_output)
44     def cached_output(*args, **kwargs):
45         if fs_options_output:
46             raise ValueError(f'No fs_options support for output/convert')
47         if 'chunk_size' not in kwargs:
48             # make it memory efficient by default
49             kwargs['chunk_size'] = 5_000_000
50         _from_csv_convert_and_read(path_input, maybe_convert_path=path_output, fs_options=fs_options_input, fs=fs_input, **kwargs)
51     cached_output(*args, **kwargs)
52 
53 
54 def _from_csv_convert_and_read(filename_or_buffer, maybe_convert_path, chunk_size, fs_options, fs=None, copy_index=False, **kwargs):
55     # figure out the CSV file path
56     if isinstance(maybe_convert_path, str):
57         csv_path = re.sub(r'\.hdf5$', '', str(maybe_convert_path), flags=re.IGNORECASE)
58     elif isinstance(filename_or_buffer, str):
59         csv_path = filename_or_buffer
60     else:
61         raise ValueError('Cannot derive filename to use for converted HDF5 file, '
62                          'please specify it using convert="my.csv.hdf5"')
63 
64     combined_hdf5 = _convert_name(csv_path)
65 
66     # convert CSV chunks to separate HDF5 files
67     import pandas as pd
68     converted_paths = []
69     with vaex.file.open(filename_or_buffer, fs_options=fs_options, fs=fs, for_arrow=True) as f:
70         csv_reader = pd.read_csv(filename_or_buffer, chunksize=chunk_size, **kwargs)
71         for i, df_pandas in enumerate(csv_reader):
72             df = vaex.from_pandas(df_pandas, copy_index=copy_index)
73             filename_hdf5 = _convert_name(csv_path, suffix='_chunk%d' % i)
74             df.export_hdf5(filename_hdf5)
75             converted_paths.append(filename_hdf5)
76             log.info('saved chunk #%d to %s' % (i, filename_hdf5))
77 
78     # combine chunks into one HDF5 file
79     if len(converted_paths) == 1:
80         # no need to merge several HDF5 files
81         os.rename(converted_paths[0], combined_hdf5)
82     else:
83         log.info('converting %d chunks into single HDF5 file %s' % (len(converted_paths), combined_hdf5))
84         dfs = [vaex.open(p) for p in converted_paths]
85         df_combined = vaex.concat(dfs)
86         df_combined.export_hdf5(combined_hdf5)
87 
88         log.info('deleting %d chunk files' % len(converted_paths))
89         for df, df_path in zip(dfs, converted_paths):
90             try:
91                 df.close()
92                 os.remove(df_path)
93             except Exception as e:
94                 log.error('Could not close or delete intermediate hdf5 file %s used to convert %s to hdf5: %s' % (
95                     df_path, csv_path, e))
96 
97 
98 def main(argv):
99     import argparse
100     parser = argparse.ArgumentParser(argv[0])
101     parser.add_argument('--verbose', '-v', action='count', default=0)
102     parser.add_argument('--quiet', '-q', default=False, action='store_true', help="do not output anything")
103     parser.add_argument('--list', '-l', default=False, action='store_true', help="list columns of input")
104     parser.add_argument('--progress', help="show progress (default: %(default)s)", default=True, action='store_true')
105     parser.add_argument('--no-progress', dest="progress", action='store_false')
106     parser.add_argument('--no-delete', help="Delete file on failure (default: %(default)s)", dest='delete', default=True, action='store_false')
107     parser.add_argument('--shuffle', "-s", dest="shuffle", action='store_true', default=False)
108     parser.add_argument('--sort', dest="sort", default=None)
109     parser.add_argument('--fraction', "-f", dest="fraction", type=float, default=1.0, help="fraction of input dataset to export")
110     parser.add_argument('--filter', dest="filter", default=None, help="filter to apply before exporting")
111     parser.add_argument("input", help="input source or file, when prefixed with @ it is assumed to be a text file with a file list (one file per line)")
112     parser.add_argument("output", help="output file (ends in .hdf5)")
113     parser.add_argument("columns", help="list of columns to export (or all when empty)", nargs="*")
114 
115     args = parser.parse_args(argv[1:])
116 
117     verbosity = ["ERROR", "WARNING", "INFO", "DEBUG"]
118     logging.getLogger("vaex").setLevel(verbosity[min(3, args.verbose)])
119     if args.input[0] == "@":
120         inputs = open(args.input[1:]).readlines()
121         df = vaex.open_many(inputs)
122     else:
123         df = vaex.open(args.input)
124 
125     if df:
126         df.set_active_fraction(args.fraction)
127     if args.list:
128         print("\n".join(df.get_column_names()))
129     else:
130         if args.columns:
131             all_columns = df.get_column_names()
132             columns = args.columns
133             for column in columns:
134                 if column not in all_columns:
135                     # if not args.quiet:
136                     print("column %r does not exist, run with --list or -l to list all columns" % column)
137                     return 1
138             df = df[columns]
139         else:
140             columns = df.get_column_names()
141 
142         if not args.quiet:
143             print("exporting %d rows and %d columns" % (len(df), len(columns)))
144             print("columns: " + " ".join(columns))
145 
146         if args.filter:
147             df = df.filter(args.filter)
148         if args.sort:
149             df = df.sort(args.sort)
150         try:
151             df.export(args.output, progress=args.progress)
152             if not args.quiet:
153                 print("\noutput to %s" % os.path.abspath(args.output))
154             df.close()
155         except:
156             if not args.quiet:
157                 print("\nfailed to write to%s" % os.path.abspath(args.output))
158             if args.delete:
159                 if args.delete:
160                     os.remove(args.output)
161                     print("\ndeleted output %s (pass --no-delete to avoid that)" % os.path.abspath(args.output))
162             raise
163 
164 
165     return 0
166
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/packages/vaex-core/vaex/convert.py b/packages/vaex-core/vaex/convert.py
--- a/packages/vaex-core/vaex/convert.py
+++ b/packages/vaex-core/vaex/convert.py
@@ -47,19 +47,14 @@
         if 'chunk_size' not in kwargs:
             # make it memory efficient by default
             kwargs['chunk_size'] = 5_000_000
-        _from_csv_convert_and_read(path_input, maybe_convert_path=path_output, fs_options=fs_options_input, fs=fs_input, **kwargs)
+        _from_csv_convert_and_read(path_input, path_output=path_output, fs_options=fs_options_input, fs=fs_input, **kwargs)
     cached_output(*args, **kwargs)
 
 
-def _from_csv_convert_and_read(filename_or_buffer, maybe_convert_path, chunk_size, fs_options, fs=None, copy_index=False, **kwargs):
+def _from_csv_convert_and_read(filename_or_buffer, path_output, chunk_size, fs_options, fs=None, copy_index=False, **kwargs):
     # figure out the CSV file path
-    if isinstance(maybe_convert_path, str):
-        csv_path = re.sub(r'\.hdf5$', '', str(maybe_convert_path), flags=re.IGNORECASE)
-    elif isinstance(filename_or_buffer, str):
-        csv_path = filename_or_buffer
-    else:
-        raise ValueError('Cannot derive filename to use for converted HDF5 file, '
-                         'please specify it using convert="my.csv.hdf5"')
+    csv_path = vaex.file.stringyfy(filename_or_buffer)
+    path_output_bare, ext, _ = vaex.file.split_ext(path_output)
 
     combined_hdf5 = _convert_name(csv_path)
 
@@ -70,20 +65,20 @@
         csv_reader = pd.read_csv(filename_or_buffer, chunksize=chunk_size, **kwargs)
         for i, df_pandas in enumerate(csv_reader):
             df = vaex.from_pandas(df_pandas, copy_index=copy_index)
-            filename_hdf5 = _convert_name(csv_path, suffix='_chunk%d' % i)
-            df.export_hdf5(filename_hdf5)
-            converted_paths.append(filename_hdf5)
-            log.info('saved chunk #%d to %s' % (i, filename_hdf5))
+            chunk_name = f'{path_output_bare}_chunk_{i}.{ext}'
+            df.export(chunk_name)
+            converted_paths.append(chunk_name)
+            log.info('saved chunk #%d to %s' % (i, chunk_name))
 
     # combine chunks into one HDF5 file
     if len(converted_paths) == 1:
         # no need to merge several HDF5 files
-        os.rename(converted_paths[0], combined_hdf5)
+        os.rename(converted_paths[0], path_output)
     else:
-        log.info('converting %d chunks into single HDF5 file %s' % (len(converted_paths), combined_hdf5))
+        log.info('converting %d chunks into single file %s' % (len(converted_paths), path_output))
         dfs = [vaex.open(p) for p in converted_paths]
         df_combined = vaex.concat(dfs)
-        df_combined.export_hdf5(combined_hdf5)
+        df_combined.export(path_output)
 
         log.info('deleting %d chunk files' % len(converted_paths))
         for df, df_path in zip(dfs, converted_paths):
@@ -91,8 +86,7 @@
                 df.close()
                 os.remove(df_path)
             except Exception as e:
-                log.error('Could not close or delete intermediate hdf5 file %s used to convert %s to hdf5: %s' % (
-                    df_path, csv_path, e))
+                log.error('Could not close or delete intermediate file %s used to convert %s to single file: %s', (df_path, csv_path, path_output))
 
 
 def main(argv):

{"golden_diff": "diff --git a/packages/vaex-core/vaex/convert.py b/packages/vaex-core/vaex/convert.py\n--- a/packages/vaex-core/vaex/convert.py\n+++ b/packages/vaex-core/vaex/convert.py\n@@ -47,19 +47,14 @@\n         if 'chunk_size' not in kwargs:\n             # make it memory efficient by default\n             kwargs['chunk_size'] = 5_000_000\n-        _from_csv_convert_and_read(path_input, maybe_convert_path=path_output, fs_options=fs_options_input, fs=fs_input, **kwargs)\n+        _from_csv_convert_and_read(path_input, path_output=path_output, fs_options=fs_options_input, fs=fs_input, **kwargs)\n     cached_output(*args, **kwargs)\n \n \n-def _from_csv_convert_and_read(filename_or_buffer, maybe_convert_path, chunk_size, fs_options, fs=None, copy_index=False, **kwargs):\n+def _from_csv_convert_and_read(filename_or_buffer, path_output, chunk_size, fs_options, fs=None, copy_index=False, **kwargs):\n     # figure out the CSV file path\n-    if isinstance(maybe_convert_path, str):\n-        csv_path = re.sub(r'\\.hdf5$', '', str(maybe_convert_path), flags=re.IGNORECASE)\n-    elif isinstance(filename_or_buffer, str):\n-        csv_path = filename_or_buffer\n-    else:\n-        raise ValueError('Cannot derive filename to use for converted HDF5 file, '\n-                         'please specify it using convert=\"my.csv.hdf5\"')\n+    csv_path = vaex.file.stringyfy(filename_or_buffer)\n+    path_output_bare, ext, _ = vaex.file.split_ext(path_output)\n \n     combined_hdf5 = _convert_name(csv_path)\n \n@@ -70,20 +65,20 @@\n         csv_reader = pd.read_csv(filename_or_buffer, chunksize=chunk_size, **kwargs)\n         for i, df_pandas in enumerate(csv_reader):\n             df = vaex.from_pandas(df_pandas, copy_index=copy_index)\n-            filename_hdf5 = _convert_name(csv_path, suffix='_chunk%d' % i)\n-            df.export_hdf5(filename_hdf5)\n-            converted_paths.append(filename_hdf5)\n-            log.info('saved chunk #%d to %s' % (i, filename_hdf5))\n+            chunk_name = f'{path_output_bare}_chunk_{i}.{ext}'\n+            df.export(chunk_name)\n+            converted_paths.append(chunk_name)\n+            log.info('saved chunk #%d to %s' % (i, chunk_name))\n \n     # combine chunks into one HDF5 file\n     if len(converted_paths) == 1:\n         # no need to merge several HDF5 files\n-        os.rename(converted_paths[0], combined_hdf5)\n+        os.rename(converted_paths[0], path_output)\n     else:\n-        log.info('converting %d chunks into single HDF5 file %s' % (len(converted_paths), combined_hdf5))\n+        log.info('converting %d chunks into single file %s' % (len(converted_paths), path_output))\n         dfs = [vaex.open(p) for p in converted_paths]\n         df_combined = vaex.concat(dfs)\n-        df_combined.export_hdf5(combined_hdf5)\n+        df_combined.export(path_output)\n \n         log.info('deleting %d chunk files' % len(converted_paths))\n         for df, df_path in zip(dfs, converted_paths):\n@@ -91,8 +86,7 @@\n                 df.close()\n                 os.remove(df_path)\n             except Exception as e:\n-                log.error('Could not close or delete intermediate hdf5 file %s used to convert %s to hdf5: %s' % (\n-                    df_path, csv_path, e))\n+                log.error('Could not close or delete intermediate file %s used to convert %s to single file: %s', (df_path, csv_path, path_output))\n \n \n def main(argv):\n", "issue": "[BUG-REPORT] groupby AttributeError: 'str' object has no attribute 'kind'\n**Description**\r\nHave a `numpy.ndarray` of shape `(25, 24)` that I'm adding as columns to a `df` with:\r\n```\r\nfor i in range(len(hours_col) - 1):\r\n    df.add_column(\"hour_%d\" % i, hours_col.T[i], dtype='uint16')\r\n```\r\n\r\nIt gets added fine & I see:\r\n```\r\nColumns:\r\ncolumn\ttype\tunit\tdescription\texpression\r\nIDN_NUM int64\t\t\r\ndow\tvirtual column\t                astype(dt_dayofweek(__START_DT),` 'uint8')\r\nhour_0\tuint16\t\t\r\nhour_1\tuint16\t\t\r\nhour_2\tuint16\t\t\r\nhour_3\tuint16\t\t\r\nhour_4\tuint16\t\t\r\nhour_5\tuint16\t\t\r\nhour_6\tuint16\t\t\r\nhour_7\tuint16\t\t\r\nhour_8\tuint16\t\t\r\nhour_9\tuint16\t\t\r\nhour_10\tuint16\t\t\r\nhour_11\tuint16\t\t\r\nhour_12\tuint16\t\t\r\nhour_13\tuint16\t\t\r\nhour_14\tuint16\t\t\r\nhour_15\tuint16\t\t\r\nhour_16\tuint16\t\t\r\nhour_17\tuint16\t\t\r\nhour_18\tuint16\t\t\r\nhour_19\tuint16\t\t\r\nhour_20\tuint16\t\t\r\nhour_21\tuint16\t\t\r\nhour_22\tuint16\t\t\r\nhour_23\tuint16\t\t\r\n```\r\n\r\nHowever, when I do `type(df.hour_0.dtype)`, it says `str`.\r\nWhereas, `type(df.dow.dtype)` says `numpy.dtype`.\r\n\r\n`df.hour_0` says:\r\n```\r\nExpression = hour_0\r\nLength: 25 dtype: uint16 (column)\r\n---------------------------------\r\n 0     0\r\n 1  1777\r\n 2     0\r\n 3     0\r\n 4     0\r\n  ...   \r\n20     0\r\n21   122\r\n22     0\r\n23     0\r\n24     0\r\n```\r\n\r\nSo, the following `groupby` errors out:\r\n```\r\ndf = df.groupby([df.IDN_NUM]).agg('sum')\r\n```\r\n\r\n\r\nIs this because `hour_%` are `expressions`, or maybe because `dow` is a `virtual column` that needs to be materialized somehow?\r\n**EDIT:** I've just tried `df = df.materialize('dow')` and the error still remains.\r\n\r\n\r\n**Error**\r\n```\r\nc:\\program files\\python38\\lib\\site-packages\\vaex\\agg.py in _prepare_types(self, df)\r\n     95                 self.dtype_out = np.dtype('int64')\r\n     96             if self.short_name in ['sum', 'summoment']:\r\n---> 97                 self.dtype_out = vaex.utils.upcast(self.dtype_in)\r\n     98 \r\n     99     def add_operations(self, agg_task, **kwargs):\r\n\r\nc:\\program files\\python38\\lib\\site-packages\\vaex\\utils.py in upcast(dtype)\r\n    910 \r\n    911 def upcast(dtype):\r\n--> 912     if dtype.kind == \"b\":\r\n    913         return np.dtype('int64')\r\n    914     if dtype.kind == \"i\":\r\n\r\nAttributeError: 'str' object has no attribute 'kind'\r\n```\r\n\r\n\r\n**Software information**\r\n - Vaex version: \r\n ```{'vaex': '3.0.0',\r\n 'vaex-core': '2.0.3',\r\n 'vaex-viz': '0.4.0',\r\n 'vaex-hdf5': '0.6.0',\r\n 'vaex-server': '0.3.1',\r\n 'vaex-astro': '0.7.0',\r\n 'vaex-jupyter': '0.5.2',\r\n 'vaex-ml': '0.9.0',\r\n 'vaex-arrow': '0.5.1'\r\n``` \r\n - Vaex was installed via: pip\r\n - OS: WIn10\r\n\n", "before_files": [{"content": "import re\nimport logging\nimport os\n\nimport vaex\nimport vaex.file\nimport vaex.cache\n\nlog = logging.getLogger('vaex.cache')\n\n\ndef _convert_name(filenames, shuffle=False, suffix=None):\n    '''Convert a filename (or list of) to a filename with .hdf5 and optionally a -shuffle or other suffix'''\n    if not isinstance(filenames, (list, tuple)):\n        filenames = [filenames]\n    filenames = [vaex.file.stringyfy(f) for f in filenames]\n    base = filenames[0]\n    if shuffle:\n        base += '-shuffle'\n    if suffix:\n        base += suffix\n    if len(filenames) > 1:\n        return base + \"_and_{}_more.hdf5\".format(len(filenames)-1)\n    else:\n        return base + \".hdf5\"\n\n\ndef convert(path_input, fs_options_input, fs_input, path_output, fs_options_output, fs_output, *args, **kwargs):\n    @vaex.cache.output_file(\n        path_input=path_input, fs_options_input=fs_options_input, fs_input=fs_input,\n        path_output=path_output, fs_options_output=fs_options_output, fs_output=fs_output)\n    def cached_output(*args, **kwargs):\n        ds = vaex.dataset.open(path_input, fs_options=fs_options_input, fs=fs_input, *args, **kwargs)\n        if ds is not None:\n            df = vaex.from_dataset(ds)\n            df.export(path_output, fs_options=fs_options_output, fs=fs_output)\n    cached_output(*args, **kwargs)\n\n\ndef convert_csv(path_input, fs_options_input, fs_input, path_output, fs_options_output, fs_output, *args, **kwargs):\n    @vaex.cache.output_file(\n        path_input=path_input, fs_options_input=fs_options_input, fs_input=fs_input,\n        path_output=path_output, fs_options_output=fs_options_output, fs_output=fs_output)\n    def cached_output(*args, **kwargs):\n        if fs_options_output:\n            raise ValueError(f'No fs_options support for output/convert')\n        if 'chunk_size' not in kwargs:\n            # make it memory efficient by default\n            kwargs['chunk_size'] = 5_000_000\n        _from_csv_convert_and_read(path_input, maybe_convert_path=path_output, fs_options=fs_options_input, fs=fs_input, **kwargs)\n    cached_output(*args, **kwargs)\n\n\ndef _from_csv_convert_and_read(filename_or_buffer, maybe_convert_path, chunk_size, fs_options, fs=None, copy_index=False, **kwargs):\n    # figure out the CSV file path\n    if isinstance(maybe_convert_path, str):\n        csv_path = re.sub(r'\\.hdf5$', '', str(maybe_convert_path), flags=re.IGNORECASE)\n    elif isinstance(filename_or_buffer, str):\n        csv_path = filename_or_buffer\n    else:\n        raise ValueError('Cannot derive filename to use for converted HDF5 file, '\n                         'please specify it using convert=\"my.csv.hdf5\"')\n\n    combined_hdf5 = _convert_name(csv_path)\n\n    # convert CSV chunks to separate HDF5 files\n    import pandas as pd\n    converted_paths = []\n    with vaex.file.open(filename_or_buffer, fs_options=fs_options, fs=fs, for_arrow=True) as f:\n        csv_reader = pd.read_csv(filename_or_buffer, chunksize=chunk_size, **kwargs)\n        for i, df_pandas in enumerate(csv_reader):\n            df = vaex.from_pandas(df_pandas, copy_index=copy_index)\n            filename_hdf5 = _convert_name(csv_path, suffix='_chunk%d' % i)\n            df.export_hdf5(filename_hdf5)\n            converted_paths.append(filename_hdf5)\n            log.info('saved chunk #%d to %s' % (i, filename_hdf5))\n\n    # combine chunks into one HDF5 file\n    if len(converted_paths) == 1:\n        # no need to merge several HDF5 files\n        os.rename(converted_paths[0], combined_hdf5)\n    else:\n        log.info('converting %d chunks into single HDF5 file %s' % (len(converted_paths), combined_hdf5))\n        dfs = [vaex.open(p) for p in converted_paths]\n        df_combined = vaex.concat(dfs)\n        df_combined.export_hdf5(combined_hdf5)\n\n        log.info('deleting %d chunk files' % len(converted_paths))\n        for df, df_path in zip(dfs, converted_paths):\n            try:\n                df.close()\n                os.remove(df_path)\n            except Exception as e:\n                log.error('Could not close or delete intermediate hdf5 file %s used to convert %s to hdf5: %s' % (\n                    df_path, csv_path, e))\n\n\ndef main(argv):\n    import argparse\n    parser = argparse.ArgumentParser(argv[0])\n    parser.add_argument('--verbose', '-v', action='count', default=0)\n    parser.add_argument('--quiet', '-q', default=False, action='store_true', help=\"do not output anything\")\n    parser.add_argument('--list', '-l', default=False, action='store_true', help=\"list columns of input\")\n    parser.add_argument('--progress', help=\"show progress (default: %(default)s)\", default=True, action='store_true')\n    parser.add_argument('--no-progress', dest=\"progress\", action='store_false')\n    parser.add_argument('--no-delete', help=\"Delete file on failure (default: %(default)s)\", dest='delete', default=True, action='store_false')\n    parser.add_argument('--shuffle', \"-s\", dest=\"shuffle\", action='store_true', default=False)\n    parser.add_argument('--sort', dest=\"sort\", default=None)\n    parser.add_argument('--fraction', \"-f\", dest=\"fraction\", type=float, default=1.0, help=\"fraction of input dataset to export\")\n    parser.add_argument('--filter', dest=\"filter\", default=None, help=\"filter to apply before exporting\")\n    parser.add_argument(\"input\", help=\"input source or file, when prefixed with @ it is assumed to be a text file with a file list (one file per line)\")\n    parser.add_argument(\"output\", help=\"output file (ends in .hdf5)\")\n    parser.add_argument(\"columns\", help=\"list of columns to export (or all when empty)\", nargs=\"*\")\n\n    args = parser.parse_args(argv[1:])\n\n    verbosity = [\"ERROR\", \"WARNING\", \"INFO\", \"DEBUG\"]\n    logging.getLogger(\"vaex\").setLevel(verbosity[min(3, args.verbose)])\n    if args.input[0] == \"@\":\n        inputs = open(args.input[1:]).readlines()\n        df = vaex.open_many(inputs)\n    else:\n        df = vaex.open(args.input)\n\n    if df:\n        df.set_active_fraction(args.fraction)\n    if args.list:\n        print(\"\\n\".join(df.get_column_names()))\n    else:\n        if args.columns:\n            all_columns = df.get_column_names()\n            columns = args.columns\n            for column in columns:\n                if column not in all_columns:\n                    # if not args.quiet:\n                    print(\"column %r does not exist, run with --list or -l to list all columns\" % column)\n                    return 1\n            df = df[columns]\n        else:\n            columns = df.get_column_names()\n\n        if not args.quiet:\n            print(\"exporting %d rows and %d columns\" % (len(df), len(columns)))\n            print(\"columns: \" + \" \".join(columns))\n\n        if args.filter:\n            df = df.filter(args.filter)\n        if args.sort:\n            df = df.sort(args.sort)\n        try:\n            df.export(args.output, progress=args.progress)\n            if not args.quiet:\n                print(\"\\noutput to %s\" % os.path.abspath(args.output))\n            df.close()\n        except:\n            if not args.quiet:\n                print(\"\\nfailed to write to%s\" % os.path.abspath(args.output))\n            if args.delete:\n                if args.delete:\n                    os.remove(args.output)\n                    print(\"\\ndeleted output %s (pass --no-delete to avoid that)\" % os.path.abspath(args.output))\n            raise\n\n\n    return 0\n", "path": "packages/vaex-core/vaex/convert.py"}], "after_files": [{"content": "import re\nimport logging\nimport os\n\nimport vaex\nimport vaex.file\nimport vaex.cache\n\nlog = logging.getLogger('vaex.cache')\n\n\ndef _convert_name(filenames, shuffle=False, suffix=None):\n    '''Convert a filename (or list of) to a filename with .hdf5 and optionally a -shuffle or other suffix'''\n    if not isinstance(filenames, (list, tuple)):\n        filenames = [filenames]\n    filenames = [vaex.file.stringyfy(f) for f in filenames]\n    base = filenames[0]\n    if shuffle:\n        base += '-shuffle'\n    if suffix:\n        base += suffix\n    if len(filenames) > 1:\n        return base + \"_and_{}_more.hdf5\".format(len(filenames)-1)\n    else:\n        return base + \".hdf5\"\n\n\ndef convert(path_input, fs_options_input, fs_input, path_output, fs_options_output, fs_output, *args, **kwargs):\n    @vaex.cache.output_file(\n        path_input=path_input, fs_options_input=fs_options_input, fs_input=fs_input,\n        path_output=path_output, fs_options_output=fs_options_output, fs_output=fs_output)\n    def cached_output(*args, **kwargs):\n        ds = vaex.dataset.open(path_input, fs_options=fs_options_input, fs=fs_input, *args, **kwargs)\n        if ds is not None:\n            df = vaex.from_dataset(ds)\n            df.export(path_output, fs_options=fs_options_output, fs=fs_output)\n    cached_output(*args, **kwargs)\n\n\ndef convert_csv(path_input, fs_options_input, fs_input, path_output, fs_options_output, fs_output, *args, **kwargs):\n    @vaex.cache.output_file(\n        path_input=path_input, fs_options_input=fs_options_input, fs_input=fs_input,\n        path_output=path_output, fs_options_output=fs_options_output, fs_output=fs_output)\n    def cached_output(*args, **kwargs):\n        if fs_options_output:\n            raise ValueError(f'No fs_options support for output/convert')\n        if 'chunk_size' not in kwargs:\n            # make it memory efficient by default\n            kwargs['chunk_size'] = 5_000_000\n        _from_csv_convert_and_read(path_input, path_output=path_output, fs_options=fs_options_input, fs=fs_input, **kwargs)\n    cached_output(*args, **kwargs)\n\n\ndef _from_csv_convert_and_read(filename_or_buffer, path_output, chunk_size, fs_options, fs=None, copy_index=False, **kwargs):\n    # figure out the CSV file path\n    csv_path = vaex.file.stringyfy(filename_or_buffer)\n    path_output_bare, ext, _ = vaex.file.split_ext(path_output)\n\n    combined_hdf5 = _convert_name(csv_path)\n\n    # convert CSV chunks to separate HDF5 files\n    import pandas as pd\n    converted_paths = []\n    with vaex.file.open(filename_or_buffer, fs_options=fs_options, fs=fs, for_arrow=True) as f:\n        csv_reader = pd.read_csv(filename_or_buffer, chunksize=chunk_size, **kwargs)\n        for i, df_pandas in enumerate(csv_reader):\n            df = vaex.from_pandas(df_pandas, copy_index=copy_index)\n            chunk_name = f'{path_output_bare}_chunk_{i}.{ext}'\n            df.export(chunk_name)\n            converted_paths.append(chunk_name)\n            log.info('saved chunk #%d to %s' % (i, chunk_name))\n\n    # combine chunks into one HDF5 file\n    if len(converted_paths) == 1:\n        # no need to merge several HDF5 files\n        os.rename(converted_paths[0], path_output)\n    else:\n        log.info('converting %d chunks into single file %s' % (len(converted_paths), path_output))\n        dfs = [vaex.open(p) for p in converted_paths]\n        df_combined = vaex.concat(dfs)\n        df_combined.export(path_output)\n\n        log.info('deleting %d chunk files' % len(converted_paths))\n        for df, df_path in zip(dfs, converted_paths):\n            try:\n                df.close()\n                os.remove(df_path)\n            except Exception as e:\n                log.error('Could not close or delete intermediate file %s used to convert %s to single file: %s', (df_path, csv_path, path_output))\n\n\ndef main(argv):\n    import argparse\n    parser = argparse.ArgumentParser(argv[0])\n    parser.add_argument('--verbose', '-v', action='count', default=0)\n    parser.add_argument('--quiet', '-q', default=False, action='store_true', help=\"do not output anything\")\n    parser.add_argument('--list', '-l', default=False, action='store_true', help=\"list columns of input\")\n    parser.add_argument('--progress', help=\"show progress (default: %(default)s)\", default=True, action='store_true')\n    parser.add_argument('--no-progress', dest=\"progress\", action='store_false')\n    parser.add_argument('--no-delete', help=\"Delete file on failure (default: %(default)s)\", dest='delete', default=True, action='store_false')\n    parser.add_argument('--shuffle', \"-s\", dest=\"shuffle\", action='store_true', default=False)\n    parser.add_argument('--sort', dest=\"sort\", default=None)\n    parser.add_argument('--fraction', \"-f\", dest=\"fraction\", type=float, default=1.0, help=\"fraction of input dataset to export\")\n    parser.add_argument('--filter', dest=\"filter\", default=None, help=\"filter to apply before exporting\")\n    parser.add_argument(\"input\", help=\"input source or file, when prefixed with @ it is assumed to be a text file with a file list (one file per line)\")\n    parser.add_argument(\"output\", help=\"output file (ends in .hdf5)\")\n    parser.add_argument(\"columns\", help=\"list of columns to export (or all when empty)\", nargs=\"*\")\n\n    args = parser.parse_args(argv[1:])\n\n    verbosity = [\"ERROR\", \"WARNING\", \"INFO\", \"DEBUG\"]\n    logging.getLogger(\"vaex\").setLevel(verbosity[min(3, args.verbose)])\n    if args.input[0] == \"@\":\n        inputs = open(args.input[1:]).readlines()\n        df = vaex.open_many(inputs)\n    else:\n        df = vaex.open(args.input)\n\n    if df:\n        df.set_active_fraction(args.fraction)\n    if args.list:\n        print(\"\\n\".join(df.get_column_names()))\n    else:\n        if args.columns:\n            all_columns = df.get_column_names()\n            columns = args.columns\n            for column in columns:\n                if column not in all_columns:\n                    # if not args.quiet:\n                    print(\"column %r does not exist, run with --list or -l to list all columns\" % column)\n                    return 1\n            df = df[columns]\n        else:\n            columns = df.get_column_names()\n\n        if not args.quiet:\n            print(\"exporting %d rows and %d columns\" % (len(df), len(columns)))\n            print(\"columns: \" + \" \".join(columns))\n\n        if args.filter:\n            df = df.filter(args.filter)\n        if args.sort:\n            df = df.sort(args.sort)\n        try:\n            df.export(args.output, progress=args.progress)\n            if not args.quiet:\n                print(\"\\noutput to %s\" % os.path.abspath(args.output))\n            df.close()\n        except:\n            if not args.quiet:\n                print(\"\\nfailed to write to%s\" % os.path.abspath(args.output))\n            if args.delete:\n                if args.delete:\n                    os.remove(args.output)\n                    print(\"\\ndeleted output %s (pass --no-delete to avoid that)\" % os.path.abspath(args.output))\n            raise\n\n\n    return 0\n", "path": "packages/vaex-core/vaex/convert.py"}]}