Parquet_integration

.gitignore

@@ -21,3 +21,5 @@ tests/__pycache__/
 .pytest_cache/
 *.ipynb
 .DS_Store
+.vscode/settings.json
+.vscode/launch.json

.vscode/launch.json

@@ -0,0 +1,15 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python: Current File",
+            "type": "python",
+            "request": "launch",
+            "program": "${file}",
+            "console": "integratedTerminal"
+        }
+    ]
+}

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "python.pythonPath": "/opt/anaconda3/envs/cytominer1/bin/python"
+}

README.rst

@@ -1,3 +1,4 @@
+==================
 cytominer-database
 ==================
 
@@ -17,16 +18,93 @@ high-throughput imaging experiment. The measurements are stored across thousands
 cytominer-database helps you organize these data into a single database backend, such as SQLite.
 
 Why cytominer-database?
------------------------
+=======================
 While tools like CellProfiler can store measurements directly in databases, it is usually infeasible to create a
 centralized database in which to store these measurements. A more scalable approach is to create a set of CSVs per
 “batch” of images, and then later merge these CSVs.
 
-cytominer-database ingest reads these CSVs, checks for errors, then ingests them into a database backend, including
+cytominer-database ingest reads these CSVs, checks for errors, then ingests
+them into a database backend, including
 SQLite, MySQL, PostgresSQL, and several other backends supported by odo.
 
 .. code-block:: sh
 
 	cytominer-database ingest source_directory sqlite:///backend.sqlite -c ingest_config.ini
 
 will ingest the CSV files nested under source_directory into a SQLite backend
+
+How to use the configuration file
+=================================
+The configuration file ingest_config.ini must be located in the source_directory
+and can be modified to specify the ingestion.
+There are three different sections.
+
+The [filenames] section
+-----------------------
+
+.. code-block::
+
+  [filenames]
+  image   = image.csv      #or: Image.csv
+  object  = object.csv     #or: Object.csv
+
+Cytominer-Database is currently limited to the following measurement file kinds:
+Cells.csv, Cytoplasm.csv, Nuclei.csv, Image.csv, Object.csv.
+The [filenames] section in the configuration file saves the correct basename of existing measurement files
+(this may be important in the case of inconsistent capitalization).
+
+The [database_engine] section
+-----------------------------
+
+.. code-block::
+
+  [ingestion_engine]
+  engine = Parquet      #or: SQLite
+
+The [database_engine] section specifies the backend. Possible key-value pairs are:
+**engine** = *SQLite* or **engine** = *Parquet*.
+
+The [schema] section
+--------------------
+
+.. code-block::
+
+ [schema]
+ reference_option = sample         #or: path/to/reference/folder relative to source_directory
+ ref_fraction     = 1              #or: any decimal value in [0, 1]
+ type_conversion  = int2float      #or: all2string
+
+The [schema] section specifies how to manage incompatibilities in the table schema
+of the files.
+In that case, a Parquet file is fixed to a schema with which it was first opened,
+i.e. by the first file which is written (the reference file). To append the data
+of all .csv files of that file-kind, it is important to assure the reference file
+satisfies certain incompatibility requirements, e.g. does not miss any columns
+and all existing files can be automatically converted to the reference schema.
+This section is used only if the files are ingested to Parquet format and was
+developed to handle the special cases in which tables that cannot be concatenated automatically.
+
+There are two options for the key **reference_option**:
+The first option is to create a designated folder containing one .csv reference file for every kind of file ("Cytoplasm.csv", "Nuclei.csv", ...)
+and save the folder path in the config file as **reference_option** = *path/to/reference/folder*,
+where the path is relative to the source_directory from the ingest command.
+These reference files' schema will determine the schema of the Parquet file into which all .csv files of its kind will be ingested.
+
+
+**This option relies on manual selection, hence the chosen reference files must be checked explicitly: Make sure the .csv files are complete in the number of columns and contain no NaN values.**
+
+Alternatively, the reference files can be found automatically from a sampled subset of all existing files.
+This is the case if **reference_option** = *sample* is set.
+A subset of all files is sampled uniformly at random and the first table with
+the maximum number of columns among all sampled .csv files is chosen as the reference table.
+If this case, an additional key **ref_fraction** can be set, which specifies the fraction of files
+sampled among all files. The default value is **ref_fraction** = *1* , for which
+all tables are compared in width. This key is only used if "reference_option=sample".
+
+Lastly, the key **type_conversion** determines how the schema types are handled in the case of disagreement.
+The default value is *int2float*, for which all integer columns are converted to floats.
+This has been proven helpful for trivial columns (0-valued column), which may be of "int" type
+and cannot be written into the same table as non-trivial files with non-zero float values.
+Automatic type conversion can be avoided by converting all values to string-type.
+This can be done by setting **type_conversion** = *all2string*.
+However, the loss of type information might be a disadvantage in downstream tasks.

cytominer_database/commands/command_ingest.py

@@ -57,6 +57,6 @@
 )
 def command(source, target, config_file, munge, skip_image_prefix):
     if munge:
-        cytominer_database.munge.munge(config_file=config_file, source=source)
+        cytominer_database.munge.munge(config_path=config_file, source=source)
 
     cytominer_database.ingest.seed(source, target, config_file, skip_image_prefix)

cytominer_database/config/config_cellpainting.ini

@@ -3,3 +3,10 @@ image = Image.csv
 object = object.csv
 experiment = Experiment.csv
 
+[database_engine]
+database = SQLite   
+
+[schema]
+reference_option = sample
+ref_fraction = 1
+type_conversion = int2float 

cytominer_database/config/config_default.ini

@@ -1,3 +1,12 @@
 [filenames]
 image = Image.csv
 experiment = Experiment.csv
+
+[database_engine]
+database = SQLite   
+
+[schema]
+reference_option = sample
+ref_fraction = 1
+type_conversion = int2float 
+

cytominer_database/config/config_htqc.ini

@@ -2,3 +2,11 @@
 image = image.csv
 object = object.csv
 experiment = Experiment.csv
+
+[database_engine]
+database = SQLite   
+
+[schema]
+reference_option = sample
+ref_fraction = 1
+type_conversion = int2float 

cytominer_database/ingest.py

@@ -69,48 +69,36 @@ def into(input, output, name, identifier, skip_table_prefix=False):
      from the names of columns.
     """
 
-    with backports.tempfile.TemporaryDirectory() as directory:
-        source = os.path.join(directory, os.path.basename(input))
-
-        # create a temporary CSV file which is identical to the input CSV file
-        # but with the column names prefixed with the name of the compartment
-        # (or `Image`, if this is an image CSV file, and `skip_table_prefix` is False)
-        with open(input, "r") as fin, open(source, "w") as fout:
-            reader = csv.reader(fin)
-            writer = csv.writer(fout)
-
-            headers = next(reader)
-            if not skip_table_prefix:
-                headers = [__format__(name, header) for header in headers]
-
-            # The first column is `TableNumber`, which is the unique identifier for the image CSV
-            headers = ["TableNumber"] + headers
-
-            writer.writerow(headers)
-
-            [writer.writerow([identifier] + row) for row in reader]
-
-        # Now ingest the temp CSV file (with the modified column names) into the database backend
-        # the rows of the CSV file are inserted into a table with name `name`.
-        with warnings.catch_warnings():
-            # Suppress the following warning on Python 3:
-            #
-            #   /usr/local/lib/python3.6/site-packages/odo/utils.py:128: DeprecationWarning: inspect.getargspec() is
-            #     deprecated, use inspect.signature() or inspect.getfullargspec()
-            warnings.simplefilter("ignore", category=DeprecationWarning)
-
-            engine = create_engine(output)
-            con = engine.connect()
-
-            df = pd.read_csv(source, index_col=0)
-            df.to_sql(name=name, con=con, if_exists="append")
+    with warnings.catch_warnings():
+        # Suppress the following warning on Python 3:
+        #
+        #   /usr/local/lib/python3.6/site-packages/odo/utils.py:128: DeprecationWarning: inspect.getargspec() is
+        #     deprecated, use inspect.signature() or inspect.getfullargspec()
+        warnings.simplefilter("ignore", category=DeprecationWarning)
+        engine = create_engine(output)
+        con = engine.connect()
+
+        df = pd.read_csv(input)
+        #print("In into(): df['Index']") # no index prepended yet
+        #print(df['Index'])
+        # add "name"prefix to column headers
+        if not skip_table_prefix:
+            no_prefix = ["ImageNumber", "ObjectNumber"]  # exception columns
+            df.columns = [
+                i if i in no_prefix else name + "_" + i for i in df.columns
+            ]
+        # add TableNumber
+        number_of_rows, _ = df.shape
+        table_number_column = [identifier] * number_of_rows  # create additional column
+        df.insert(0, "TableNumber", table_number_column, allow_duplicates=False)
+        print("In into(): df.shape is ", df.shape)
+        df.to_sql(name=name, con=con, if_exists="append", index = False)
 
 def checksum(pathname, buffer_size=65536):
     """
     Generate a 32-bit unique identifier for a file.
-
-    :param pathname: input file
-    :param buffer_size: buffer size   
+
+    :param buffer_size: buffer size
     """
     with open(pathname, "rb") as stream:
         result = zlib.crc32(bytes(0))
@@ -125,36 +113,36 @@ def checksum(pathname, buffer_size=65536):
 
     return result & 0xffffffff
 
-def seed(source, target, config_file, skip_image_prefix=True):
+def seed(source, target, config_path, skip_image_prefix=True):
     """
     Read CSV files into a database backend.
 
     :param config_file: Configuration file.
     :param source: Directory containing subdirectories that contain CSV files.
     :param target: Connection string for the database.
-    :param skip_image_prefix: True if the prefix of image table name should be excluded
+    :param  : True if the prefix of image table name should be excluded
      from the names of columns from per image table
     """
-    config_file = cytominer_database.utils.read_config(config_file)
+    config_file = cytominer_database.utils.read_config(config_path)
 
     # list the subdirectories that contain CSV files
     directories = sorted(list(cytominer_database.utils.find_directories(source)))
 
     for directory in directories:
-
         # get the image CSV and the CSVs for each of the compartments
         try:
             compartments, image = cytominer_database.utils.validate_csv_set(config_file, directory)
         except IOError as e:
             click.echo(e)
-
+            continue
+        except sqlalchemy.exc.DatabaseError as e:
+            click.echo(e)
             continue
 
         # get a unique identifier for the image CSV. This will later be used as the TableNumber column
         # the casting to int is to allow the database to be readable by CellProfiler Analyst, which
         # requires TableNumber to be an integer.
         identifier = checksum(image)
-
         name, _ = os.path.splitext(config_file["filenames"]["image"])
 
         # ingest the image CSV
@@ -163,7 +151,6 @@ def seed(source, target, config_file, skip_image_prefix=True):
                  skip_table_prefix=skip_image_prefix)
         except sqlalchemy.exc.DatabaseError as e:
             click.echo(e)
-
             continue
 
         # ingest the CSV for each compartment