daq-3i/db_model.py at master · smtwkla/daq-3i · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
from sqlalchemy import Column, ForeignKey, Integer, String, DateTime, Numeric, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine

Base = declarative_base()


class Channels(Base):
    __tablename__ = 'channels'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    bus_id = Column(Integer, ForeignKey('buses.id'), nullable=False, )
    bus = relationship("Buses")
    device_id = Column(Integer, nullable=False)
    address = Column(Integer, nullable=False)
    timing = Column(Integer, nullable=False)
    conversion_id = Column(Integer, ForeignKey('conversions.id'), nullable=False)
    conversion = relationship("Conversions")
    func_code = Column(Integer, nullable=False)
    format_code = Column(Integer, nullable=False)
    enabled = Column(Boolean, nullable=False)
    eng_unit = Column(String(15))
    history_len = Column(Integer, nullable=False, default=1)
    long_name = Column(String(50))

    def __repr__(self):
        return self.name


class Buses(Base):
    __tablename__ = 'buses'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    protocol = Column(Integer, nullable=False)
    address = Column(String(50))
    port = Column(Integer)
    timeout = Column(Integer, nullable=False)
    enabled = Column(Boolean, nullable=False)


class Conversions(Base):
    __tablename__ = 'conversions'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    expr = Column(String(250))


class Channel_Data(Base):
    __tablename__="channel_data"
    id = Column(Integer, primary_key=True)
    channel_id = Column(Integer, ForeignKey('channels.id'), nullable=False)
    ts = Column(DateTime, nullable=False)
    value = Column(Numeric(25, 6))


class Daq_Status(Base):
    __tablename__= "daq_status"
    id = Column(Integer, primary_key=True)
    parameter = Column(String(25), nullable=False, unique=True)
    status = Column(Integer)
    ts = Column(DateTime)


def getdburl(c):
    con = f"{c['db']['dbdialect']}://{c['db']['user']}:{c['db']['pass']}@{c['db']['host']}:{c['db']['port']}/{c['db']['database']}"
    return con

def create_tables(engine):

    import configparser
    config = configparser.ConfigParser()
    config.read('config.ini')

    con = getdburl(config)
    # Create an engine
    engine = create_engine(con)

    # Create all tables in the engine.
    Base.metadata.create_all(engine)