Add sort of non-blockong shell

This commit is contained in:
Igor Dunaev 2023-11-15 23:55:37 +03:00
parent 3dbd8309d0
commit e70077c9bc
4 changed files with 164 additions and 14 deletions

110
controls/either.py Normal file
View File

@ -0,0 +1,110 @@
from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Callable
L = TypeVar('L')
R = TypeVar('R')
N = TypeVar('N')
class Either(ABC, Generic[L, R]):
'A monad-like object to handle errors in a functional style.'
@abstractmethod
def is_left(self) -> bool:
'Returns `True` if the objects belongs to `Left` class'
pass
@abstractmethod
def is_right(self) -> bool:
'Returns `True` if the objects belongs to `Right` class'
pass
@abstractmethod
def map_left(self, f: Callable[[L], N]) -> 'Either[N, R]':
'''Applies transform `f` to the value if type is `Left`,
but does not touch the value if type is `Right`.'''
pass
@abstractmethod
def map_right(self, f: Callable[[R], N]) -> 'Either[L, N]':
'''Applies transform `f` to the right if type is `Right`,
but does not touch the value if type is `Left`.'''
pass
@abstractmethod
def get_left_or(self, default: L) -> L:
'''Unwraps the underlying value if `Left`, otherwise returns the provided default value.'''
pass
@abstractmethod
def get_left(self) -> L:
'''Unwraps the underlying value if `Left`, panics if `Right`.'''
pass
@abstractmethod
def get_right_or(self, default: R) -> R:
'''Unwraps the underlying value if `Right`, otherwise returns the provided default value.'''
pass
@abstractmethod
def get_right(self) -> R:
'''Unwraps the underlying error if `Right`, panics if `Left`.'''
pass
class Left(Either[L, R]):
def __init__(self, value: L) -> None:
super().__init__()
self._left_value = value
def is_left(self) -> bool:
return True
def is_right(self) -> bool:
return not self.is_left()
def map_left(self, f: Callable[[L], N]) -> 'Either[N, R]':
return Left(f(self._left_value))
def map_right(self, f: Callable[[R], N]) -> 'Either[L, N]':
return Left(self._left_value)
def get_left_or(self, default: L) -> L:
return self._left_value
def get_left(self) -> L:
return self._left_value
def get_right_or(self, default: R) -> R:
return default
def get_right(self) -> R:
raise RuntimeError(f'Object {self.__repr__()} is nor of Error type')
class Right(Either[L, R]):
def __init__(self, value: R) -> None:
super().__init__()
self._right_value = value
def is_left(self) -> bool:
return False
def is_right(self) -> bool:
return not self.is_left()
def map_left(self, f: Callable[[L], N]) -> 'Either[N, R]':
return Right(self._right_value)
def map_right(self, f: Callable[[R], N]) -> 'Either[L, N]':
return Right(f(self._right_value))
def get_left_or(self, default: L) -> L:
return default
def get_left(self) -> L:
raise RuntimeError(f'Attempt to unwrap object {self.__repr__()}, which is of Error type')
def get_right_or(self, default: R) -> R:
return self._right_value
def get_right(self) -> R:
return self._right_value

View File

@ -1,4 +1,4 @@
from turtle import Turtle from turtle import Turtle, bye
from typing import Any, Collection, Optional from typing import Any, Collection, Optional
from types import FunctionType, MethodType from types import FunctionType, MethodType
from inspect import getfullargspec from inspect import getfullargspec
@ -29,7 +29,7 @@ class TurtleDevice(SynchronyDevice):
if self.state == DeviceLifecycleState.CLOSE: if self.state == DeviceLifecycleState.CLOSE:
return return
self._logical_traits = {k: None for k in self._logical_traits.keys()} self._logical_traits = {k: None for k in self._logical_traits.keys()}
self._device = None bye()
super().close() super().close()
def action_descriptors(self) -> Collection[ActionDescriptor]: def action_descriptors(self) -> Collection[ActionDescriptor]:

View File

@ -2,16 +2,49 @@ import cmd
import threading import threading
from queue import Queue from queue import Queue
from functools import reduce
from typing import Collection, Any, Optional
from controls.device import ActionDescriptor, DeviceError
from controls.either import Either, Left, Right
from equipment.turtle_device import TurtleDevice from equipment.turtle_device import TurtleDevice
class TurtleDeviceThread(threading.Thread): def select(fr: Collection[ActionDescriptor], where: tuple[str, Any]) -> Optional[ActionDescriptor]:
# TODO(Homework 4) results = list(filter(lambda r: r.__dict__[where[0]] == where[1], fr))
return None if len(results) == 0 else results[0]
class TurtleDeviceThread():
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.device = TurtleDevice() self._device = TurtleDevice()
self.queue = Queue() self._device.open()
self._queue = Queue()
def send(self, task: list[str]) -> None:
self._queue.put(task)
def _parse_args(self, args: list[str]) -> Either[tuple, str]:
action_name, action_args = args[0], args[1:]
try:
descr = select(self._device.action_descriptors(), ('name', action_name))
except DeviceError as e:
return Right(f'Device error: {e}')
if descr is None:
return Right(f'Unknown action: "{action_name}"')
return Left(reduce(lambda acc, a: acc + (int(a),), action_args, (action_name,)))
def run(self): # ! NoReturn
while (True):
task = self._queue.get(block=True, timeout=None)
if task[0] == 'stop':
self._device.close()
return
args = self._parse_args(task)
if args.is_right():
print(f'*** {args.get_right()}')
else:
self._device.execute(*args.get_left())
class NoBlockingTurtleShell(cmd.Cmd): class NoBlockingTurtleShell(cmd.Cmd):
@ -20,16 +53,23 @@ class NoBlockingTurtleShell(cmd.Cmd):
file = None file = None
def __init__(self, turtle_thread: TurtleDeviceThread): def __init__(self, turtle_thread: TurtleDeviceThread):
pass # TODO(Homework 4) super(NoBlockingTurtleShell, self).__init__()
self._turtle_thread = turtle_thread
def do_execute(self, arg): def do_execute(self, arg: str):
pass # TODO(Homework 4) 'Execute a turtle command: EXECUTE COMMAND ARG1 ARG2 ...'
self._turtle_thread.send(arg.split(' '))
def do_exit(self, arg): def do_exit(self, arg):
pass # TODO(Homework 4) 'close the turtle window, and exit: EXIT'
print('Waiting for the turtle to finish jobs...')
self._turtle_thread.send(['stop'])
return True
if __name__ == '__main__': if __name__ == '__main__':
turtle_thread = TurtleDeviceThread() turtle_thread = TurtleDeviceThread()
# TODO(Homework 4: Correct start thread) shell = NoBlockingTurtleShell(turtle_thread)
NoBlockingTurtleShell(turtle_thread).cmdloop() cmd_thread = threading.Thread(target=shell.cmdloop)
cmd_thread.start()
turtle_thread.run()

View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "controls" name = "controls"
version = "0.0.2" version = "0.0.3"
license = {file = "LICENSE"} license = {file = "LICENSE"}
description = "Example SCADA System" description = "Example SCADA System"
authors = [{name = "teldufalsari et al (cockroaches)"}] authors = [{name = "teldufalsari et al (cockroaches)"}]