Compare commits
3 Commits
82e09bfbf3
...
3f4f48fb55
Author | SHA1 | Date | |
---|---|---|---|
3f4f48fb55 | |||
0cf47dbf82 | |||
|
fb45d7291c |
91
controls/device.py
Normal file
91
controls/device.py
Normal file
@ -0,0 +1,91 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Collection, Any
|
||||
from abc import ABC, abstractmethod
|
||||
from enum import Enum, auto
|
||||
|
||||
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(ABC):
|
||||
_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
|
||||
|
||||
def close(self):
|
||||
return super().close()
|
0
equipment/__init__.py
Normal file
0
equipment/__init__.py
Normal file
155061
equipment/turtle
Normal file
155061
equipment/turtle
Normal file
File diff suppressed because it is too large
Load Diff
66
equipment/turtle_device.py
Normal file
66
equipment/turtle_device.py
Normal file
@ -0,0 +1,66 @@
|
||||
from turtle import Turtle
|
||||
from typing import Any, Collection, Optional
|
||||
from controls.device import ActionDescriptor, SynchronyDevice, TraitDescriptor
|
||||
|
||||
class TurtleDevice(SynchronyDevice):
|
||||
def open(self, name, **kwargs):
|
||||
self.turtle = Turtle(kwargs)
|
||||
super().open()
|
||||
|
||||
def close(self):
|
||||
del self.turtle
|
||||
super().close()
|
||||
|
||||
# @property
|
||||
# def trait_descriptors(self):
|
||||
# traits = [attribute for attribute in vars(self.turtle).items()
|
||||
# if (attribute[0].startswith('_') == False)]
|
||||
|
||||
# return Collection[[TraitDescriptor(*tr) for tr in traits]]
|
||||
|
||||
@property
|
||||
def trait_descriptors(self):
|
||||
traits = [attribute for attribute in vars(self.turtle).items()
|
||||
if (attribute[0].startswith('_') == False)]
|
||||
|
||||
self.traits = [TraitDescriptor(*tr) for tr in traits]
|
||||
return self.traits
|
||||
|
||||
# @property
|
||||
# def action_descriptors(self):
|
||||
# actions = [(attribute, getattr(t, attribute)) for attribute in dir(t)
|
||||
# if (attribute.startswith('_') == False)]
|
||||
|
||||
# return Collection[[TraitDescriptor(*ac) for ac in actions]]
|
||||
|
||||
@property
|
||||
def action_descriptors(self):
|
||||
actions = [(attribute, getattr(t, attribute)) for attribute in dir(t)
|
||||
if (attribute.startswith('_') == False)]
|
||||
|
||||
self.actions = [TraitDescriptor(*ac) for ac in actions]
|
||||
return self.acions
|
||||
|
||||
def read(self, trait_name):
|
||||
collection = self.traits
|
||||
return ([collection[i].info for i in range(len(collection))
|
||||
if collection[i].name == trait_name])
|
||||
|
||||
def write(self, trait_name: str, value: Any) -> bool:
|
||||
collection = self.traits
|
||||
for i in range(len(collection)):
|
||||
if collection[i].name == trait_name:
|
||||
collection[i].info=value
|
||||
|
||||
def execute(self, action_name: str, *args, **kwargs):
|
||||
self.turtle.action_name(args, *kwargs)
|
||||
|
||||
def invalidate(self, trait_name: str):
|
||||
return super().invalidate(trait_name)
|
||||
|
||||
def __getitem__(self, trait_name: str) -> Any | None:
|
||||
collection = self.traits
|
||||
return ([collection[i] for i in range(len(collection))
|
||||
if collection[i].name == trait_name])
|
||||
|
||||
|
12
tests/for_controls/test_device.py
Normal file
12
tests/for_controls/test_device.py
Normal file
@ -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)
|
13
tests/for_equipment/test_turtle_device.py
Normal file
13
tests/for_equipment/test_turtle_device.py
Normal file
@ -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()
|
Loading…
Reference in New Issue
Block a user