feat: architecture, some problems to solve before the machine can step
This commit is contained in:
@@ -0,0 +1 @@
|
||||
.idea
|
||||
@@ -0,0 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2026 Studiosi
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@@ -0,0 +1,10 @@
|
||||
# SNanoSM
|
||||
|
||||
A pure python, minimalistic, typed library for the implementation of Mealy Finite State Machines.
|
||||
|
||||
Part of the Nobody Industry's MFFP (made from first principles) set of libraries.
|
||||
|
||||
## Features
|
||||
|
||||
- Typed
|
||||
- Full test coverage
|
||||
@@ -0,0 +1,27 @@
|
||||
[build-system]
|
||||
requires = ["hatchling >= 1.26"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "snanosm"
|
||||
version = "0.0.1"
|
||||
authors = [
|
||||
{ name="Studiosi", email="hello@nobodyownsthisdomain.org" },
|
||||
]
|
||||
description = "A pure python small state machine library"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.9"
|
||||
classifiers = [
|
||||
"Programming Language :: Python :: 3",
|
||||
"Operating System :: OS Independent",
|
||||
"Development Status :: 2 - Pre-Alpha",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Topic :: Software Development :: Libraries",
|
||||
"Typing :: Typed"
|
||||
]
|
||||
license = "MIT"
|
||||
license-files = ["LICEN[CS]E*"]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://github.com/pypa/sampleproject"
|
||||
Issues = "https://github.com/pypa/sampleproject/issues"
|
||||
@@ -0,0 +1,119 @@
|
||||
from typing import Optional, Callable, Protocol, Self, TypeVar, Generic, List
|
||||
|
||||
|
||||
# Anything hashable and equatable can be used as an input
|
||||
class InputProtocol(Protocol):
|
||||
def __eq__(self, __o: Self) -> bool:
|
||||
...
|
||||
|
||||
def __hash__(self) -> int:
|
||||
...
|
||||
|
||||
|
||||
Input = TypeVar("Input", bound=InputProtocol)
|
||||
|
||||
|
||||
class State:
|
||||
def __init__(self, name: str, data: object) -> None:
|
||||
self.__name = name
|
||||
self.__data = data
|
||||
|
||||
def get_name(self) -> str:
|
||||
return self.__name
|
||||
|
||||
def get_data(self) -> object:
|
||||
return self.__data
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"[State {self.__name}]"
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(self.__name)
|
||||
|
||||
|
||||
class Transition(Generic[Input]):
|
||||
def __init__(self, state_input: Input, origin_name: str, destination_name: str,
|
||||
output_function: Callable[[Self, object], None]) -> None:
|
||||
self.__state_input = state_input
|
||||
self.__origin_name = origin_name
|
||||
self.__destination_name = destination_name
|
||||
self.__output_function = output_function
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"[Transition ({self.__origin_name}, {self.__destination_name}, {self.__state_input})]"
|
||||
|
||||
def __hash__(self) -> int:
|
||||
# A tuple is only hashable if all its elements are hashable
|
||||
return hash((self.__origin_name, self.__destination_name, self.__state_input))
|
||||
|
||||
|
||||
class Machine(Generic[Input]):
|
||||
def __init__(self) -> None:
|
||||
self.__start_state: Optional[int] = None
|
||||
self.__end_states: List[int] = []
|
||||
self.__current_state: Optional[int] = None
|
||||
self.__states: dict[int, State] = {}
|
||||
self.__transitions: dict[int, Transition] = {}
|
||||
|
||||
def add_transition(self, state_input: Input, origin_name: str, destination_name: str,
|
||||
output_function: Callable[[Self, object], None]) -> None:
|
||||
ho = hash(origin_name)
|
||||
if ho not in self.__states.keys():
|
||||
raise ValueError(f"Origin state {origin_name} does not exist.")
|
||||
hd = hash(destination_name)
|
||||
if hd not in self.__states.keys():
|
||||
raise ValueError(f"Destination state {destination_name} does not exist.")
|
||||
t = Transition(state_input, origin_name, destination_name, output_function)
|
||||
ht = hash(t)
|
||||
if ht in self.__transitions.keys():
|
||||
raise ValueError(f"Transition ({origin_name}, {destination_name}, {state_input}) already exists.")
|
||||
self.__transitions[ht] = t
|
||||
|
||||
def add_state(self, state_name: str, state_data: object, is_start_state: bool = False,
|
||||
is_end_state: bool = False) -> None:
|
||||
h: int = hash(state_name)
|
||||
if is_start_state:
|
||||
if self.__start_state is None:
|
||||
self.__start_state = h
|
||||
else:
|
||||
raise ValueError("Adding a start state when one is already set.")
|
||||
if is_end_state:
|
||||
# Multiple end states possible
|
||||
self.__end_states.append(h)
|
||||
if h in self.__states.keys():
|
||||
raise ValueError(f"State {state_name} already exists.")
|
||||
state = State(state_name, state_data)
|
||||
self.__states[h] = state
|
||||
|
||||
def get_state(self, name: str) -> Optional[State]:
|
||||
h = hash(name)
|
||||
# KeyError if the state is not found
|
||||
return self.__states[h]
|
||||
|
||||
def reset(self) -> None:
|
||||
if self.__start_state is None:
|
||||
raise ValueError("Resetting a machine without starting state.")
|
||||
self.__current_state = self.__start_state
|
||||
|
||||
def get_current_state(self) -> State:
|
||||
if self.__current_state is None:
|
||||
raise ValueError("Getting current state before the machine has one.")
|
||||
return self.__states[self.__current_state]
|
||||
|
||||
def process_input(self, i: Input) -> None:
|
||||
if self.__current_state is None:
|
||||
if self.__start_state is None:
|
||||
raise ValueError("Processing input without a start state.")
|
||||
self.__current_state = self.__start_state
|
||||
# Get all transitions for the current state
|
||||
#transition = self.__transitions[(self.__current_state, i)]
|
||||
#self.__current_state = transition.get_next_state()
|
||||
|
||||
def __str__(self) -> str:
|
||||
r = [f"=MACHINE=", "\tSTATES"]
|
||||
for state in self.__states.values():
|
||||
r.append(f"\t\t{state}")
|
||||
r.append("\tTRANSITIONS")
|
||||
for transition in self.__transitions.values():
|
||||
r.append(f"\t\t{transition}")
|
||||
return "\n".join(r)
|
||||
@@ -0,0 +1,10 @@
|
||||
import unittest
|
||||
|
||||
from src.snanosm.mealy import Machine
|
||||
|
||||
|
||||
class TestMachine(unittest.TestCase):
|
||||
|
||||
def test_init(self):
|
||||
m = Machine()
|
||||
self.assertIsNotNone(m)
|
||||
Reference in New Issue
Block a user