Я написал веб-приложение, которое запрашивает данные из базы данных на основе некоторых входных данных (время начала / окончания, идентификатор машины и идентификатор параметра) и показывает их в виде боке:
Как вы можете видеть, пока он работает как задумано, но у меня есть некоторые планы по расширению этого приложения:
- Разрешить загрузку данных из разных пакетов (с разными временными метками начала / конца) в один и тот же график.
- Выполните некоторый статистический анализ различных партий, например, средние, стандартные отклонения, контрольные пределы и т. д.
- Получите обновления потоковой передачи параметров для разных машин и / или параметров.
Итак, я сейчас нахожусь в точке, где приложение начинает усложняться, и я хочу реорганизовать код в поддерживаемый и расширяемый формат. В настоящее время код написан процедурно, и я хотел бы перейти к MVC-подобной модели, чтобы отделить запрос данных от визуализаций боке и статистических вычислений, но я не уверен, как лучше подойти к этому.
Как я могу реорганизовать мой код лучше всего?
import logging
import pymssql, pandas
from dateutil import parser
from datetime import datetime, timedelta
from bokeh import layouts, models, plotting, settings
from bokeh.models import widgets
SETTINGS = {
'server': '',
'user': '',
'password': '',
'database': ''
}
def get_timestamps(datetimes):
""" DB timestamps are in milliseconds """
return [int(dt.timestamp()*1000) for dt in datetimes]
def get_db_names(timestamps):
logging.debug('Started getting DB names ...')
query = """
SELECT
[DBName]
FROM [tblDBNames]
WHERE {timestamp_ranges}
""".format(
timestamp_ranges = ' OR '.join([f'({timestamp} BETWEEN [LStart] AND [LStop])' for timestamp in timestamps])
)
logging.debug(query)
db_names = []
with pymssql.connect(**SETTINGS) as conn:
with conn.cursor(as_dict=True) as cursor:
cursor.execute(query)
for row in cursor:
db_names.append(row['DBName'])
#logging.debug(db_names)
logging.debug('Finished getting DB names')
return list(set(db_names))
def get_machines():
logging.debug('Started getting machines ...')
query = """
SELECT
CONVERT(VARCHAR(2),[ID]) AS [ID],
[Name]
FROM [tblMaschinen]
WHERE NOT [Name] = 'TestLine4'
ORDER BY [Name]
"""
logging.debug(query)
with pymssql.connect(**SETTINGS) as conn:
with conn.cursor(as_dict=False) as cursor:
cursor.execute(query)
data = cursor.fetchall()
#logging.debug(data)
logging.debug('Finished getting machines')
return data
def get_parameters(machine_id, parameters):
logging.debug('Started getting process parameteres ...')
query = """
SELECT
CONVERT(VARCHAR(4), TrendConfig.ID) AS [ID],
TrendConfig_Text.description AS [Description]
FROM [TrendConfig]
INNER JOIN TrendConfig_Text
ON TrendConfig.ID = TrendConfig_Text.ID
WHERE (TrendConfig_Text.languageText_KEY = 'nl')
AND TrendConfig.MaschinenID = {machine_id}
AND TrendConfig_Text.description IN ('{parameters}')
ORDER BY TrendConfig_Text.description
""".format(
machine_id = machine_id,
parameters = "', '".join(parameters)
)
logging.debug(query)
with pymssql.connect(**SETTINGS) as conn:
with conn.cursor(as_dict=False) as cursor:
cursor.execute(query)
data = cursor.fetchall()
#logging.debug(data)
logging.debug('Finished getting process parameters')
return data
def get_process_data(query):
logging.debug('Started getting process data ...')
with pymssql.connect(**SETTINGS) as conn:
return pandas.read_sql(query, conn, parse_dates={'LTimestamp': 'ms'}, index_col='LTimestamp')
logging.debug('Finished getting process data')
batches = widgets.Slider(start=1, end=10, value=1, step=1, title="Batches")
# TODO: make custom datetime widget - https://bokeh.pydata.org/en/latest/docs/user_guide/extensions_gallery/widget.html
now, min_date = datetime.now(), datetime.fromtimestamp(1316995200)
date_start = widgets.DatePicker(title="Start date:", value=str(now.date()), min_date=str(min_date), max_date=str(now.date()))
time_start = widgets.TextInput(title="Start time:", value=str((now-timedelta(hours=1)).replace(microsecond=0).time()))
start_row = layouts.Row(children=[date_start, time_start], width = 300)
date_end = widgets.DatePicker(title="End date:", value=str(now.date()), min_date=str(min_date), max_date=str(now.date()))
time_end = widgets.TextInput(title="End time:", value=str(now.replace(microsecond=0).time()))
end_row = layouts.Row(children=[date_end, time_end], width = 300)
datetimes = layouts.Column(children=[start_row, end_row])
## Machine list
machines = get_machines()
def select_machine_cb(attr, old, new):
logging.debug(f'Changed machine ID: old={old}, new={new}')
parameters = get_parameters(select_machine.value, default_params)
select_parameters.options = parameters
select_parameters.value = [parameters[0][0]]
select_machine = widgets.Select(
options = machines,
value = machines[0][0],
title = 'Machine:'
)
select_machine.on_change('value', select_machine_cb)
## Parameters list
default_params = [
'Debiet acuteel',
'Extruder energie',
'Extruder kWh/kg',
'Gewicht bunker',
'RPM Extruder acuteel',
'Temperatuur Kop'
]
parameters = get_parameters(select_machine.value, default_params)
select_parameters = widgets.MultiSelect(
options = parameters,
value = [parameters[0][0]],
title = 'Parameter:'
)
def btn_update_cb(arg):
logging.debug('btn_update clicked')
datetime_start = parser.parse(f'{date_start.value} {time_start.value}')
datetime_end = parser.parse(f'{date_end.value} {time_end.value}')
datetimes = [datetime_start, datetime_end]
timestamps = get_timestamps(datetimes)
db_names = get_db_names(timestamps)
machine_id = select_machine.value
parameter_ids = select_parameters.value
query = """
SELECT
[LTimestamp],
[TrendConfigID],
[Text],
[Value]
FROM ({derived_table}) [Trend]
LEFT JOIN [TrendConfig] AS [TrendConfig]
ON [Trend].[TrendConfigID] = [TrendConfig].[ID]
WHERE [LTimestamp] BETWEEN {timestamp_range}
AND [Trend].[TrendConfigID] IN ({id_range})
""".format(
derived_table = ' UNION ALL '.join([f'SELECT * FROM [{db_name}].[dbo].[Trend_{machine_id}]' for db_name in db_names]),
timestamp_range = ' AND '.join(map(str,timestamps)),
id_range = ' ,'.join(parameter_ids)
)
logging.debug(query)
df = get_process_data(query)
ds = models.ColumnDataSource(df)
plot.renderers = [] # clear plot
#view = models.CDSView(source=ds, filters=[models.GroupFilter(column_name='TrendConfigID', group='')])
#plot = plotting.figure(plot_width=600, plot_height=300, x_axis_type='datetime')
plot.line(x='LTimestamp', y='Value', source=ds, name='line')
btn_update = widgets.Button(
label="Update",
button_type="primary",
width = 150
)
btn_update.on_click(btn_update_cb)
btn_row = layouts.Row(children=[btn_update])
column = layouts.Column(children=[batches, datetimes, select_machine, select_parameters, btn_row], width = 300)
plot = plotting.figure(plot_width=600, plot_height=300, x_axis_type='datetime')
row = layouts.Row(children=[column, layouts.Spacer(width=20), plot])
tab1 = models.Panel(child=row, title="Viewer")
tab2 = models.Panel(child=layouts.Spacer(), title="Settings")
tabs = models.Tabs(tabs=[tab1, tab2])
plotting.curdoc().add_root(tabs)