Skip to content
Snippets Groups Projects
Commit 71164cae authored by mjbonifa's avatar mjbonifa
Browse files

(docs) updated module descriptions

parent f29f45bc
No related branches found
No related tags found
No related merge requests found
"""
logging_config.py
This module defines functions to setup logging for acmc across all module.
Constants
- DEFAULT_LOG_FILE: The default acmc application log filename.
"""
import pandas as pd import pandas as pd
import logging import logging
DEFAULT_LOG_FILE = "acmc.log" DEFAULT_LOG_FILE = "acmc.log" # default log filename
def setup_logger(log_level: int = logging.INFO): def setup_logger(log_level: int = logging.INFO):
"""Sets up logger as a singleton outputing to file and sysout syserr""" """Sets up acmc logger as a singleton outputing to file and sysout syserr."""
# Create a logger # Create a logger
logger = logging.getLogger("acmc_logger") logger = logging.getLogger("acmc_logger")
logger.setLevel(logging.INFO) logger.setLevel(logging.INFO)
...@@ -20,7 +30,6 @@ def setup_logger(log_level: int = logging.INFO): ...@@ -20,7 +30,6 @@ def setup_logger(log_level: int = logging.INFO):
stream_handler.setLevel(logging.INFO) stream_handler.setLevel(logging.INFO)
# Create a formatter for how the log messages should look # Create a formatter for how the log messages should look
# Add the formatter to both handlers # Add the formatter to both handlers
file_formatter = logging.Formatter( file_formatter = logging.Formatter(
"%(asctime)s - - %(levelname)s - %(message)s" "%(asctime)s - - %(levelname)s - %(message)s"
...@@ -37,9 +46,9 @@ def setup_logger(log_level: int = logging.INFO): ...@@ -37,9 +46,9 @@ def setup_logger(log_level: int = logging.INFO):
def set_log_level(log_level: int): def set_log_level(log_level: int):
"""Sets the log level for the acmc logger""" """Sets the log level for the acmc logger."""
logger = logging.getLogger("acmc_logger") logger = logging.getLogger("acmc_logger")
logger.setLevel(log_level) # Set logger level logger.setLevel(log_level)
# Also update handlers to match the new level # Also update handlers to match the new level
for handler in logger.handlers: for handler in logger.handlers:
......
""" """
OMOP Module omop module
================ ================
This module provides functionality to manage OMOP vocabularies. This module provides functionality to manage OMOP vocabularies.
""" """
...@@ -62,9 +62,18 @@ omop_vocab_types = { ...@@ -62,9 +62,18 @@ omop_vocab_types = {
} }
# Populate SQLite3 Database with default OMOP CONCEPTS
def install(omop_zip_file: str, version: str): def install(omop_zip_file: str, version: str):
"""Installs the OMOP release csv files in a file-based sql database""" """"Installs the OMOP release csv files in a file-based sql database
Args:
omop_zip_file (str): vocabularies zip file distributed by OHDSI Athena
version (str): version of the vocabularies distributed by OHDSI Athena
Raises:
ValueError: if the zip file does not exist
ValueError: if the file is not a zip file
Exception: if error reading omop csv files
"""
logger.info(f"Installing OMOP from zip file: {omop_zip_file}") logger.info(f"Installing OMOP from zip file: {omop_zip_file}")
omop_zip_path = Path(omop_zip_file) omop_zip_path = Path(omop_zip_file)
...@@ -73,6 +82,7 @@ def install(omop_zip_file: str, version: str): ...@@ -73,6 +82,7 @@ def install(omop_zip_file: str, version: str):
msg = f"{omop_zip_path} does not exist." msg = f"{omop_zip_path} does not exist."
logger.error(msg) logger.error(msg)
raise ValueError(msg) raise ValueError(msg)
if not zipfile.is_zipfile(omop_zip_path): if not zipfile.is_zipfile(omop_zip_path):
msg = f"Error: {omop_zip_path} is not a valid ZIP file." msg = f"Error: {omop_zip_path} is not a valid ZIP file."
logger.error(msg) logger.error(msg)
...@@ -127,7 +137,11 @@ def install(omop_zip_file: str, version: str): ...@@ -127,7 +137,11 @@ def install(omop_zip_file: str, version: str):
def write_version_file(version: str): def write_version_file(version: str):
"""Writes the OMOP vocaburaries and version to a file""" """Writes the OMOP vocaburaries and version to a file
Args:
version (str): version of the vocabularies distributed by OHDSI Athena
"""
vocabularies["version"] = version vocabularies["version"] = version
with open(VERSION_PATH, "w") as file: with open(VERSION_PATH, "w") as file:
yaml.dump( yaml.dump(
...@@ -141,7 +155,14 @@ def write_version_file(version: str): ...@@ -141,7 +155,14 @@ def write_version_file(version: str):
def clear(db_path: Path): def clear(db_path: Path):
"""Clears the OMOP sql database""" """Clears the OMOP sql database
Args:
db_path (Path): the path to the omop sqllite database
Raises:
FileNotFoundError: if the omop sqllite database does not exist
"""
logger.info(f"Clearing OMOP data from database") logger.info(f"Clearing OMOP data from database")
if not db_path.is_file(): if not db_path.is_file():
raise FileNotFoundError(f"Error: OMOP DB file '{db_path}' does not exist.") raise FileNotFoundError(f"Error: OMOP DB file '{db_path}' does not exist.")
...@@ -161,7 +182,15 @@ def clear(db_path: Path): ...@@ -161,7 +182,15 @@ def clear(db_path: Path):
def delete(db_path: Path): def delete(db_path: Path):
"""Deletes the OMOP sql database""" """Deletes the OMOP sql database
Args:
db_path (Path): the path to the omop sqllite database
Raises:
FileNotFoundError: if the omop sqllite database does not exist
"""
logger.info(f"Deleting OMOP database") logger.info(f"Deleting OMOP database")
if not db_path.is_file(): if not db_path.is_file():
raise FileNotFoundError(f"Error: OMOP DB file '{db_path}' does not exist.") raise FileNotFoundError(f"Error: OMOP DB file '{db_path}' does not exist.")
...@@ -171,7 +200,16 @@ def delete(db_path: Path): ...@@ -171,7 +200,16 @@ def delete(db_path: Path):
def table_exists(cursor: sqlite3.Cursor, table_name: str) -> bool: def table_exists(cursor: sqlite3.Cursor, table_name: str) -> bool:
# Query to check if the table exists """Query to check if the table exists
Args:
cursor (sqlite3.Cursor): a sqllite database cursor
table_name (str): the table name to check
Returns:
bool: true if table exists
"""
cursor.execute( cursor.execute(
""" """
SELECT name SELECT name
...@@ -188,7 +226,16 @@ def table_exists(cursor: sqlite3.Cursor, table_name: str) -> bool: ...@@ -188,7 +226,16 @@ def table_exists(cursor: sqlite3.Cursor, table_name: str) -> bool:
def vocab_exists(cursor: sqlite3.Cursor, vocab_id: str) -> bool: def vocab_exists(cursor: sqlite3.Cursor, vocab_id: str) -> bool:
# Query to check if the table exists """Query to check if the vocabulary exists
Args:
cursor (sqlite3.Cursor): a sqllite database cursor
vocab_id (str): the vocabulary id to check
Returns:
bool: true if vocabulary id exists
"""
cursor.execute( cursor.execute(
""" """
SELECT vocabulary_id SELECT vocabulary_id
...@@ -205,6 +252,16 @@ def vocab_exists(cursor: sqlite3.Cursor, vocab_id: str) -> bool: ...@@ -205,6 +252,16 @@ def vocab_exists(cursor: sqlite3.Cursor, vocab_id: str) -> bool:
def concept_set_exist(cursor: sqlite3.Cursor, concept_set_name: str) -> bool: def concept_set_exist(cursor: sqlite3.Cursor, concept_set_name: str) -> bool:
"""Query to check if the concept set exists
Args:
cursor (sqlite3.Cursor): a sqllite database cursor
concept_set_name (str): the concept set name to check
Returns:
bool: true if concept set exists
"""
query = f"SELECT EXISTS (SELECT 1 FROM CONCEPT_SET WHERE concept_set_name = ?)" query = f"SELECT EXISTS (SELECT 1 FROM CONCEPT_SET WHERE concept_set_name = ?)"
cursor.execute(query, (concept_set_name,)) cursor.execute(query, (concept_set_name,))
...@@ -212,7 +269,20 @@ def concept_set_exist(cursor: sqlite3.Cursor, concept_set_name: str) -> bool: ...@@ -212,7 +269,20 @@ def concept_set_exist(cursor: sqlite3.Cursor, concept_set_name: str) -> bool:
return cursor.fetchone()[0] == 1 return cursor.fetchone()[0] == 1
def export(map_path: Path, export_path: Path, version: str, omop_metadata) -> Path: def export(map_path: Path, export_path: Path, version: str, omop_metadata: dict) -> Path:
"""Export concept sets to omop database in csv format
Args:
map_path (Path): path to the acmc map directory containing concept sets in csv format
export_path (Path): path to the directory where the omop database csv files are to be written
version (str): phenotype version for omop vocabulary version
omop_metadata (dict): phenotype omop metadata for omop vocabulary metadata
Returns:
Path: path to the exported sqllite database
"""
logger.debug(f"exporting with metadata {omop_metadata} at version {version}") logger.debug(f"exporting with metadata {omop_metadata} at version {version}")
# copy the baseline omop database # copy the baseline omop database
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment