backend/exhauster_analytics/analytics/services.py

import pandas
from rest_framework.exceptions import NotFound

from exhauster_analytics.analytics.models import (
    ExgausterSignal,
    Record,
    RecordApproximation,
    ExgausterRecordSignal, ExgausterRecordApproximationSignal,
)


def get_signal_values(signals, approximation=None, time_from=None, time_until=None):
    exhauster_signals = []
    signals_by_id = {}
    for signal in signals:
        try:
            if "." in signals:
                x, y = map(
                    int, signal[signal.find("[") + 1 : signal.find("]")].split(".")
                )
                type = "digital"
            else:
                x, y = map(
                    int, signal[signal.find("[") + 1 : signal.find("]")].split(":")
                )
                type = "analog"
            signal_obj = ExgausterSignal.objects.get(place_x=x, place_y=y, type=type)
            signals_by_id[signal_obj.id] = signal
        except (ExgausterSignal.DoesNotExist, ValueError):
            raise NotFound("signal not found")
        exhauster_signals.append(signal_obj)

    if not approximation or approximation == 1:
        if time_from:
            if time_until:
                records = Record.objects.filter(
                    timestamp__gte=time_from, timestamp__lte=time_until
                )
            else:
                records = Record.objects.filter(timestamp__gte=time_from)
        else:
            if time_until:
                records = Record.objects.filter(timestamp__lte=time_until)
            else:
                records = Record.objects.all()
        vals = ExgausterRecordSignal.objects.filter(
            record__in=records, signal__in=exhauster_signals
        ).values_list("signal", "value")
    else:
        if time_from:
            if time_until:
                records = RecordApproximation.objects.filter(
                    timestamp__gte=time_from,
                    timestamp__lte=time_until,
                    amount=approximation,
                )
            else:
                records = RecordApproximation.objects.filter(
                    timestamp__gte=time_from, amount=approximation
                )
        else:
            if time_until:
                records = RecordApproximation.objects.filter(
                    timestamp__lte=time_until, amount=approximation
                )
            else:
                records = RecordApproximation.objects.filter(amount=approximation)
        vals = ExgausterRecordApproximationSignal.objects.filter(
            record__in=records, signal__in=exhauster_signals
        ).values_list("signal_id", "value")
    d = {}
    for x, y in vals:
        d.setdefault(signals_by_id[x], []).append(y)
    return d


# function to parse signal mapping from xlsx file
def f(ex, num):
    df = pandas.read_excel(
        "Маппинг сигналов.xlsx", sheet_name=num, nrows=105, index_col=[4]
    )
    df = df.fillna(method="ffill")
    for index, row in df.iterrows():
        row = row.values.tolist()
        s = index
        if "." in s:
            x, y = map(int, s[s.find("[") + 1 : s.find("]")].split("."))
        else:
            x, y = map(int, s[s.find("[") + 1 : s.find("]")].split(":"))
        ExgausterSignal.objects.create(
            exgauster=ex,
            place_x=x,
            place_y=y,
            type=row[5],
            comment=row[4],
            active=bool(row[6]),
            item=row[0],
            characteristics=row[1],
            characteristics_description=row[2],
            item_name=row[3],
        )

    df = pandas.read_excel(
        "Маппинг сигналов.xlsx", sheet_name=num, skiprows=105, nrows=4, index_col=[4]
    )
    df = df.fillna(method="ffill")
    for index, row in df.iterrows():
        row = row.values.tolist()
        s = index
        if "." in s:
            x, y = map(int, s[s.find("[") + 1 : s.find("]")].split("."))
        else:
            x, y = map(int, s[s.find("[") + 1 : s.find("]")].split(":"))
        ExgausterSignal.objects.create(
            exgauster=ex,
            place_x=x,
            place_y=y,
            type=row[5],
            comment=row[4],
            active=bool(row[6]),
            item=row[0],
            characteristics=row[1],
            characteristics_description=row[2],
            item_name=row[3],
        )

    df = pandas.read_excel(
        "Маппинг сигналов.xlsx", sheet_name=num, skiprows=109, index_col=[4]
    )
    df = df.fillna(method="ffill")
    for index, row in df.iterrows():
        row = row.values.tolist()
        s = index
        if "." in s:
            x, y = map(int, s[s.find("[") + 1 : s.find("]")].split("."))
        else:
            x, y = map(int, s[s.find("[") + 1 : s.find("]")].split(":"))
        ExgausterSignal.objects.create(
            exgauster=ex,
            place_x=x,
            place_y=y,
            type=row[5],
            comment=row[4],
            active=bool(row[6]),
            item=row[0],
            item_name=row[3],
        )