Notebook_for_wpipe2.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time, subprocess, os\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "from sqlalchemy import create_engine, Column, Sequence, ForeignKey, func\n",
    "from sqlalchemy import Integer, BigInteger, Float, String, DateTime, Boolean\n",
    "\n",
    "from sqlalchemy.sql import select\n",
    "from sqlalchemy.orm import sessionmaker, relationship, backref\n",
    "from sqlalchemy.ext.declarative import declarative_base\n",
    "from sqlalchemy.dialects import postgresql\n",
    "\n",
    "from sqlalchemy_utils import database_exists, create_database, drop_database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "username = 'rubab'\n",
    "dbname = 'tempDB1'\n",
    "\n",
    "engine = create_engine('postgres://%s@localhost/%s'%(username,dbname))\n",
    "\n",
    "if not database_exists(engine.url):\n",
    "    create_database(engine.url)\n",
    "else:\n",
    "    drop_database(engine.url)\n",
    "    create_database(engine.url)\n",
    "    \n",
    "Session = sessionmaker()\n",
    "Session.configure(bind=engine)\n",
    "session = Session()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "Base = declarative_base()\n",
    "\n",
    "class BaseMixin():\n",
    "    def create(self):\n",
    "        session.add(self)\n",
    "        session.commit()\n",
    "        return self\n",
    "    \n",
    "    def getOpt(self,how='sql'):\n",
    "        rs = self.options\n",
    "        return rs\n",
    "\n",
    "class Options(Base):\n",
    "    __tablename__= 'options'\n",
    "    \n",
    "    opt_id = Column(BigInteger, Sequence('opt_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "\n",
    "    target_id = Column(Integer, ForeignKey('targets.target_id'))\n",
    "    dp_id     = Column(Integer, ForeignKey('data_products.dp_id'))\n",
    "    job_id    = Column(Integer, ForeignKey('jobs.job_id'))\n",
    "    event_id  = Column(Integer, ForeignKey('events.event_id'))\n",
    "    \n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    value = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    \n",
    "    def __init__(self, name='any', value=0):\n",
    "        self.name = str(name)\n",
    "        self.value = str(value)\n",
    "        \n",
    "    def create(self,opt={'any':1}):\n",
    "        opts = []\n",
    "        for item in opt.items():\n",
    "            opts.append(Options(item[0],item[1]))\n",
    "        session.add_all(opts)\n",
    "        session.commit()\n",
    "        return opts\n",
    "\n",
    "class User(BaseMixin,Base):\n",
    "    __tablename__= 'users'\n",
    "    user_id = Column(Integer, Sequence('user_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    name = Column(postgresql.VARCHAR(32),nullable=False)\n",
    "    \n",
    "    pipelines = relationship('Pipeline',\n",
    "                    backref=backref('users',\n",
    "                    uselist=True,passive_updates=False,\n",
    "                    cascade='delete,all'))\n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "    \n",
    "    def __init__(self,name='any'):\n",
    "        self.name = str(name)\n",
    "\n",
    "    def add_pipeline(self,obj):\n",
    "        self.pipelines.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    @staticmethod\n",
    "    def get(user_name,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(User)\\\n",
    "            .filter_by(name=str(user_name)).one()\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([User])\n",
    "                    .where(User.name==str(user_name))\n",
    "                           ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs\n",
    "    \n",
    "\n",
    "class Pipeline(BaseMixin,Base):\n",
    "    __tablename__= 'pipelines'\n",
    "    pipeline_id = Column(Integer, Sequence('pipeline_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    \n",
    "    user_id = Column(Integer, ForeignKey('users.user_id'))\n",
    "    \n",
    "    targets = relationship('Target',\n",
    "                        backref=backref('pipelines',\n",
    "                        uselist=True,passive_updates=False,\n",
    "                        cascade='delete,all'))\n",
    "    \n",
    "    tasks = relationship('Task',\n",
    "                        backref=backref('pipelines',\n",
    "                        uselist=True,passive_updates=False,\n",
    "                        cascade='delete,all'))\n",
    "    \n",
    "    software_root = Column(postgresql.VARCHAR(256))\n",
    "    data_root = Column(postgresql.VARCHAR(256))\n",
    "    pipe_root = Column(postgresql.VARCHAR(256))\n",
    "    config_root = Column(postgresql.VARCHAR(256))\n",
    "    description = Column(postgresql.VARCHAR(512))\n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "\n",
    "    def __init__(self,name='any',software_root='',\n",
    "                 data_root='',pipe_root='',config_root='',\n",
    "                 description=''):\n",
    "        self.name = str(name)  \n",
    "        self.software_root = str(software_root)\n",
    "        self.data_root = str(data_root)\n",
    "        self.pipe_root = str(pipe_root)\n",
    "        self.config_root = str(config_root)\n",
    "        self.description = str(description)\n",
    "    \n",
    "    def add_target(self,obj,create_dir=False):\n",
    "        self.targets.append(obj)\n",
    "        obj.add_paths(self.pipeline_id,\n",
    "                      create_dir)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    def add_task(self,obj):\n",
    "        self.tasks.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "                                           \n",
    "    @staticmethod        \n",
    "    def get(pipeline_id,how='sql'):\n",
    "        if how=='sql':            \n",
    "            rs = session.query(Pipeline).get(int(pipeline_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([Pipeline])\n",
    "                        .where(Pipeline.pipeline_id==int(pipeline_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs\n",
    "\n",
    "\n",
    "class Target(BaseMixin,Base):\n",
    "    __tablename__= 'targets'\n",
    "    \n",
    "    target_id = Column(Integer, Sequence('target_id_seq'),\n",
    "                       primary_key=True, nullable=False)\n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    \n",
    "    pipeline_id = Column(Integer, ForeignKey('pipelines.pipeline_id'))\n",
    "    \n",
    "    configurations = relationship('Configuration',\n",
    "                        backref=backref('targets',\n",
    "                        uselist=True,passive_updates=False,\n",
    "                        cascade='delete,all'))  \n",
    "    \n",
    "    options = relationship('Options',\n",
    "                           backref=backref('targets',\n",
    "                           uselist=True,passive_updates=False,\n",
    "                           cascade='delete,all'))\n",
    "    \n",
    "    relativepath = Column(postgresql.VARCHAR(256))\n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "    \n",
    "    def __init__(self,name='any'):\n",
    "        self.name = str(name)\n",
    "        \n",
    "    def add_config(self,obj,create_dir=False):\n",
    "        self.configurations.append(obj)\n",
    "        obj.add_paths(self.target_id,\n",
    "                      create_dir)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    def add_options(self,obj):\n",
    "        for opt in obj:\n",
    "            self.options.append(opt)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    def add_paths(self,pipeline_id,create_dir=False):\n",
    "        pipeline = Pipeline.get(int(pipeline_id))\n",
    "        self.relativepath = str(pipeline.data_root)+'/'+str(self.name)\n",
    "        if create_dir:\n",
    "            _t = subprocess.run(['mkdir', '-p', str(self.relativepath)],\n",
    "                                stdout=subprocess.PIPE)\n",
    "        return\n",
    "    \n",
    "    @staticmethod\n",
    "    def get(target_id,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(Target).get(int(target_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([Target])\n",
    "                        .where(Target.target_id==int(target_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs\n",
    "           \n",
    "class Configuration(BaseMixin,Base):\n",
    "    __tablename__= 'configurations'\n",
    "    \n",
    "    config_id = Column(Integer, Sequence('config_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    \n",
    "    target_id = Column(Integer, ForeignKey('targets.target_id'))\n",
    "    \n",
    "    data_products = relationship('DataProduct',\n",
    "                           backref=backref('configurations',\n",
    "                           uselist=True,passive_updates=False,\n",
    "                           cascade='delete,all'))\n",
    "    \n",
    "    parameters = relationship('Parameters',\n",
    "                           backref=backref('configurations',\n",
    "                           uselist=True,passive_updates=False,\n",
    "                           cascade='delete,all')) \n",
    "    \n",
    "    jobs = relationship('Job',\n",
    "                    backref=backref('configurations',\n",
    "                    uselist=False,passive_updates=False))\n",
    "    \n",
    "    relativepath = Column(postgresql.VARCHAR(256))\n",
    "    logpath = Column(postgresql.VARCHAR(256))\n",
    "    confpath = Column(postgresql.VARCHAR(256))\n",
    "    rawpath = Column(postgresql.VARCHAR(256))\n",
    "    procpath = Column(postgresql.VARCHAR(256))\n",
    "    description = Column(postgresql.VARCHAR(512))\n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "\n",
    "    def __init__(self,name='any',description=''):\n",
    "        self.name = str(name)\n",
    "        self.description = str(description)\n",
    "\n",
    "    def add_dp(self,obj):\n",
    "        self.data_products.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "\n",
    "    def add_params(self,obj):\n",
    "        for param in obj:\n",
    "            self.parameters.append(param)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    def add_paths(self,target_id,create_dir=False):\n",
    "        target = Target.get(int(target_id))\n",
    "        self.relativepath = str(target.relativepath)\n",
    "        self.logpath = str(target.relativepath)+'/log_'+str(self.name)\n",
    "        self.confpath = str(target.relativepath)+'/conf_'+str(self.name)\n",
    "        self.rawpath = str(target.relativepath)+'/raw_'+str(self.name)\n",
    "        self.procpath = str(target.relativepath)+'/proc_'+str(self.name)\n",
    "        \n",
    "        if create_dir:\n",
    "            for _path in [self.rawpath,self.confpath,self.procpath,self.logpath]:\n",
    "                _t = subprocess.run(['mkdir', '-p', str(_path)], stdout=subprocess.PIPE)        \n",
    "        return\n",
    "    \n",
    "    def add_job(self,obj):\n",
    "        self.jobs.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    @staticmethod            \n",
    "    def get(config_id,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(Configuration).get(int(config_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([Configuration])\n",
    "                        .where(Configuration.config_id==int(config_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs    \n",
    "    \n",
    "    \n",
    "class Parameters(Base):\n",
    "    __tablename__= 'parameters'\n",
    "    \n",
    "    param_id = Column(BigInteger, Sequence('param_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "\n",
    "    config_id = Column(Integer, ForeignKey('configurations.config_id'))\n",
    "                        \n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    value = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    \n",
    "    def __init__(self, name='any', value=0):\n",
    "        self.name = str(name)\n",
    "        self.value = str(value)\n",
    "        \n",
    "    def create(self,param={'any':1}):\n",
    "        params = []\n",
    "        for item in param.items():\n",
    "            params.append(Parameters(item[0],item[1]))\n",
    "        session.add_all(params)\n",
    "        session.commit()\n",
    "        return params\n",
    "    \n",
    "    @staticmethod\n",
    "    def getParams(config_id):\n",
    "        with engine.connect() as conn:\n",
    "            _df = pd.read_sql_query(select([Parameters])\n",
    "                    .where(Parameters.config_id==int(config_id))\n",
    "                    ,conn)\n",
    "        return dict(zip(_df['name'],_df['value']))\n",
    "    \n",
    "    \n",
    "class DataProduct(BaseMixin,Base):\n",
    "    __tablename__= 'data_products'\n",
    "    \n",
    "    dp_id = Column(BigInteger, Sequence('dp_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    \n",
    "    config_id = Column(Integer, ForeignKey('configurations.config_id'))\n",
    "    \n",
    "    filename = Column(postgresql.VARCHAR(128))\n",
    "    relativepath = Column(postgresql.VARCHAR(256))\n",
    "    suffix = Column(postgresql.VARCHAR(8))\n",
    "    data_type = Column(postgresql.VARCHAR(16))\n",
    "    subtype = Column(postgresql.VARCHAR(16))\n",
    "    group = Column(postgresql.VARCHAR(8))\n",
    "    filtername = Column(postgresql.VARCHAR(8))\n",
    "    \n",
    "    ra = Column(Float)\n",
    "    dec = Column(Float)\n",
    "    pointing_angle = Column(Float)\n",
    "    \n",
    "    options = relationship('Options',\n",
    "                           backref=backref('data_products',\n",
    "                           uselist=True,passive_updates=False,\n",
    "                           cascade='delete,all'))    \n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "    \n",
    "    def __init__(self,filename='',relativepath='',group='',\n",
    "                 data_type='',subtype='',filtername='',\n",
    "                 ra=0,dec=0,pointing_angle=0):\n",
    "        self.filename = str(filename)\n",
    "        self.relativepath = str(relativepath)\n",
    "\n",
    "        _suffix = ' '\n",
    "        if '.' in filename:\n",
    "            _suffix = filename.split('.')[-1]\n",
    "        if _suffix not in ['fits','txt','head','cl',\n",
    "           'py','pyc','pl','phot','png','jpg','ps',\n",
    "           'gz','dat','lst','sh']:\n",
    "            _suffix = 'other'                                      \n",
    "        self.suffix = str(_suffix)\n",
    "        \n",
    "        if not(data_type): data_type = _suffix\n",
    "        self.data_type = str(data_type)\n",
    "        self.subtype = str(subtype)\n",
    "\n",
    "        if group not in ['proc','conf','log','raw']:\n",
    "            group = 'other'\n",
    "        self.group = str('other')\n",
    "\n",
    "        self.filtername = str(filtername)\n",
    "        self.ra = float(ra)\n",
    "        self.dec = float(dec)\n",
    "        self.pointing_angle = float(pointing_angle)\n",
    "\n",
    "    def add_options(self,obj):\n",
    "        for opt in obj:\n",
    "            self.options.append(opt)\n",
    "        session.commit()\n",
    "        return\n",
    "\n",
    "    @staticmethod            \n",
    "    def get(dp_id,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(DataProduct).get(int(dp_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([DataProduct])\n",
    "                        .where(DataProduct.dp_id==int(dp_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs    \n",
    "    \n",
    "    \n",
    "class Task(BaseMixin,Base):\n",
    "    __tablename__= 'tasks'\n",
    "    \n",
    "    task_id = Column(Integer, Sequence('task_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    \n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    \n",
    "    pipeline_id = Column(Integer, ForeignKey('pipelines.pipeline_id'))\n",
    "\n",
    "    masks = relationship('Mask',\n",
    "                        backref=backref('tasks',\n",
    "                        uselist=True,passive_updates=False,\n",
    "                        cascade='delete,all'))\n",
    "    \n",
    "    jobs = relationship('Job',\n",
    "                        backref=backref('tasks',\n",
    "                        uselist=True,passive_updates=False,\n",
    "                        cascade='delete,all'))\n",
    "    \n",
    "    nruns = Column(Float)\n",
    "    runtime = Column(Float)\n",
    "    is_exclusive = Column(Boolean)\n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "    \n",
    "    def __init__(self,name='',\n",
    "                 nruns=0,run_time=0,\n",
    "                 is_exclusive=0):\n",
    "        self.name = str(name)        \n",
    "        self.nruns = int(nruns)\n",
    "        self.run_time = float(run_time)\n",
    "        self.is_exclusive = bool(is_exclusive)\n",
    "    \n",
    "    def add_mask(self,obj):\n",
    "        self.masks.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "        \n",
    "    def add_job(self,obj):\n",
    "        self.jobs.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    @staticmethod            \n",
    "    def get(task_id,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(Task).get(int(task_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([Task])\n",
    "                        .where(Task.task_id==int(task_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs  \n",
    "        \n",
    "class Mask(BaseMixin,Base):\n",
    "    __tablename__= 'masks'\n",
    "    \n",
    "    mask_id = Column(Integer, Sequence('mask_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    \n",
    "    task_id = Column(Integer, ForeignKey('tasks.task_id'))\n",
    "    \n",
    "    source = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    value = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "    \n",
    "    def __init__(self,source='',name='',value=''):\n",
    "        self.source = str(source)\n",
    "        self.name   = str(name)\n",
    "        self.value  = str(value)\n",
    "                \n",
    "    @staticmethod            \n",
    "    def get(mask_id,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(Mask).get(int(mask_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([Mask])\n",
    "                        .where(Mask.mask_id==int(mask_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs \n",
    "\n",
    "    \n",
    "class Job(BaseMixin,Base):\n",
    "    __tablename__= 'jobs'\n",
    "    \n",
    "    job_id = Column(BigInteger, Sequence('job_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    \n",
    "    task_id = Column(Integer, ForeignKey('tasks.task_id'))\n",
    "    \n",
    "    config_id = Column(Integer, ForeignKey('configurations.config_id'))\n",
    "    \n",
    "    options = relationship('Options',\n",
    "                           backref=backref('jobs',\n",
    "                           uselist=True,passive_updates=False,\n",
    "                           cascade='delete,all'))\n",
    "\n",
    "    events = relationship('Event', secondary='job_event_link')\n",
    "    \n",
    "    nodes = relationship('Node', secondary='job_node_link')\n",
    "    \n",
    "    state = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    starttime = Column(DateTime, default=func.now())\n",
    "    endtime = Column(DateTime, default=func.now())\n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "    \n",
    "    \n",
    "    def __init__(self,state='new'):\n",
    "        self.state = str(state)\n",
    "    \n",
    "    def add_options(self,obj):\n",
    "        for opt in obj:\n",
    "            self.options.append(opt)\n",
    "        session.commit()\n",
    "        return\n",
    " \n",
    "    def add_event(self,obj):\n",
    "        self.events.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "        \n",
    "    def add_node(self,obj):\n",
    "        self.nodes.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    @staticmethod            \n",
    "    def get(job_id,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(Job).get(int(job_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([Job])\n",
    "                        .where(Job.job_id==int(job_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs \n",
    "    \n",
    "class Event(BaseMixin,Base):\n",
    "    __tablename__= 'events'\n",
    "    \n",
    "    event_id = Column(BigInteger, Sequence('event_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    \n",
    "    jobs = relationship('Job', secondary='job_event_link')\n",
    "    \n",
    "    jargs = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    value = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "\n",
    "    options = relationship('Options',\n",
    "                           backref=backref('events',\n",
    "                           uselist=True,passive_updates=False,\n",
    "                           cascade='delete,all'))\n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "    \n",
    "    def __init__(self,name='any',value='',jargs=''):\n",
    "        self.jargs  = str(jargs)\n",
    "        self.name   = str(name)\n",
    "        self.value  = str(value)\n",
    "        \n",
    "    def add_job(self,obj):\n",
    "        self.jobs.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "\n",
    "    @staticmethod            \n",
    "    def get(event_id,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(Event).get(int(event_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([Event])\n",
    "                        .where(Event.event_id==int(event_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs\n",
    "      \n",
    "    \n",
    "class Node(BaseMixin,Base):\n",
    "    __tablename__= 'nodes'\n",
    "    \n",
    "    node_id = Column(Integer, Sequence('node_id_seq'),\n",
    "                     primary_key=True, nullable=False)\n",
    "    name = Column(postgresql.VARCHAR(64),nullable=False)\n",
    "    \n",
    "    jobs = relationship('Job', secondary='job_node_link')\n",
    "    \n",
    "    int_ip = Column(postgresql.INET)\n",
    "    ext_ip = Column(postgresql.INET)\n",
    "    \n",
    "    timestamp = Column(DateTime, default=func.now())\n",
    "    \n",
    "    def __init__(self,name='any',int_ip='',ext_ip=''):\n",
    "        self.name = str(name)\n",
    "        self.int_ip = int_ip\n",
    "        self.ext_ip = ext_ip\n",
    "        \n",
    "    @classmethod\n",
    "    def add_job(self,obj):\n",
    "        self.jobs.append(obj)\n",
    "        session.commit()\n",
    "        return\n",
    "    \n",
    "    @staticmethod            \n",
    "    def get(node_id,how='sql'):\n",
    "        if how=='sql':\n",
    "            rs = session.query(Node).get(int(node_id))\n",
    "        elif how=='pd':\n",
    "            with engine.connect() as conn:\n",
    "                rs = pd.read_sql_query(select([Node])\n",
    "                        .where(Node.node_id==int(node_id))\n",
    "                        ,conn)\n",
    "                rs = rs.iloc[0]\n",
    "        return rs\n",
    "\n",
    "    \n",
    "class JobEventLink(Base):\n",
    "    __tablename__ = 'job_event_link'    \n",
    "    job_id = Column(Integer, ForeignKey('jobs.job_id'), primary_key=True)\n",
    "    event_id = Column(Integer, ForeignKey('events.event_id'), primary_key=True)\n",
    "\n",
    "    \n",
    "class JobNodeLink(Base):\n",
    "    __tablename__ = 'job_node_link'    \n",
    "    job_id = Column(Integer, ForeignKey('jobs.job_id'), primary_key=True)\n",
    "    node_id = Column(Integer, ForeignKey('nodes.node_id'), primary_key=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "Base.metadata.create_all(engine)\n",
    "session.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a new user\n",
    "newUser1 = User('ben').create()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fetch that user from the database\n",
    "myUser = User.get('ben')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Create a new pipeline and check the pipeline number\n",
    "# Can also specify software_root,data_root,pipe_root,config_root\n",
    "newPipe1 = Pipeline(name='test_pipe',\n",
    "                 description='Testing pipeline').create()\n",
    "newPipe1.pipeline_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Create another pipeline\n",
    "newPipe2 = Pipeline(name='test_pipe',\n",
    "                 description='Testing pipeline').create()\n",
    "newPipe2.pipeline_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Give both pipelines to the fetched user\n",
    "myUser.add_pipeline(newPipe1)\n",
    "myUser.add_pipeline(newPipe2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "1\n",
      "1\n"
     ]
    }
   ],
   "source": [
    "# Did it work?\n",
    "print(myUser.user_id)\n",
    "print(newPipe1.user_id)\n",
    "print(newPipe2.user_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Create a Target, check its ID\n",
    "newTarget = Target(name='test_target').create()\n",
    "newTarget.target_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Add it to newPipe1, check if it worked\n",
    "# In actual code, say \"create_dir=True\"\n",
    "newPipe1.add_target(newTarget,create_dir=False)\n",
    "newTarget.pipeline_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "target_id                                1\n",
       "name                           test_target\n",
       "pipeline_id                              1\n",
       "relativepath                  /test_target\n",
       "timestamp       2019-01-12 23:37:19.970756\n",
       "Name: 0, dtype: object"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Did the relativepath get added? \n",
    "Target.get(1,'pd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Create a configuration\n",
    "newConfig = Configuration(name='test_config').create()\n",
    "newConfig.config_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add it to newTarget\n",
    "newTarget.add_config(newConfig,create_dir=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "config_id                                   1\n",
       "name                              test_config\n",
       "target_id                                   1\n",
       "relativepath                     /test_target\n",
       "logpath          /test_target/log_test_config\n",
       "confpath        /test_target/conf_test_config\n",
       "rawpath          /test_target/raw_test_config\n",
       "procpath        /test_target/proc_test_config\n",
       "description                                  \n",
       "timestamp          2019-01-12 23:37:20.114414\n",
       "Name: 0, dtype: object"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Check the paths\n",
    "Configuration.get(1,'pd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create and add parameters to newConfig\n",
    "# Note that this \"create\" is different\n",
    "params={'a':0,'x':12,'note':'testing this'}\n",
    "newParams = Parameters().create(params)\n",
    "newConfig.add_params(newParams)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'a': '0', 'x': '12', 'note': 'testing this'}"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Check if it worked\n",
    "Parameters.getParams(newConfig.config_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create some data products, add them to newConfig\n",
    "myDP1 = DataProduct(filename='test_file1.fits',group='raw',\n",
    "                    filtername='H158').create()\n",
    "myDP2 = DataProduct(filename='test_file23.txt',group='conf',\n",
    "                    filtername='').create()\n",
    "myDP3 = DataProduct(filename='test_file34.cl',group='proc',\n",
    "                    filtername='').create()\n",
    "myDP4 = DataProduct(filename='test_file62.log',group='log',\n",
    "                    filtername='').create()\n",
    "newConfig.add_dp(myDP1)\n",
    "newConfig.add_dp(myDP2)\n",
    "newConfig.add_dp(myDP3)\n",
    "newConfig.add_dp(myDP4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dp_id                                      3\n",
       "config_id                                  1\n",
       "filename                      test_file34.cl\n",
       "relativepath                                \n",
       "suffix                                    cl\n",
       "data_type                                 cl\n",
       "subtype                                     \n",
       "group                                  other\n",
       "filtername                                  \n",
       "ra                                         0\n",
       "dec                                        0\n",
       "pointing_angle                             0\n",
       "timestamp         2019-01-12 23:37:20.446748\n",
       "Name: 0, dtype: object"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Check one of those\n",
    "DataProduct.get(3,'pd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a task for newPipe1\n",
    "newTask = Task('new_task').create()\n",
    "newPipe1.add_task(newTask)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create some masks, add them to newTask\n",
    "mask1 = Mask('*','some_mask','task_name').create()\n",
    "mask2 = Mask('*','another_mask','other_name').create()\n",
    "newTask.add_mask(mask1)\n",
    "newTask.add_mask(mask2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a job, add it to newTask and newConfig\n",
    "# Creating jobs usually won't take any parameters\n",
    "newJob = Job().create()\n",
    "newTask.add_job(newJob)\n",
    "newConfig.add_job(newJob)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "job_id                                1\n",
       "task_id                               1\n",
       "config_id                             1\n",
       "state                               new\n",
       "starttime    2019-01-12 23:37:20.635384\n",
       "endtime      2019-01-12 23:37:20.635384\n",
       "timestamp    2019-01-12 23:37:20.635384\n",
       "Name: 0, dtype: object"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Let's take a look at the job\n",
    "Job.get(newJob.job_id,'pd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create an event, add the job to it\n",
    "# e.g., an event is starting a job\n",
    "newEvent1 = Event('newJobFound',16).create()\n",
    "newEvent1.add_job(newJob)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create annother event, add it to the job\n",
    "# e.g., a job is triggering a event\n",
    "newEvent2 = Event('newEventProduced',100).create()\n",
    "newJob.add_event(newEvent2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create some options and add them to newTarget\n",
    "opts={'what':'this','how_much':11}\n",
    "newOpts = Options().create(opts)\n",
    "newTarget.add_options(newOpts)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'what'"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "newTarget.options[0].name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'this'"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "newTarget.options[0].value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "task_id                                  1\n",
       "name                              new_task\n",
       "pipeline_id                              1\n",
       "nruns                                    0\n",
       "runtime                               None\n",
       "is_exclusive                         False\n",
       "timestamp       2019-01-12 23:37:20.510768\n",
       "Name: 0, dtype: object"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Let's take a look at a task\n",
    "Task.get(1,'pd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Increment its nruns a few times\n",
    "myTask = Task.get(1)\n",
    "myTask.nruns += 1\n",
    "session.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "task_id                                  1\n",
       "name                              new_task\n",
       "pipeline_id                              1\n",
       "nruns                                    1\n",
       "runtime                               None\n",
       "is_exclusive                         False\n",
       "timestamp       2019-01-12 23:37:20.510768\n",
       "Name: 0, dtype: object"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Did it work?\n",
    "Task.get(1,'pd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}