_scenario_model.py 2.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768
  1. # Copyright 2021-2024 Avaiga Private Limited
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  4. # the License. You may obtain a copy of the License at
  5. #
  6. # http://www.apache.org/licenses/LICENSE-2.0
  7. #
  8. # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
  9. # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
  10. # specific language governing permissions and limitations under the License.
  11. from dataclasses import dataclass
  12. from typing import Any, Dict, List, Optional
  13. from .._repository._base_taipy_model import _BaseModel
  14. from ..cycle.cycle_id import CycleId
  15. from ..data.data_node_id import DataNodeId
  16. from ..task.task_id import TaskId
  17. from .scenario_id import ScenarioId
  18. @dataclass
  19. class _ScenarioModel(_BaseModel):
  20. id: ScenarioId
  21. config_id: str
  22. tasks: List[TaskId]
  23. additional_data_nodes: List[DataNodeId]
  24. properties: Dict[str, Any]
  25. creation_date: str
  26. primary_scenario: bool
  27. subscribers: List[Dict]
  28. tags: List[str]
  29. version: str
  30. sequences: Optional[Dict[str, Dict]] = None
  31. cycle: Optional[CycleId] = None
  32. @staticmethod
  33. def from_dict(data: Dict[str, Any]):
  34. return _ScenarioModel(
  35. id=data["id"],
  36. config_id=data["config_id"],
  37. tasks=_BaseModel._deserialize_attribute(data["tasks"]),
  38. additional_data_nodes=_BaseModel._deserialize_attribute(data["additional_data_nodes"]),
  39. properties=_BaseModel._deserialize_attribute(data["properties"]),
  40. creation_date=data["creation_date"],
  41. primary_scenario=data["primary_scenario"],
  42. subscribers=_BaseModel._deserialize_attribute(data["subscribers"]),
  43. tags=_BaseModel._deserialize_attribute(data["tags"]),
  44. version=data["version"],
  45. sequences=_BaseModel._deserialize_attribute(data["sequences"]),
  46. cycle=CycleId(data["cycle"]) if "cycle" in data else None,
  47. )
  48. def to_list(self):
  49. return [
  50. self.id,
  51. self.config_id,
  52. _BaseModel._serialize_attribute(self.tasks),
  53. _BaseModel._serialize_attribute(self.additional_data_nodes),
  54. _BaseModel._serialize_attribute(self.properties),
  55. self.creation_date,
  56. self.primary_scenario,
  57. _BaseModel._serialize_attribute(self.subscribers),
  58. _BaseModel._serialize_attribute(self.tags),
  59. self.version,
  60. _BaseModel._serialize_attribute(self.sequences),
  61. self.cycle,
  62. ]