# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python (Prod Env)
#     language: python
#     name: prod
# ---

# Начнем с подключения неободимых библиотек, применение каждой из них будет рассмотрено отдельно позднее.

# +
import requests
import psycopg2
import petl as etl
import numpy as np

from datetime import datetime as dt
# -

# # Библиотека PETL

# ### Загрузка данных
# `petl` поддерживает различные источники данных, мы рассмотрим следующие:
# - Загрузка из xlsx-файла  
# - Использование открытых источников через API
# - Работа с базой данных

# ### Данные из xlsx-файла

# Рассмотрим работу с `petl` на наборе результатов летних олимпиад по странам. Нам понадобится файл `datasets/summer_olympics.xlsx`, посмотрим на первые строки, пока не сохраняя таблицу в переменную.

etl.fromxlsx('datasets/summer_olympics.xlsx')

# Видим, что данные загрузились без ошибок, однако заголовки столбцов не определились, потому что в начале файла есть лишняя пустая строка. Исправим это, используя функцию `skip` и поместим результат в переменную `olympics`

olympics = etl.fromxlsx('datasets/summer_olympics.xlsx').skip(1)

olympics

# Теперь заголовки у столбцов корректные, однако не достаточно информативны, исправим это, задав заголовки в ручную.

olympics2 = olympics.setheader(['country','games','gold','silver','bronze'])

olympics2

# Мы начали выстраивать цепочку преобразования таблицы, это удобно, так как можно просмотреть результат работы на каждом этапе. Иногда, наоборот, удобнее объединить цепочку сразу в одной команде. Например, совместим изменение заголовка с сортировкой по количеству золотых медалей.

olympics2 = olympics.setheader(['country','games','gold','silver','bronze']).sort('gold', reverse=True)

olympics2

# Теперь мы можем посчитать общее количество медалей и сохранить его в новом столбце, используя функцию `addfield`.
#
# Мы также применим мощный инструмент Python - **Анонимные функции**. **Анонимная функция** (функция без имени) - это запись вида `lambda x: <функция от x>`. Читается как: "То, что было подано на вход этого выражения, будет положено в `x`, а результатом исполнения будет `<функция от x>`. В PETL это часто применяется, чтобы выполнить быстрое преобразования значения какого-либо из полей. Например, если нужно все значения таблицы `table` в поле `field` умножить на два, это можно написать как `table.convert('field', lambda x : x * 2)`. В примере ниже функция применяется не к отдельным значениям, а к строке целиком.

olympics2.addfield('total', lambda row : row['gold'] + row['silver'] + row['bronze'])

# Вместо того, чтобы получить суммы, мы просто склеили значения. Чтобы такого не происходило, будем преобразовывать формат данных в целочисленные. Выясним, какая страна смогла набрать наибольшее число медалей, отсортировав сразу таблицу по новому столбцу по убыванию, с помощью функции `sort`. Также используем символ `\`, чтобы разбить команду на несколько строк для улучшения читаемости.

olympics3 = olympics2\
    .addfield('total', lambda x: int(x['gold']) + int(x['silver']) + int(x['bronze']))\
    .sort('total', reverse=True)

olympics3

# Видим, что в таблице есть сумма по всем странам, что нас не интересует в данной задаче. Можем выбрать из таблицы все строки, кроме строки со значением `country == Totals`. Воспользуемся функцией `select`.  
#
# Кроме того, дополнительно рассчитаем новый показатель - результативность страны, определив её как среднее число медалей за игру.

olympics4 = olympics3\
    .select(lambda x: x.country != 'Totals')\
    .addfield('effectiveness', lambda x: round(x['total'] / float(x['games']), 2))

olympics4

# Сохраним полученные результаты в новый xlsx-файл.

olympics4.toxlsx('olympics.xlsx')

# Готово! Теперь обработанный файл можно скачать или загрузить в BI-систему

# ### Данные из открытого источника рынка акций

# Рассмотрим немного более продвинутый пример - получение данных из веб-сервиса по API. Это также делается очень просто с использованием библиотеки `requests`

response = requests.get('https://www.quandl.com/api/v3/datasets/WIKI/AAPL.json?start_date=2017-05-01&end_date=2017-07-01')

# Посмотрим, что мы получили в ответ. Мы увидим данные в формате JSON, которые нужно будет промотать до конца

stock_prices_json = response.json()
stock_prices_json

# Видим, что в полученном JSON сама таблица с данными лежит в разделе `dataset`. Посмотрим, какие в ней есть поля.

stock_prices_json['dataset'].keys()

# Нас интересуют два поля ответа: `column_names`, который мы будем использовать в качестве заголовков таблицы, и `data`, содержащий все необходимые данные построчно. Для преобразования данных из объекта `dict` в таблицу `petl` сделаем следующее:  
# - Транспонируем содержимое `data`, чтобы превратить строки в столбцы  
# - Используем `column_names` в качестве значения параметра `header` функции `fromcolumns`

stock_prices = etl.fromcolumns(stock_prices_json['dataset']['data']).skip(1)\
    .transpose()\
    .setheader(stock_prices_json['dataset']['column_names'])

stock_prices

# Уберём часть столбцов, все, содержащие `'Adj'`, переведём все значения в числа (где это возможно), вычислим разницу курса на определённую дату. 
#
# В этом примере мы используем **List comprehension**, инструмент Python, который позволяет делать довольно сложные преобразования в наглядном функциональном стиле и без циклов.
#
# **List comprehension** - это запись вида `(<функция от x> for x in <список> if <условие от x)`, которая читается как: "Возьми все элементы из `<список>`, отбери те их них, для которых истинно `<условие от x>`, выполни над каждым `<функция от x>` и верни результаты в виде списка.
# Например, есть массив чисел `arr` и нужно отобрать из него четные числа и разделить их на 4. Это можно записать как `(x/4 for x in arr if x % 2 == 0)`

# +
stock_prices2 = stock_prices\
    .cutout(*(x for x in stock_prices.fieldnames() if 'Adj' in x))\
    .convertnumbers()\
    .addfield('Difference', lambda row: round(row.Close - row.Open, 2))

stock_prices2
# -

# Сохраним полученную табличку в csv-файл.

stock_prices2.tocsv('stock.csv')

# Через несколько секунд созданный файл появится в панели файлов слева, и вы сможете просмотреть или скачать его.

# ### Данные из БД (PostgreSQL)

# В состав ViXtract входит предустановленная СУБД PostgreSQL, её удобно использовать как промежуточное хранилище данных, из которого их уже забирает BI-система. Похожие подходы могут быть использованы и с любой другой СУБД.
#
# Рассмотрим следующий пример.  
# Доступны данные о состояниях различных типов транспортных средств. В базе есть 2 таблицы:
# - `status_ts` содержит информацию о состояниях различных ТС  
# - `ts_types` содержит наименования типов ТС  
#
# Необходимо подготовить таблицу, содержащую валидные данные по бульдозерам:
# - В данных не должно быть пропусков  
# - Время указано в формате datetime  
# - Кроме данных по бульдозерам других нет  
# - Все состояния, кроме отсутствия данных  
# - Для каждого состояния рассчитана продолжительность

# +
connection = psycopg2.connect("dbname=demo user=demo password=<PASSWORD>", host='localhost')

statuses = etl.fromdb(connection, 'SELECT * FROM status_ts')
ts_types = etl.fromdb(connection, 'SELECT * FROM ts_types')

# Вспомогательные функции
# Определяем фильтр для исключения строк с пустыми значениями
row_without_nones = lambda x: all(x[field] != '' for field in statuses.fieldnames())
# Перевод отметки времени в формат datetime
to_datetime = lambda x: dt.fromtimestamp(int(x))
# -

# Чтобы исключить строки с пропусками, используем функцию `select` и определенный выше фильтр `row_without_nones`

statuses.select(row_without_nones)

# Переведём столбцы со временем в требуемый формат. Для этого необходимо воспользоваться функцией `convert`.
#
# Сразу можем добавить расчёт продолжительности функцией `addfield`.  

statuses.\
    convert('Начало', to_datetime).\
    convert('Окончание', to_datetime).\
    addfield('Продолжительность', lambda x: x['Окончание'] - x['Начало'])

# Объединим обе таблицы и выберем данные только по бульдозерам, сразу уберём строки с состоянием "Отсутствие данных".

statuses.\
    join(ts_types, lkey='id ТС', rkey='id').\
    select(lambda x: 'Бульдозер' in x['Тип ТС'] and x['Состояние'] != 'Отсутствие данных')

# Все перечисленные операции можно произвести за раз, сформируем цепочку функций. 
#
# Заметим, что столбец `id ТС` уже не требуется, его можно убрать функцией `cutout`.
#
# В дополнение ко всему отсортируем таблицу по времени начала состояний, применив `sort`.

result = statuses.\
    join(ts_types, lkey='id ТС', rkey='id').\
    select(lambda x: 'Бульдозер' in x['Тип ТС'] and x['Состояние'] != 'Отсутствие данных').\
    select(row_without_nones).\
    convert('Начало', to_datetime).\
    convert('Окончание', to_datetime).\
    addfield('Продолжительность', lambda x: x['Окончание'] - x['Начало']).\
    convert('Начало', str).convert('Окончание', str).convert('Продолжительность', str).\
    cutout('id ТС').\
    sort('Начало')

result

# +
# Импортируем библиотеку, позволяющую создавать таблицы в БД
import sqlalchemy as db

# Подготовим подключение
_user = 'demo'
_pass = '<PASSWORD>'
_host = 'localhost'
_port = 5432
target_db = db.create_engine(f"postgres://{_user}:{_pass}@{_host}:{_port}/etl")

# Пробуем пересоздать таблицу (удалить и создать заново). Если таблицы нет - просто создаем новую.
try:
    result.todb(target_db, 'status_cleaned', create=True, drop=True, sample=0)
except:
    result.todb(target_db, 'status_cleaned', create=True, sample=0)
# -

# Проверим, что таблица создалась. Обратите внимание, что схема таблицы (типы полей, их названия и так далее) была создана полностью автоматически.

etl.fromdb(connection, 'SELECT * FROM status_cleaned')

# Поздравляем, вы завершили вводную часть по PETL и готовы решать свои задачи!
#
# Теперь вы можете перейти к рассмотрению примеров, более детальному изучению функций Python или настройке планировщика. 
#
# Подробнее на www.vixtract.ru

StarcoderdataJupyterScriptsDedupFiltered