Merge pull request #10 from CogStack/devel

Merge devel into master.

Author: vladd-bit

.gitattributes

@@ -0,0 +1,44 @@
+###############################
+# Git Line Endings            #
+###############################
+
+# Set default behaviour to automatically normalize line endings.
+* text=auto
+
+# Force batch scripts to always use CRLF line endings so that if a repo is accessed
+# in Windows via a file share from Linux, the scripts will work.
+*.{cmd,[cC][mM][dD]} text eol=crlf
+*.{bat,[bB][aA][tT]} text eol=crlf
+
+# Force bash scripts to always use LF line endings so that if a repo is accessed
+# in Unix via a file share from Windows, the scripts will work.
+*.sh text eol=lf
+
+###############################
+# Git Large File System (LFS) #
+###############################
+
+# Archives
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.br filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+
+# Documents
+*.pdf filter=lfs diff=lfs merge=lfs -text
+
+# Images
+*.gif filter=lfs diff=lfs merge=lfs -text
+*.ico filter=lfs diff=lfs merge=lfs -text
+*.jpg filter=lfs diff=lfs merge=lfs -text
+*.pdf filter=lfs diff=lfs merge=lfs -text
+*.png filter=lfs diff=lfs merge=lfs -text
+*.psd filter=lfs diff=lfs merge=lfs -text
+*.webp filter=lfs diff=lfs merge=lfs -text
+
+# Fonts
+*.woff2 filter=lfs diff=lfs merge=lfs -text
+
+# Other
+*.exe filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -1,4 +1,11 @@
+# IDE
 .DS_Store
 .idea
+.vscode
+
+# python
 *.pyc
 venv
+__pycache__
+venv-test
+.mypy_cache

Dockerfile

@@ -1,11 +1,12 @@
-FROM python:3.8
+FROM python:3.9
 
 WORKDIR /app
 
 # configure and install the required packages
 ENV PYTHONPATH="/app:${PYTHONPATH}"
 COPY ./requirements.txt /app
 
+RUN pip3 install --upgrade pip
 RUN pip3 install -r requirements.txt
 
 # copy the source and config files

README.md

@@ -15,7 +15,6 @@ Firstly install the Python libraries specified in `requirements.txt`.
 To run:
 `python main.py --config config/config.yml`
 
-
 # Configuration
 The ingestion process properties are configured in `config.yml`.
 
@@ -31,18 +30,36 @@ Entries under `sink` key specify the ElasticSearch sink.
 
 #### Security and SSL certificates
 This only applies when ElasticSearch cluster is using X-Pack / Open Distro and requires secure connections with using SSL certificates. Under the key `source` and/or `sink` can be optionally specified `security` entry that will contain necessary configuration -- these are:
+' `ca-file-path` -- path to CA certificate file (PEM) used solely with SSL, (DO NOT USE THIS along with the other ca/client parameters mentioned underneath, use it solely)
 - `ca-certs-path` -- the path to CA certificates file (PEM),
 - `client-cert-path` -- the path to client certificate file (PEM),
 - `client-key-path` -- the path to client key (PEM).
 
-### Extra ES connection options
+### Extra ES sink/source connection options
 Entires under the key `extra_params` are optional, useful for test cases or deployments where we only use internal resources:
 - `use_ssl` -- use ssl connection
 - `verify_certs` -- verify SSL certificates 
 
+- `credentials`
+    - `username` and `password` can be used to provide connection credentials
+    - `use-api-key` if this is enabled the username and password fields will be used as api_id and api_key 
+
+### NLP service
+- `endpoint-url` 
+- `endpoint-request-mode` , this is either left empty, or in case of use with the GATE NLP Annie annotation service it should be set to `gate-nlp`
+- `use-bulk-indexing` ingest in bulk mode (1000 docs / bulk chunk), 
+
+- `credentials`
+    - `username` and `password` can be used to provide connection credentials
+
 ### Fields mapping
 Entries under `mapping` key define the mapping of the document fields for the ingestion.
 
+The sub-entry `index-ingest-mode` defines: 
+- `same-index`: `False` , set to `True` if you wish to ingest annotations into the same index
+- `use-nested-objects` : 'False` , set to True if you wish to ingest into the same index with nested object type, useful but beware of search query speed impact
+- `es-nested-object-schema-mapping` : for medcat annotations use `medcat-separate-index` or `medcat-nested-object` , for GATE-nlp use `gate-nlp-separate-index` or `gate-nlp-nested-object`
+
 The sub-entry `source` defines the field names that contain:
 - `text-field` - the free text to be processed, 
 - `docid-field` - the unique identifier of the document,
@@ -57,7 +74,7 @@ The sub-entry `batch` defines the possible portion of documents to be processed
 - `threads` - the number of processing threads to speed up the ingestion.
 
 The sub-entry `sink` specifies additional options during sending the processed annotations:
-- `split-index-by-field` - the name of the field in the returned annotations the value of which will be used as a prefix for the index name (e.g., used to send annotations of different types to separate indices).
+- `split-index-by-field` - the name of the field in the returned annotations the value of which will be used as a prefix for the index name (e.g., used to send annotations of different types to separate indices). If you don't want this functionality simply leave the field empty, otherwise , to split by annotation type use `type`
 
 The sub-entry `nlp` specifies additional options during processing the documents with NLP:
 - `skip-processed-doc-check` - whether to skip checking for already processed documents in ElasticSearch,

config/config.yml

@@ -1,49 +1,72 @@
 source:
   es:
-    hosts: ["https://cogstack:cogstack@localhost:9200"]
-    index-name: 'cogstack_observations_view'
-    extra_params:
-      use_ssl: False
-      verify_certs: False
+    hosts: ["http://127.0.0.1:9200"]
+    credentials:
+      username : "elastic"
+      password : "admin"
+      use-api-key : False
+    index-name: 'medical_reports_text'
+    extra-params:
+      use-ssl: False
+      verify-certs: False
     security:
+      ca-file-path : "/app/config/cert.pem"
       ca-certs-path: "/app/config/root-ca.pem"
       client-cert-path: "/app/config/client.pem"
       client-key-path: "/app/config/client.key"
 
 sink:
   es:
-    hosts: ["http://cogstack:cogstack@localhost:9200"]
-    index-name: 'cogstack_atomic_annotations'
-    extra_params:
-      use_ssl: False
-      verify_certs: False
+    hosts: ["http://127.0.0.1:9200"]
+    credentials:
+      username : "elastic"
+      password : "admin"
+      use-api-key : False
+    index-name: 'medical_reports_text_annotations'
+    extra-params:
+      use-ssl: False
+      verify-certs: False
     security:
+      ca-file-path : "/app/config/cert.pem"
       ca-certs-path: "/app/config/root-ca.pem"
       client-cert-path: "/app/config/client.pem"
       client-key-path: "/app/config/client.key"
 
 nlp-service:
-  endpoint-url: 'http://localhost:8095/api/process'
+  endpoint-url: 'http://localhost:5555/api/process'
+  endpoint-request-mode : ''
+  use-bulk-indexing : True
+  annotation-response:
+    dict-key : "annotations"
+    result-key : "result"
+  credentials :  
+    username : ""
+    password : ""
 
 mapping:
+  index-ingest-mode:
+    same-index: False
+    use-nested-objects: False
+    es-nested-object-schema-mapping : ""
   source:
-    text-field: 'document_content'
-    docid-field: 'encounter_id'
+    text-field: 'document'
+    docid-field: '_id'
     persist-fields:
-      - 'encounter_id'
-      - 'patient_id'
-      - 'encounter_start'
-      - 'encounter_end'
+     - '_id'
     batch:
-      date-field: 'encounter_start'
+      date-field: 'dct'
       date-format: 'yyyy-MM-dd'
       python-date-format: '%Y-%m-%d' 
-      interval: 30 
-      date-start: '2010-01-01'
-      date-end: '2018-06-01'
+      interval: 30
+      date-start: '1999-01-01'
+      date-end: '2021-02-01'
       threads: 8
   sink:
-    split-index-by-field: 'type'
+    split-index-by-field: '' # 'type', splits into different indices with separate prefix
   nlp:
-    skip-processed-doc-check: 'true'
+    skip-processed-doc-check: True
     annotation-id-field: 'id'
+
+# DEBUG = 10 , INFO = 20, WARNING = 30, ERROR = 40, CRITICAL = 50
+logging-level: "20"
+

docker-compose.yml

@@ -1,6 +1,5 @@
 version: "3.5"
 services:
-
  annotation-ingester:
     build: .
     shm_size : 128mb

ingester/__main__.py

@@ -8,7 +8,6 @@
 from ingester.nlp_service import *
 from ingester.annotations_indexer import *
 
-
 class AppConfig:
     """
     The configuration file for the indexer application
@@ -32,7 +31,6 @@ def __init__(self, file_path):
         except FileNotFoundError:
             raise Exception("Cannot open configuration file")
 
-
 if __name__ == "__main__":
     # parse the input parameters
     parser = argparse.ArgumentParser(description='ElasticSearch-to-ElasticSearch annotations indexer')
@@ -43,76 +41,84 @@ def __init__(self, file_path):
         parser.print_usage()
         exit(0)
 
-    # setup logging
-    log_format = '[%(asctime)s] [%(levelname)s] %(name)s: %(message)s'
-    logging.basicConfig(format=log_format, level=logging.INFO)
-
     try:
         config = AppConfig(args.config)
 
-        # set up Elastic logger for the initialization time
-        logging.getLogger('elasticsearch').setLevel(level=logging.INFO)
+        # setup logging
+        log_format = '[%(asctime)s] [%(levelname)s] %(name)s: %(message)s'
+        if config.params['logging-level']:
+            logging.basicConfig(format=log_format, level=int(config.params['logging-level']))
+        else:
+            logging.basicConfig(format=log_format, level=logging.INFO)
+        logging.getLogger('elasticsearch')
 
         # initialize the elastic source
         source_params = config.params['source']
-        if 'security' in source_params['es']:
-            src_sec = source_params['es']['security']
-            source_ssl_config = SslConnectionConfig(ca_certs_path=src_sec['ca-certs-path'],
-                                                    client_cert_path=src_sec['client-cert-path'],
-                                                    client_key_path=src_sec['client-key-path'])
-            es_source_conf = ElasticConnectorConfig(hosts=source_params['es']['hosts'],
-                                                    ssl_config=source_ssl_config)
-        elif 'extra_params' in source_params['es']:
-            e_params = source_params['es']['extra_params']
-            es_source_conf = ElasticConnectorConfig(hosts=source_params['es']['hosts'], extra_params=e_params)
-        else:
-            es_source_conf = ElasticConnectorConfig(hosts=source_params['es']['hosts'])
-        es_source_conn = ElasticConnector(es_source_conf)
 
+        source_credentials = source_params['es']['credentials'] if source_params['es']['credentials'] else None
+        source_extra_params = source_params['es']['extra-params'] if source_params['es']['extra-params'] else None
+        source_security = source_params['es']['security']  if source_params['es']['security'] else None
+
+        source_ssl_config = SslConnectionConfig(ca_file_path=source_security['ca-file-path'],
+                                                ca_certs_path=source_security['ca-certs-path'],
+                                                client_cert_path=source_security['client-cert-path'],
+                                                client_key_path=source_security['client-key-path'])
+
+        es_source_conf = ElasticConnectorConfig(hosts=source_params['es']['hosts'], credentials=source_credentials, extra_params=source_extra_params,
+                                                   ssl_config=source_ssl_config)
+
+        es_source_conn = ElasticConnector(es_source_conf)
         es_source = ElasticRangedIndexer(es_source_conn, source_params['es']['index-name'])
 
         # initialize NLP service
-        nlp_service = NlpService(config.params['nlp-service']['endpoint-url'])
+        nlp_service = NlpService(config.params['nlp-service']['endpoint-url'],
+          endpoint_request_mode=config.params['nlp-service']['endpoint-request-mode'],
+          use_bulk_indexing=config.params['nlp-service']['use-bulk-indexing'],
+          username=config.params['nlp-service']['credentials']['username'],
+          password=config.params['nlp-service']['credentials']['password'])
 
         # initialize the elastic sink
         sink_params = config.params['sink']
-        if 'security' in sink_params['es']:
-            sink_sec = sink_params['es']['security']
-            sink_ssl_config = SslConnectionConfig(ca_certs_path=sink_sec['ca-certs-path'],
-                                                  client_cert_path=sink_sec['client-cert-path'],
-                                                  client_key_path=sink_sec['client-key-path'])
-
-            es_sink_conf = ElasticConnectorConfig(hosts=sink_params['es']['hosts'],
-                                                  ssl_config=sink_ssl_config)
-        elif 'extra_params' in sink_params['es']:
-            e_params = sink_params['es']['extra_params']
-            es_sink_conf = ElasticConnectorConfig(hosts=sink_params['es']['hosts'], extra_params=e_params)
-        else:
-            es_sink_conf = ElasticConnectorConfig(hosts=sink_params['es']['hosts'])
+
+        sink_credentials = sink_params['es']['credentials'] if sink_params['es']['credentials'] else None
+        sink_extra_params = sink_params['es']['extra-params'] if sink_params['es']['extra-params'] else None
+        sink_security = sink_params['es']['security']  if sink_params['es']['security'] else None
+
+        sink_ssl_config = SslConnectionConfig(ca_file_path=sink_security['ca-file-path'],
+                                              ca_certs_path=sink_security['ca-certs-path'],
+                                              client_cert_path=sink_security['client-cert-path'],
+                                              client_key_path=sink_security['client-key-path'])
+
+        es_sink_conf =  ElasticConnectorConfig(hosts=sink_params['es']['hosts'], credentials=sink_credentials, extra_params=sink_extra_params,
+                                                   ssl_config=sink_ssl_config)
 
         es_sink_conn = ElasticConnector(es_sink_conf)
         es_sink = ElasticIndexer(es_sink_conn, sink_params['es']['index-name'])
 
-
         # initialize the indexer
         mapping = config.params['mapping']
-        indexer = BatchAnnotationsIndexer(nlp_service=nlp_service,
+        annoation_indexer_config = AnnotationIndexerConfig(nlp_service=nlp_service,
                                           source_indexer=es_source,
                                           source_text_field=mapping['source']['text-field'],
                                           source_docid_field=mapping['source']['docid-field'],
                                           source_fields_to_persist=mapping['source']['persist-fields'],
-                                          split_index_by_field=mapping['sink']['split-index-by-field'],
                                           sink_indexer=es_sink,
+                                          split_index_by_field=mapping['sink']['split-index-by-field'],
                                           source_batch_date_field=mapping['source']['batch']['date-field'],
                                           batch_date_format=mapping['source']['batch']['date-format'],
-                                          skip_doc_check=mapping['nlp']['skip-processed-doc-check'].lower() == "true",
+                                          skip_doc_check=mapping['nlp']['skip-processed-doc-check'],
                                           nlp_ann_id_field=mapping['nlp']['annotation-id-field'],
                                           threads=mapping['source']['batch']['threads'],
                                           python_date_format=mapping['source']['batch']['python-date-format'],
-                                          interval=mapping['source']['batch']['interval'])
+                                          interval=mapping['source']['batch']['interval'],
+                                          same_index_ingest=mapping['index-ingest-mode']['same-index'],
+                                          use_nested_objects=mapping['index-ingest-mode']['use-nested-objects'],
+                                          es_nested_object_schema_mapping=mapping['index-ingest-mode']['es-nested-object-schema-mapping'])
+                                          
+        indexer = BatchAnnotationsIndexer(annoation_indexer_config)
+
     except Exception as e:
-        log = logging.getLogger('main')
-        log.error("Cannot initialize the application: " + str(e))
+        logging.error("Cannot initialize the application: " + str(e))
         exit(1)
 
     # set up the Elastic logger to be more verbose and run the indexer