Coverage for /usr/local/lib/python3.11/site-packages/twinpad_backend/models.py: 89%

1from functools import cached_property 

2import os 

3import io 

4import time 

5import csv 

6from typing import Self, ClassVar, Any, Optional 

7import datetime 

8import math 

9import bisect 

10from enum import Enum 

11import logging 

12from typing import Literal, Dict 

13 

14import zipfile 

15import ping3 

16import pytz 

17from bson.objectid import ObjectId 

18from pymongo import ASCENDING, ReturnDocument 

19from pydantic import BaseModel, computed_field, Field, create_model 

20import numpy as npy 

21 

22# from scipy import signal as signal_scipy 

23 

24from twinpad_backend.db import ( 

25 get_collection, 

26 get_signal_collection, 

27 systems_database, 

28 signals_database, 

29 devices_states_database, 

30) 

31from twinpad_backend.responses import ListResponse 

32from twinpad_backend.messages import send_mode_change, send_signal_value 

33 

34TYPES = ({"int": int, "float": float, "str": str, "bool": bool, "epoch": float},) 

35 

36 

37RABBITMQ_HOST = os.environ.get("RABBITMQ_HOST", "cloud-broker") 

38MONGO_HOST = os.environ.get("MONGO_HOST", "time-series-db") 

39SIGNAL_STORAGE_HOST = os.environ.get("SIGNAL_STORAGE_HOST", "signal-storage") 

40HEARTBEAT_STORAGE_HOST = os.environ.get("HEARTBEAT_STORAGE_HOST", "heartbeat-storage") 

41DATA_ANALYZER_HOST = os.environ.get("DATA_ANALYZER_HOST", "data-analyzer") 

42 

43DEVICE_TIMEOUT = 5.0 

44NUMBER_SAMPLES_DATABASE_UPDATE = 120 

45 

46logger = logging.getLogger("uvicorn.error") 

47 

48number_samples_collection = get_collection(systems_database, "number_samples", create=True, time_series=True) 

49 

50 

51def create_update_model(model): 

52 fields = {} 

53 

54 for field_name, field_annotation in model.model_fields.items(): 

55 if field_name != "id": 

56 fields[field_name] = (Optional[field_annotation.annotation], None) 

57 

58 query_name = model.__name__ + "Update" 

59 return create_model(query_name, **fields) 

60 

61 

62def get_utc_date_from_timestamp(timestamp: float): 

63 return ( 

64 datetime.datetime.fromtimestamp(timestamp).replace(tzinfo=pytz.UTC).strftime("%Y-%m-%d %-H:%M:%S.%f")[:-3] 

65 + " UTC" 

66 ) 

67 

68 

69# Models 

70class TwinPadModel(BaseModel): 

71 @classmethod 

72 def dict_to_object(cls, dict_): 

73 return cls.model_validate(dict_) 

74 

75 def to_dict(self, exclude=None): 

76 dict_ = self.model_dump(exclude=exclude) 

77 return dict_ 

78 

79 

80class GenericMongo(TwinPadModel): 

81 id: str | None = None 

82 

83 @classmethod 

84 def collection(self): 

85 return get_collection(systems_database, self.collection_name, create=True) 

86 

87 @classmethod 

88 def response_from_query(cls, query) -> ListResponse[Self]: 

89 req_filter = query.mongodb_filter() 

90 items = [] 

91 if ":" in query.sort_by: 

92 sort_field, sort_order = query.sort_by.split(":") 

93 sort_order = int(sort_order) 

94 else: 

95 sort_field = query.sort_by 

96 sort_order = 1 

97 collection = get_collection(systems_database, cls.collection_name, create=True) 

98 total = collection.count_documents(req_filter) 

99 cursor = collection.find(req_filter).sort(sort_field, sort_order).skip(query.offset) 

100 if (query.limit is not None) and (query.limit != 0): 

101 cursor = cursor.limit(query.limit) 

102 for item_dict in cursor: 

103 items.append(cls.mongo_dict_to_object(item_dict)) 

104 return ListResponse(items=items, limit=query.limit, offset=query.offset, sort_by=query.sort_by, total=total) 

105 

106 @classmethod 

107 def get_from_id(cls, item_id) -> Self | None: 

108 # collection = get_collection(systems_database, cls.collection_name, create=True) 

109 

110 dict_ = cls.collection().find_one({"_id": ObjectId(item_id)}) 

111 if dict_ is None: 

112 return None 

113 return cls.mongo_dict_to_object(dict_) 

114 

115 @classmethod 

116 def mongo_dict_to_object(cls, mongo_dict): 

117 mongo_dict["id"] = str(mongo_dict["_id"]) 

118 del mongo_dict["_id"] 

119 return cls.dict_to_object(mongo_dict) 

120 

121 @classmethod 

122 def get_by_attribute(cls, attribute_name: str, attribute_value): 

123 """Returns all items that match the attribute with value.""" 

124 items = cls.collection().find({attribute_name: attribute_value}) 

125 if items is None: 

126 return None 

127 return [cls.mongo_dict_to_object(d) for d in items] 

128 

129 @classmethod 

130 def get_one_by_attribute(cls, attribute_name: str, attribute_value): 

131 item = cls.collection().find_one({attribute_name: attribute_value}) 

132 if item is None: 

133 return None 

134 return cls.mongo_dict_to_object(item) 

135 

136 @classmethod 

137 def get_all(cls, sort_by="_id") -> list[Self]: 

138 items = [] 

139 for dict_ in cls.collection().find({}).sort(sort_by, ASCENDING): 

140 items.append(cls.mongo_dict_to_object(dict_)) 

141 return items 

142 

143 def insert(self): 

144 insert_result = self.collection().insert_one(self.to_dict(exclude={id})) 

145 self.id = str(insert_result.inserted_id) 

146 return self.id 

147 

148 def update(self, update_dict): 

149 for key, value in update_dict.items(): 

150 setattr(self, key, value) 

151 self.collection().find_one_and_update( 

152 {"_id": ObjectId(self.id)}, 

153 {"$set": update_dict}, 

154 return_document=ReturnDocument.AFTER, 

155 ) 

156 

157 return self 

158 

159 def delete(self): 

160 result = self.collection().delete_one({"_id": ObjectId(self.id)}) 

161 return result.deleted_count > 0 

162 

163 

164class Mode(TwinPadModel): 

165 mode_id: int 

166 name: str 

167 frequency_multiplier: float 

168 min_frequency: float 

169 

170 

171class DeviceUpdate(TwinPadModel): 

172 mode_id: int 

173 

174 

175class Device(GenericMongo): 

176 collection_name: ClassVar[str] = "devices" 

177 

178 device_id: str 

179 name: str 

180 description: str = "" 

181 modes: list[Mode] 

182 current_mode_id: int | None = None 

183 last_ping: float | None = None 

184 petri_network: Any 

185 pid: Any 

186 load: float | None = None 

187 tokens: list[int] = Field(default_factory=list) 

188 

189 def update(self, update_device): 

190 send_mode_change(self.device_id, update_device.mode_id) 

191 

192 @computed_field 

193 @property 

194 def status(self) -> str: 

195 now = time.time() 

196 if self.last_ping is None: 

197 return "down" 

198 if (now - self.last_ping) > DEVICE_TIMEOUT: 

199 return "down" 

200 return "up" 

201 

202 

203class DeviceSetup(GenericMongo): 

204 collection_name: ClassVar[str] = "device_setups" 

205 

206 device_ids: list[str] 

207 active: bool = False 

208 variable_mapping: Dict[str, str] 

209 

210 

211DeviceSetupUpdate = create_update_model(DeviceSetup) 

212 

213 

214class DeviceState(GenericMongo): 

215 collection_name: ClassVar[str] = "devices_states" 

216 

217 timestamp: float 

218 mode_id: int | None = None 

219 load: float | None = None 

220 tokens: list[int] = Field(default_factory=list) 

221 

222 @classmethod 

223 def get_from_id_and_query(cls, device_id: str, query) -> ListResponse[Self]: 

224 req_filter = query.mongodb_filter() 

225 items = [] 

226 if ":" in query.sort_by: 

227 sort_field, sort_order = query.sort_by.split(":") 

228 sort_order = int(sort_order) 

229 else: 

230 sort_field = query.sort_by 

231 sort_order = 1 

232 collection = get_collection(devices_states_database, device_id, create=True) 

233 total = collection.count_documents(req_filter) 

234 cursor = collection.find(req_filter).sort(sort_field, sort_order).skip(query.offset) 

235 if (query.limit is not None) and (query.limit != 0): 

236 cursor = cursor.limit(query.limit) 

237 for item_dict in cursor: 

238 items.append( 

239 cls( 

240 timestamp=item_dict.get("precise_timestamp"), 

241 mode_id=item_dict.get("mode_id", None), 

242 load=item_dict.get("load", None), 

243 tokens=item_dict.get("tokens", Field(default_factory=list)), 

244 ) 

245 ) 

246 return ListResponse(items=items, limit=query.limit, offset=query.offset, sort_by=query.sort_by, total=total) 

247 

248 

249class SignalSample(TwinPadModel): 

250 signal_id: str 

251 timestamp: float 

252 value: float | int | str | bool | None 

253 forced_value: float | int | str | bool | None = None 

254 

255 @classmethod 

256 def get_first_from_signal_id(cls, signal_id: str) -> Self | None: 

257 

258 collection = get_signal_collection(signal_id) 

259 if collection is None: 

260 return None 

261 

262 # This is a workaround using the internal collection associated to a timeseries to find the most/least recent document 

263 # found here: https://www.mongodb.com/community/forums/t/time-series-how-to-get-the-most-recent-document/149455/4 

264 bucket = get_signal_collection(f"system.buckets.{signal_id}") 

265 last_bucket = None 

266 if bucket is not None: 

267 last_bucket = bucket.find_one({}, sort={"control.min.timestamp": 1}) 

268 if last_bucket is not None: 

269 sample_data = collection.find_one({"timestamp": last_bucket["control"]["min"]["timestamp"]}) 

270 else: 

271 sample_data = collection.find_one({}, sort=[("timestamp", 1)]) 

272 

273 if sample_data is None: 

274 return None 

275 

276 timestamp = sample_data["precise_timestamp"] 

277 

278 return cls( 

279 signal_id=signal_id, 

280 timestamp=timestamp, 

281 value=sample_data.get("value", None), 

282 forced_value=sample_data.get("forced_value", None), 

283 ) 

284 

285 @classmethod 

286 def get_first_from_signal_ids(cls, signal_ids: list[str]) -> Self | None: 

287 return [cls.get_first_from_signal_id(sid) for sid in signal_ids] 

288 

289 @classmethod 

290 def get_last_from_signal_id(cls, signal_id: str) -> Self | None: 

291 

292 collection = get_signal_collection(signal_id) 

293 if collection is None: 

294 return None 

295 

296 # This is a workaround using the internal collection associated to a timeseries to find the most/least recent document 

297 # found here: https://www.mongodb.com/community/forums/t/time-series-how-to-get-the-most-recent-document/149455/4 

298 bucket = get_signal_collection(f"system.buckets.{signal_id}") 

299 last_bucket = None 

300 if bucket is not None: 

301 last_bucket = bucket.find_one({}, sort={"control.max.timestamp": -1}) 

302 if last_bucket is not None: 

303 sample_data = collection.find_one({"timestamp": last_bucket["control"]["max"]["timestamp"]}) 

304 else: 

305 sample_data = collection.find_one({}, sort=[("timestamp", -1)]) 

306 

307 if sample_data is None: 

308 return None 

309 

310 timestamp = sample_data["precise_timestamp"] 

311 device = Device.get_one_by_attribute("device_id", signal_id.split(".")[0]) 

312 if device is not None and device.last_ping is not None: 

313 if timestamp is None: 

314 timestamp = device.last_ping 

315 else: 

316 timestamp = max(timestamp, device.last_ping) 

317 return cls( 

318 signal_id=signal_id, 

319 timestamp=timestamp, 

320 value=sample_data.get("value", None), 

321 forced_value=sample_data.get("forced_value", None), 

322 ) 

323 

324 @classmethod 

325 def get_last_from_signal_ids(cls, signal_ids: list[str]) -> Self | None: 

326 return [cls.get_last_from_signal_id(sid) for sid in signal_ids] 

327 

328 

329class SignalData(TwinPadModel): 

330 signal_id: str 

331 time_vector: list[float] 

332 values: list[float | int | str | None] 

333 forced_values: list[float | int | str | None] 

334 

335 data_start: float | None = None 

336 data_end: float | None = None 

337 

338 number_samples: int = 0 

339 number_samples_db: int = 0 

340 

341 db_query_time: float = 0.0 

342 init_time: float = 0.0 

343 data_processing_time: float = 0.0 

344 

345 @classmethod 

346 def get_from_signal_id( 

347 cls, 

348 signal_id: str, 

349 min_timestamp: float = None, 

350 max_timestamp: float = None, 

351 interpolate_bounds: bool = True, 

352 max_documents: int = None, 

353 ) -> Self: 

354 

355 now = time.time() 

356 

357 req_signal = {} 

358 if min_timestamp is not None: 

359 req_signal.setdefault("timestamp", {}) 

360 req_signal["timestamp"]["$gte"] = datetime.datetime.fromtimestamp(min_timestamp) 

361 if max_timestamp is not None: 

362 req_signal.setdefault("timestamp", {}) 

363 req_signal["timestamp"]["$lte"] = datetime.datetime.fromtimestamp(max_timestamp) 

364 

365 collection = get_signal_collection(signal_id) 

366 if collection is None: 

367 return cls( 

368 signal_id=signal_id, time_vector=[], values=[], forced_values=[], number_samples=0, number_samples_db=0 

369 ) 

370 

371 db_req_start = time.time() 

372 

373 sort_step = {"$sort": {"precise_timestamp": 1}} 

374 number_results = collection.count_documents(req_signal) 

375 

376 pipeline = [] 

377 if req_signal: 

378 pipeline.append({"$match": req_signal}) # Filter data if needed 

379 

380 pipeline.extend( 

381 [ 

382 {"$project": {"precise_timestamp": 1, "value": 1, "forced_value": 1, "_id": 0}}, 

383 sort_step, 

384 ] 

385 ) 

386 

387 if max_documents is not None and max_documents < number_results: 

388 unsampling_ratio = math.ceil(number_results / max_documents) 

389 logger.info(f"unsampling ratio: {unsampling_ratio} (number_results:{number_results})") 

390 pipeline.extend( 

391 [ 

392 { 

393 "$setWindowFields": { 

394 "sortBy": {"precise_timestamp": 1}, 

395 "output": {"index": {"$documentNumber": {}}}, 

396 } 

397 }, 

398 {"$set": {"group_id": {"$floor": {"$divide": ["$index", unsampling_ratio]}}}}, 

399 {"$group": {"_id": "$group_id", "doc": {"$first": "$$ROOT"}}}, 

400 {"$replaceRoot": {"newRoot": "$doc"}}, 

401 {"$unset": ["index", "group_id"]}, 

402 {"$sort": {"precise_timestamp": 1}}, 

403 ] 

404 ) 

405 

406 # logger.info(f"pipeline: {pipeline}") 

407 cursor = collection.aggregate(pipeline) 

408 db_req_time = time.time() - db_req_start 

409 

410 init_time = time.time() 

411 number_samples_db = collection.count_documents(req_signal) 

412 

413 results = list(cursor) 

414 time_vector = [] 

415 values = [] 

416 forced_values = [] 

417 for s in results: 

418 time_vector.append(s["precise_timestamp"]) 

419 values.append(s.get("value", None)) 

420 forced_values.append(s.get("forced_value", None)) 

421 

422 signal = Signal.get_from_signal_id(signal_id) 

423 class_ = signal.signal_data_class 

424 

425 if interpolate_bounds: 

426 # Fetching left side value & interpolation 

427 if min_timestamp and (not time_vector or time_vector[0] != min_timestamp): 

428 sample_left = collection.find_one( 

429 { 

430 "timestamp": {"$lte": datetime.datetime.fromtimestamp(min_timestamp)}, 

431 "value": {"$exists": True}, 

432 }, 

433 sort=[("timestamp", -1)], 

434 ) 

435 

436 if sample_left: 

437 if time_vector: 

438 left_sd = class_( 

439 signal_id=signal_id, 

440 time_vector=[sample_left["precise_timestamp"], time_vector[0]], 

441 values=[sample_left["value"], values[0]], 

442 forced_values=[sample_left.get("forced_value", None), forced_values[0]], 

443 ) 

444 min_ts_value = left_sd.interpolate_values([min_timestamp])[0] 

445 min_ts_forced_value = left_sd.interpolate_forced_values([min_timestamp])[0] 

446 else: 

447 min_ts_value = sample_left.get("value", None) 

448 min_ts_forced_value = sample_left.get("forced_value", None) 

449 time_vector.insert(0, min_timestamp) 

450 values.insert(0, min_ts_value) 

451 forced_values.insert(0, min_ts_forced_value) 

452 

453 # Fetching right side value & interpolation 

454 if max_timestamp is not None and (not time_vector or time_vector[-1] != max_timestamp): 

455 sample_right = collection.find_one( 

456 { 

457 "timestamp": {"$gte": datetime.datetime.fromtimestamp(max_timestamp)}, 

458 "value": {"$exists": True}, 

459 }, 

460 sort=[("timestamp", 1)], 

461 ) 

462 if sample_right: 

463 if time_vector: 

464 right_sd = class_( 

465 signal_id=signal_id, 

466 time_vector=[time_vector[-1], sample_right["precise_timestamp"]], 

467 values=[values[-1], sample_right.get("value", None)], 

468 forced_values=[forced_values[-1], sample_right.get("forced_value", None)], 

469 ) 

470 max_ts_value = right_sd.interpolate_values([max_timestamp])[0] 

471 max_ts_forced_value = right_sd.interpolate_forced_values([max_timestamp])[0] 

472 else: 

473 max_ts_value = sample_right.get("value", None) 

474 max_ts_forced_value = sample_right.get("forced_value", None) 

475 time_vector.append(max_timestamp) 

476 values.append(max_ts_value) 

477 forced_values.append(max_ts_forced_value) 

478 

479 if values: 

480 # TODO: check below. a bit strange 

481 if ((max_timestamp is None) or (max_timestamp >= now)) and (now - time_vector[-1]) > DEVICE_TIMEOUT: 

482 # Adding last value as it should be repeated 

483 time_vector.append(now) 

484 values.append(values[-1]) 

485 forced_values.append(forced_values[-1]) 

486 

487 init_time = time.time() - init_time 

488 

489 # See line 292 for explanation 

490 bucket = get_signal_collection(f"system.buckets.{signal_id}") 

491 first_bucket = None 

492 if bucket is not None: 

493 first_bucket = bucket.find_one({}, sort={"control.min.timestamp": 1}) 

494 if first_bucket is not None: 

495 data_start = first_bucket["control"]["min"]["timestamp"].replace(tzinfo=pytz.UTC).timestamp() 

496 else: 

497 data_start = None 

498 

499 if signal.repeated_sample(): 

500 data_end = time.time() 

501 else: 

502 last_bucket = None 

503 if bucket is not None: 

504 last_bucket = bucket.find_one({}, sort={"control.max.timestamp": -1}) 

505 if last_bucket is not None: 

506 data_end = last_bucket["control"]["max"]["timestamp"].replace(tzinfo=pytz.UTC).timestamp() 

507 else: 

508 data_end = time.time() 

509 

510 return class_( 

511 signal_id=signal_id, 

512 time_vector=time_vector, 

513 values=values, 

514 forced_values=forced_values, 

515 data_start=data_start, 

516 data_end=data_end, 

517 number_samples=len(values), 

518 number_samples_db=number_samples_db, 

519 db_query_time=db_req_time, 

520 init_time=init_time, 

521 ) 

522 

523 def interpolate_values(self, new_time_vector: list[float]): 

524 return self.interpolate(new_time_vector, self.values) 

525 

526 def interpolate_forced_values(self, new_time_vector: list[float]): 

527 return self.interpolate(new_time_vector, self.forced_values) 

528 

529 def uniform_desampling(self, number_samples_max: int) -> Self: 

530 data_processing_time = time.time() 

531 if number_samples_max and self.number_samples > number_samples_max: 

532 new_time_vector = npy.linspace( 

533 self.time_vector[0], self.time_vector[-1], number_samples_max, endpoint=False 

534 ).tolist() 

535 values = self.interpolate_values(new_time_vector) 

536 forced_values = self.interpolate_forced_values(new_time_vector) 

537 time_vector = new_time_vector 

538 number_samples = len(time_vector) 

539 else: 

540 time_vector = self.time_vector 

541 number_samples = len(self.values) 

542 values = self.values[:] 

543 forced_values = self.forced_values[:] 

544 data_processing_time = time.time() - data_processing_time 

545 

546 return self.__class__( 

547 signal_id=self.signal_id, 

548 time_vector=time_vector, 

549 values=values, 

550 forced_values=forced_values, 

551 number_samples=number_samples, 

552 number_samples_db=self.number_samples, 

553 data_start=self.data_start, 

554 data_end=self.data_end, 

555 db_query_time=self.db_query_time, 

556 init_time=self.init_time, 

557 data_processing_time=self.data_processing_time + data_processing_time, 

558 ) 

559 

560 def interest_window_desampling( 

561 self, 

562 window_max_number_samples: int, 

563 outside_max_number_samples: int, 

564 window_min_timestamp: float | None = None, 

565 window_max_timestamp: float | None = None, 

566 ) -> Self: 

567 """Performs a sampling in a window of interest and outside.""" 

568 

569 if not self.time_vector: 

570 return self 

571 

572 if window_min_timestamp is None: 

573 window_min_timestamp = self.time_vector[0] 

574 if window_max_timestamp is None: 

575 window_max_timestamp = self.time_vector[-1] 

576 

577 data_processing_time = time.time() 

578 

579 index_window_start = bisect.bisect_left(self.time_vector, window_min_timestamp) 

580 index_window_end = bisect.bisect_right(self.time_vector, window_max_timestamp) 

581 

582 time_vector_before = self.time_vector[:index_window_start] 

583 time_vector_window = self.time_vector[index_window_start:index_window_end] 

584 time_vector_after = self.time_vector[index_window_end:] 

585 

586 # Resampling window 

587 if time_vector_window: 

588 # Ensurring window bounds 

589 if time_vector_window[0] != window_min_timestamp: 

590 time_vector_window.insert(0, window_min_timestamp) 

591 if time_vector_window[-1] != window_max_timestamp: 

592 time_vector_window.append(window_max_timestamp) 

593 else: 

594 time_vector_window = [window_min_timestamp, window_max_timestamp] 

595 

596 if len(time_vector_window) > window_max_number_samples: 

597 # Resampling 

598 new_window_time_vector = npy.linspace( 

599 time_vector_window[0], time_vector_window[-1], window_max_number_samples, endpoint=True 

600 ).tolist() 

601 time_vector_window = new_window_time_vector 

602 

603 # Resampling outside 

604 number_samples_before = len(time_vector_before) 

605 number_samples_after = len(time_vector_after) 

606 if (number_samples_before + number_samples_after) > outside_max_number_samples: 

607 new_number_samples_before = math.ceil( 

608 outside_max_number_samples * number_samples_before / (number_samples_before + number_samples_after) 

609 ) 

610 new_number_samples_after = math.ceil( 

611 outside_max_number_samples * number_samples_after / (number_samples_before + number_samples_after) 

612 ) 

613 # Adjusting numbers as math.ceil can do +1 on sum 

614 if (new_number_samples_before + new_number_samples_after) > outside_max_number_samples: 

615 if new_number_samples_before > new_number_samples_after: 

616 new_number_samples_before -= 1 

617 else: 

618 new_number_samples_after -= 1 

619 

620 if new_number_samples_before: 

621 new_time_vector_before = npy.linspace( 

622 time_vector_before[0], time_vector_before[-1], new_number_samples_before, endpoint=False 

623 ).tolist() 

624 time_vector_before = new_time_vector_before 

625 

626 if number_samples_after: 

627 new_time_vector_after = npy.linspace( 

628 time_vector_after[-1], time_vector_after[0], new_number_samples_after, endpoint=False 

629 ).tolist()[::-1] 

630 time_vector_after = new_time_vector_after 

631 

632 new_time_vector = time_vector_before + time_vector_window + time_vector_after 

633 values = self.interpolate_values(new_time_vector) 

634 forced_values = self.interpolate_forced_values(new_time_vector) 

635 number_samples = len(values) 

636 

637 data_processing_time = time.time() - data_processing_time 

638 

639 # logger.warning(f"samples: {samples}") 

640 

641 return self.__class__( 

642 signal_id=self.signal_id, 

643 time_vector=new_time_vector, 

644 values=values, 

645 forced_values=forced_values, 

646 number_samples=number_samples, 

647 number_samples_db=self.number_samples, 

648 data_start=self.data_start, 

649 data_end=self.data_end, 

650 db_query_time=self.db_query_time, 

651 init_time=self.init_time, 

652 data_processing_time=self.data_processing_time + data_processing_time, 

653 ) 

654 

655 def csv_export(self): 

656 output = io.StringIO() 

657 writer = csv.writer(output) 

658 writer.writerow(["timestamp", "value", "forced_value"]) # Write header 

659 for ts, value, forced_value in zip(self.time_vector, self.values, self.forced_values): 

660 writer.writerow([get_utc_date_from_timestamp(ts), value, forced_value]) 

661 return output.getvalue().encode("utf-8") 

662 

663 def prestoplot_export(self): 

664 clean_signal_id = self.signal_id.replace(".", "_") 

665 if clean_signal_id[0].isnumeric(): 

666 clean_signal_id = "_" + clean_signal_id 

667 

668 output = io.StringIO() 

669 output.write("# Encoding:\tUTF-8\n") 

670 output.write(f"t\t{clean_signal_id}\t{clean_signal_id}_forced_value\n") 

671 output.write("ISO8601\tnone\tnone\n") 

672 output.write(f"# Description :\t{clean_signal_id}\n") 

673 

674 for ts, value, forced_value in zip(self.time_vector, self.values, self.forced_values): 

675 output.write( 

676 f"{get_utc_date_from_timestamp(ts)}\t{value if value is not None else 'none'}\t{forced_value if forced_value is not None else 'none'}\n" 

677 ) 

678 return output.getvalue().encode("utf-8") 

679 

680 

681class NumericSignalData(SignalData): 

682 data_type: str = "float" 

683 values: list[float | int | None] 

684 forced_values: list[float | int | None] 

685 

686 def interpolate(self, new_time_vector: list[float], items): 

687 items = [npy.nan if s is None else s for s in items] 

688 return [None if npy.isnan(s) else s for s in npy.interp(new_time_vector, self.time_vector, items)] 

689 

690 

691class StringSignalData(SignalData): 

692 data_type: str = "str" 

693 values: list[str | None] 

694 forced_values: list[str | None] 

695 

696 def interpolate(self, new_time_vector: list[float], items): 

697 # Find the indices of the values in xp that are just smaller or equal to x 

698 indices = npy.searchsorted(self.time_vector, new_time_vector, side="right") - 1 

699 indices = npy.clip(indices, 0, len(items) - 1) 

700 # Return the corresponding left string values from fp 

701 return [items[i] for i in indices] 

702 

703 

704class SignalsData(TwinPadModel): 

705 signals_data: list[NumericSignalData | StringSignalData | SignalData] 

706 data_processing_time: float 

707 data_start: float | None 

708 data_end: float | None 

709 

710 @classmethod 

711 def get_from_signal_ids( 

712 cls, signal_ids: list[str], min_timestamp: float = None, max_timestamp: float = None, max_documents: int = None 

713 ) -> Self: 

714 signals_data = [] 

715 data_start = None 

716 data_end = None 

717 if max_timestamp is None: 

718 max_timestamp = time.time() 

719 data_processing_time = 0.0 

720 for signal_id in signal_ids: 

721 signal_data = SignalData.get_from_signal_id( 

722 signal_id=signal_id, 

723 min_timestamp=min_timestamp, 

724 max_timestamp=max_timestamp, 

725 max_documents=max_documents, 

726 ) 

727 data_processing_time += signal_data.data_processing_time 

728 signals_data.append(signal_data) 

729 if signal_data.data_start is not None: 

730 if data_start is None: 

731 data_start = signal_data.data_start 

732 else: 

733 data_start = min(signal_data.data_start, data_start) 

734 if signal_data.data_end is not None: 

735 if data_end is None: 

736 data_end = signal_data.data_end