All/Python/local/ear-tag/action.py

import math
import time
import requests


class Fetcher:
    def __init__(self, farmId: "str" = "330110004"):
        self._farm: "str" = farmId
        self._sess: "requests.Session" = requests.session()

    def _login(self) -> "None":
        url = "http://119.3.44.183:8016/admin/my/loginMultilevel"
        payload = {"accountName": "superadmin", "password": "hm123456", "farmId": self._farm}
        resp = self._sess.post(url=url, json=payload)
        assert resp.status_code == 200, f"login failed: {resp.status_code}"
        self._sess.headers.setdefault("Accesstoken", resp.json()["data"]["token"])

    @staticmethod
    def _time(ts: "str") -> "int":
        return int(time.mktime(time.strptime(ts, "%Y-%m-%d %H:%M:%S")))

    @staticmethod
    def _num(data):
        return data if data is not None else 0

    def _clean(self, data: "list[dict]") -> "list[tuple]":
        # envTemp, earTemp1, earTemp2, act, timestamp
        return [(
            self._num(itr["envTemp1"]), self._num(itr["earTemp1"]),
            self._num(itr["earTemp2"]), self._num(itr["act1"]), self._time(itr["addTime"])
        ) for itr in data]

    def fetch(self, tagId: "str", start: "str", end: "str") -> "list[tuple]":
        self._login()
        url = "http://119.3.44.183:8016/manage2/eartagData/getEnvByTime"
        payload = {"earmark": tagId, "startDate": start, "endDate": end, "farmId": self._farm}
        resp = self._sess.post(url=url, json=payload)
        assert resp.status_code == 200, f"> ERROR: fetch <{tagId}> failed!"
        return self._clean(resp.json()["data"])


class Proxy:
    def __init__(self):
        self._data = None

    def get(self):
        return self._data

    def set(self, data):
        self._data = data


# 分类多
class Result:
    __m = {
        1: (1, "温度稳定于体温，运动量小", "睡眠状态"),  # 1
        2: (2, "温度稳定于体温，运动量大", "异常状态"),  # 4
        3: (3, "温度稳定于室温，运动量小", "侧躺状态"),  # 2
        4: (4, "温度稳定于室温，运动量大", "异常状态"),  # 4
        5: (5, "温度稳定，运动量小", "异常状态"),  # 4
        6: (6, "温度稳定，运动量大", "异常状态"),  # 4
        7: (7, "温度在体温附近不稳定，运动量小", "异常状态"),  # 4
        8: (8, "温度在体温附近不稳定，运动量大", "活跃状态"),  # 3
        9: (9, "温度在室温附近不稳定，运动量小", "异常状态"),  # 4
        10: (10, "温度在室温附近不稳定，运动量大", "活跃状态"),  # 3
        11: (11, "温度不稳定，运动量小", "异常状态"),  # 4
        12: (12, "温度不稳定，运动量大", "活跃状态")  # 3
    }

    @staticmethod
    def kind(t: "int") -> "int":
        if t == 1:
            return 1
        if t == 3:
            return 2
        if t in [8, 10, 12]:
            return 3
        return 4

    @staticmethod
    def _time(sec: "int") -> "str":
        return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(sec))

    def _info(self, _type: "int") -> "tuple":
        return self.__m.get(_type, (-1, "unrecognized type", "unknown"))

    def __init__(self, _type: "int", start: "int", end: "int"):
        self.start, self.end = start, end
        self.id, self.desc, self.type = self._info(_type)
        self.sTime, self.eTime = self._time(start), self._time(end)

    def __repr__(self):
        return (
            f"<id:{self.id}, desc:\"{self.desc}\", "
            f"type:\"{self.type}\", start:\"{self.sTime}\", end:\"{self.eTime}\">"
        )


class Action:
    def __init__(
            self,
            body: "float" = 35.5,
            actHold: "tuple[int, int]" = (8, 10),
            tempHold: "tuple[float, float]" = (5.5, 6.0)
    ):
        self._bodyTemp: "float" = body
        self._actMin: "int" = actHold[0]
        self._actMax: "int" = actHold[1]
        self._tempMin: "float" = tempHold[0]
        self._tempMax: "float" = tempHold[1]

    @staticmethod
    def _handle(arr: "list[Result]", data: "tuple", curr: "int", state: "Proxy", start: "Proxy"):
        if curr != state.get():
            if state.get() is not None:
                arr.append(Result(state.get(), start.get(), data[-1]))
            state.set(curr)
            start.set(data[-1])

    def analyze(self, data: "list[tuple]") -> "list[Result]":
        result, state, start = [], Proxy(), Proxy()

        for itr in data:
            envT, earT1, earT2, act, secs = itr
            if abs(earT1 - earT2) < self._tempMin:  # 温度稳定
                if abs(earT1 - self._bodyTemp) < self._tempMin:  # 在体温附近稳定
                    if act < self._actMax:
                        self._handle(result, itr, 1, state, start)
                    else:
                        self._handle(result, itr, 2, state, start)
                elif abs(earT1 - envT) < self._tempMin:  # 在环境温度附近稳定
                    if act < self._actMax:
                        self._handle(result, itr, 3, state, start)
                    else:
                        self._handle(result, itr, 4, state, start)
                else:  # 其余稳定
                    if act < self._actMax:
                        self._handle(result, itr, 5, state, start)
                    else:
                        self._handle(result, itr, 6, state, start)
            else:
                if abs(earT1 - self._bodyTemp) < self._tempMax:  # 在体温附近不稳定
                    if act < self._actMin:
                        self._handle(result, itr, 7, state, start)
                    else:
                        self._handle(result, itr, 8, state, start)
                elif abs(earT1 - envT) < self._tempMax:  # 在环境温度附近不稳定
                    if act < self._actMin:
                        self._handle(result, itr, 9, state, start)
                    else:
                        self._handle(result, itr, 10, state, start)
                else:  # 其余不稳定
                    if act < self._actMin:
                        self._handle(result, itr, 11, state, start)
                    else:
                        self._handle(result, itr, 12, state, start)

        return result


# 聚合为4类
class Join:
    __m = {1: "睡眠状态", 2: "侧躺状态", 3: "活跃状态", 4: "其它状态"}

    def __init__(self, data: "list[Result]"):
        self.data = data

    @staticmethod
    def _type(_id: "int") -> "str":
        return Join.__m.get(_id, "unknown")

    @staticmethod
    def _make(old: "Result") -> "Result":
        cid = Result.kind(old.id)
        new = Result(1, old.start, old.end)
        new.id, new.desc, new.type = cid, "-", Join._type(cid)
        return new

    def run(self) -> "list[Result]":
        res = []
        cur = self._make(self.data[0])
        for i in range(1, len(self.data)):
            cid = Result.kind(self.data[i].id)
            if cid != cur.id:
                cur.end, cur.eTime = self.data[i].start, self.data[i].sTime
                res.append(cur)
                cur = self._make(self.data[i])
        cur.end, cur.eTime = self.data[-1].end, self.data[-1].eTime
        res.append(cur)
        return res


class RespItem:
    def __init__(self, _id: "int", _type: "str", _time: "str", width: "int"):
        self.id, self.type, self.time, self.width = _id, _type, _time, width

    def __repr__(self):
        return f"<id: {self.id}, type: \"{self.type}\", time: \"{self.time}\", width: {self.width}>"


class Calc:
    def __init__(self, data: "list[Result]"):
        self.sum = {}
        self.data = data
        self.total: "int" = data[-1].end - data[0].start

    def run(self):
        result = []
        for itr in self.data:
            key = itr.sTime[:10]
            if key not in self.sum.keys():
                self.sum[key] = {1: 0, 2: 0, 3: 0, 4: 0}
            self.sum[key][itr.id] += itr.end - itr.start
            result.append(RespItem(itr.id, itr.type, itr.sTime, math.ceil((itr.end - itr.start) * 100 / self.total)))
        return result


def show(data: "list"):
    print("[")
    [print(f"\t{itr},") for itr in data]
    print("]")


def main():
    fetcher = Fetcher()
    data = fetcher.fetch("133202311130004", "2023-12-1 00:00:00", "2023-12-10 00:00:00")
    print(data)
    action = Action()
    state = action.analyze(data)
    show(state)
    join = Join(state)
    new = join.run()
    show(new)
    calc = Calc(new)
    resp = calc.run()
    show(resp)
    print(calc.sum)


if __name__ == "__main__":
    main()