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"" ) 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"" 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()