Source code for

#!/usr/bin/env python3

# std
import copy
import json
import logging
import pandas as pd
from pathlib import PurePath, Path
from typing import Union, Optional

# 3rd
import sqlalchemy

# ours
from clusterking.util.metadata import turn_into_nested_dict, nested_dict
from clusterking.util.log import get_logger
from clusterking.util.cli import handle_overwrite

[docs]class DFMD(object): """DFMD = DataFrame with MetaData. This class bundles a pandas dataframe together with metadata and provides methods to save and load such an object. """
[docs] def __init__( self, path: Optional[Union[str, PurePath]] = None, log: Optional[Union[str, logging.Logger]] = None, ): """ Initialize a DFMD object. Args: path: Optional: load from this file (specified as string or :class:`pathlib.PurePath`) log: Optional: instance of :py:class:`logging.Logger` or name of logger to be created """ # These are the three attributes of this class #: This will hold all the configuration that we will write out = None #: :py:class:`pandas.DataFrame` to hold all of the results self.df = None # type: Optional[pd.DataFrame] #: Instance of :py:class:`logging.Logger` self.log = None # todo: remember path? if not path: # Initialize blank = nested_dict() self.df = pd.DataFrame() self.log = None else: self._load(path) # Overwrite log if user wants that. if isinstance(log, logging.Logger): self.log = log elif isinstance(log, str): self.log = get_logger(log) elif log is None: if not self.log: self.log = get_logger("DFMD") else: raise ValueError( "Unsupported type '{}' for 'log' argument.".format(type(log)) )
# ************************************************************************** # Loading # ************************************************************************** def _load(self, path: Union[str, PurePath]) -> None: """Load input file as created by :py:meth:``. Args: path: Path to input file Returns: None """ path = Path(path) if not path.is_file(): raise FileNotFoundError("File '{}' doesn't exist.".format(path)) engine = sqlalchemy.create_engine("sqlite:///" + str(path.resolve())) self.df = pd.read_sql_table("df", engine) self.df.set_index("index", inplace=True) md_json = pd.read_sql_table("md", engine)["md"][0] = turn_into_nested_dict(json.loads(md_json)) # ************************************************************************** # Writing # **************************************************************************
[docs] def write(self, path: Union[str, PurePath], overwrite="ask"): """Write output files. Args: path: Path to output file overwrite: How to proceed if output file already exists: 'ask' (ask interactively for approval if we have to overwrite), 'overwrite' (overwrite without asking), 'raise' (raise Exception if file exists). Default is 'ask'. Returns: None """ path = Path(path) handle_overwrite([path], behavior=overwrite, log=self.log) if not path.parent.is_dir(): self.log.debug("Creating directory '{}'.".format(path.parent)) path.parent.mkdir(parents=True) engine = sqlalchemy.create_engine("sqlite:///" + str(path)) self.df.to_sql("df", engine, if_exists="replace") # todo: perhaps it's better to use pickle in the future? md_json = json.dumps(, sort_keys=True, indent=4) md_df = pd.DataFrame({"md": [md_json]}) md_df.to_sql("md", engine, if_exists="replace")
[docs] def copy(self, deep=True, data=True, memo=None): """Make a copy of this object. Args: deep: Make a deep copy (default True). If this is disabled, any change to the copy will also affect the original. data: Also copy data memo: Returns: New object. """ new = type(self)() if data: if deep: # Pycharm doesn't seem to recognize the memo argument: # noinspection PyArgumentList new.df = copy.deepcopy(self.df, memo) else: new.df = copy.copy(self.df) if deep: # noinspection PyArgumentList = copy.deepcopy(, memo) else: = copy.copy( new.log = copy.copy(self.log) if deep and memo is not None: memo[id(self)] = new return new
# ************************************************************************** # Magic methods # ************************************************************************** def __copy__(self): return self.copy() def __deepcopy__(self, memo): return self.copy(deep=True)