Made HW3(some part)

scada_system/controls/device.py

@@ -0,0 +1,88 @@
+from dataclasses import dataclass
+from typing import Optional, Collection, Any
+from abc import ABCMeta, abstractmethod
+from enum import Enum
+
+class DeviceLifecycleState(Enum):
+    INIT = "INIT"
+    OPEN = "OPEN"
+    CLOSE = "CLOSE"
+
+
+class DevaceError(Exception):
+    pass
+
+
+class NonReadableTrait(Exception):
+    pass
+
+
+class NonWritableTrait(Exception):
+    pass
+
+
+@dataclass
+class TraitDescriptor:
+    name: str
+    info: Optional[str] = None
+    readable: bool = True
+    writable: bool = False
+
+
+@dataclass
+class ActionDescriptor:
+    name: str
+    arguments: dict[str, type]
+    info: Optional[str] = None
+
+
+class Device(metaclass=ABCMeta):
+    _state = DeviceLifecycleState.INIT
+
+    @property
+    def state(self) -> DeviceLifecycleState:
+        return self._state
+
+    def close(self):
+        self._state = DeviceLifecycleState.CLOSE
+
+    @property
+    @abstractmethod
+    def trait_descriptors(self) -> Collection[TraitDescriptor]:
+        pass
+    
+    @property
+    @abstractmethod
+    def action_descriptors(self) -> Collection[ActionDescriptor]:
+        pass
+
+    @abstractmethod
+    def __getitem__(self, trait_name: str) -> Optional[Any]:
+        """Return logical state of trait `trait_name`."""
+        pass
+
+
+class SynchronyDevice(Device):
+
+    def open(self):
+        self._state = DeviceLifecycleState.OPEN
+
+    @abstractmethod
+    def execute(self, action_name: str, *args, **kwargs):
+        """Execute action `action_name`, using `args` and `kwargs` as action argument."""
+        pass
+
+    @abstractmethod
+    def read(self, trait_name: str) -> Any:
+        """Read physical state of trait `trait_name` from device."""
+        raise NonReadableTrait
+
+    @abstractmethod
+    def write(self, trait_name: str, value: Any) -> bool:
+        """Pass `value` to trait `trait_name` of device."""
+        raise NonWritableTrait
+
+    @abstractmethod
+    def invalidate(self, trait_name: str):
+        """Invalidate logical state of trait `trait_name`"""
+        pass

scada_system/equipment/turtle_device.py

@@ -0,0 +1,40 @@
+import turtle
+from typing import Optional, Collection, Any
+from controls.device import SynchronyDevice
+import time
+import inspect
+
+class TurtleDevice(SynchronyDevice): 
+    def open(self):
+        super().open()
+        turtle.Turtle()
+    
+    def close(self):
+        super().close()
+        turtle.bye()  
+        
+    def execute(self, action_name: str, *args, **kwargs):
+        getattr(turtle, action_name)(args[0], *kwargs)
+        
+    def read(self, trait_name: str):
+        pass
+
+    def write(self, trait_name: str, value: Any) -> bool:
+        pass
+
+    def invalidate(self, trait_name: str):
+        pass
+
+    def trait_descriptors(self):
+        traits = [attr for attr in turtle.Turtle.__dict__
+                      if attr.startswith('__') and not attr.startswith('_')]
+        return traits
+    
+    def action_descriptors(self):
+        actions = [(name, func.__doc__) for name, func in inspect.getmembers(turtle.Turtle)
+                   if callable(func) and not name.startswith('__') and not name.startswith('_')]
+        return actions
+
+    def __getitem__(self, trait_name: str):
+        """Return logical state of trait `trait_name`."""
+        pass

scada_system/tests/controls/test_device.py

@@ -0,0 +1,12 @@
+from unittest import TestCase
+
+from controls.device import DeviceLifecycleState
+
+
+class DeviceLifecycleStateTest(TestCase):
+
+    def setUp(self) -> None:
+        pass
+
+    def test_enum(self):
+        self.assertEqual(DeviceLifecycleStateTest["INIT"], DeviceLifecycleStateTest.INIT)

scada_system/tests/equipment/test_turtle_device.py

@@ -0,0 +1,13 @@
+from unittest import TestCase
+
+from equipment.turtle_device import TurtleDevice
+
+
+class TurtleDeviceTest(TestCase):
+
+    def setUp(self) -> None:
+        self.device = TurtleDevice()
+
+    def test_open(self):
+        self.device.open()
+        self.device.close()