KyMAN/AnP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#wip: Starting AI Interpreters and PseudoLoRAs.

Browse Source
This commit is contained in:
mbruzon 2026-05-29 19:11:54 +02:00
parent 9f2effb6b4
commit 6a2a778110
16 changed files with 205 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
JSON/AnP.routes.json
View File

@ -1,4 +1,4 @@
[
"get:/ /Public",
"post:/ai/new_message ai@new_message"
"[anp]get:/ /Public",
"[anp]post:/ai/new_message ai@new_message"
]

21
JSON/AnP.settings.json
View File

@ -58,20 +58,25 @@
"default_routes_files" : "/JSON/AnP.routes.json",
"AnP_RoutesManager_end" : null,
"AnP_HTTPServer_start" : null,
"http_server_host" : "",
"http_server_port" : 18000,
"AnP_HTTPServer_end" : null,
"AnP_WebSocketsServerManager_start" : null,
"default_web_sockets_server" : {
"AnP_WebSocketsServersManager_start" : null,
"default_web_sockets_servers2" : {
"anp" : {
"type" : "WebSocketServerDriver",
"host" : "localhost",
"port" : 18765
}
},
"AnP_WebSocketsServerManager_end" : null,
"AnP_WebSocketsServersManager_end" : null,
"AnP_HTTPServersManager_start" : null,
"default_http_servers" : {
"anp" : {
"type" : "HTTPDriver",
"host" : "",
"port" : 18000
}
},
"AnP_HTTPServersManager_end" : null,
"AnP_TitlesManager_start" : null,
"default_titles_files" : [

42
Python/Abstracts/AIInterpretersAbstract.py Normal file
View File

@ -0,0 +1,42 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Any, Self, Sequence
from Interfaces.Application.AnPInterface import AnPInterface
from Utils.Checks import Check
class AIInterpretersAbstract:
def __init__(self:Self, anp:AnPInterface, inputs:str|dict[str, Any|None]|Sequence[Any|None]) -> None:
if Check.is_string(inputs):
inputs = {"url" : inputs.strip()}
self.anp:AnPInterface = anp
self.url:str = self.anp.settings.get(("ai_interpreter_url", "ai_url", "url"), inputs, "")
self.maximum_tokens_per_session:int = self.anp.settings.get((
"ai_interpreter_maximum_tokens_per_session", "ai_maximum_tokens_per_session", "maximum_tokens_per_session"
), inputs, 2048)
self.sessions:dict[int, list[int]] = {}
self.sessions_i:int = 0
def start(self:Self) -> None:
pass
def close(self:Self) -> None:
self.sessions = {}
def get_session(self:Self, id:int|None = None) -> tuple[int, list[int]]:
if id is None or id not in self.sessions:
id = self.sessions_i
self.sessions_i += 1
self.sessions[id] = []
return id, self.sessions[id]
def close_session(self:Self, id:int) -> bool:
if id in self.sessions:
del self.sessions[id]
return True
return False

3
Python/Abstracts/HTTPServersAbstract.py
View File

@ -11,7 +11,7 @@ class HTTPServersAbstract(ABC):
DEFAULT_PORT:int = 8000
DEFAULT_HOST:str = ""
def __init__(self:Self, anp:AnPInterface, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
def __init__(self:Self, anp:AnPInterface, key:str, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
self.anp:AnPInterface = anp
self.__inputs:dict[str, Any|None] = Common.get_dictionary(inputs)
self.host:str = self.DEFAULT_PORT
@ -20,6 +20,7 @@ class HTTPServersAbstract(ABC):
"port": self.port,
"host": self.host
}
self.key:str = key
self.update()

16
Python/Drivers/HTTPDriver.py
View File

@ -3,13 +3,13 @@
from typing import Any, Optional, Self, Sequence
from Interfaces.Application.AnPInterface import AnPInterface
from Abstracts.ModelAbstract import ModelAbstract
from Models.RequestModel import RequestModel
from Utils.Common import Common
from threading import Thread
from http.server import BaseHTTPRequestHandler, HTTPServer
from Abstracts.HTTPServersAbstract import HTTPServersAbstract
class HTTPDriver(HTTPServersAbstract):
class HTTPDriver(HTTPServersAbstract, ModelAbstract):
class HTTPRequestHandler(BaseHTTPRequestHandler):
@ -24,7 +24,7 @@ class HTTPDriver(HTTPServersAbstract):
request.client_host = self.client_address[0]
request.client_port = self.client_address[1]
anp.routes.go(method, self.path, request)
anp.routes.go([self.server.key], method, self.path, request)
self.send_response(request.response_code)
self.send_header("Content-Type", request.response_mime)
@ -88,17 +88,18 @@ class HTTPDriver(HTTPServersAbstract):
def __init__(self:Self,
anp:AnPInterface,
key:str,
inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None
) -> None:
self.anp:AnPInterface = anp
self.__inputs:dict[str, Any|None] = Common.get_dictionary(inputs)
super().__init__(inputs)
self.__server:HTTPServer|None = None
self.__thread:Thread|None = None
super().__init__(anp, key, inputs)
self.update()
anp.settings.get(("http_server_autostart", "autostart"), inputs, True) and self.start()
def start(self:Self) -> None:
if self.__server is None:
self.__thread = Thread(target = self.__run_service)
@ -110,6 +111,7 @@ class HTTPDriver(HTTPServersAbstract):
self.__server = None
def __run_service(self:Self) -> None:
self.__server = HTTPServer((self.__host, self.__port), self.HTTPRequestHandler)
self.__server = HTTPServer((self.host, self.port), self.HTTPRequestHandler)
self.__server.anp = self.anp
self.__server.key = self.key
self.__server.serve_forever()

43
Python/Managers/AIInterpretersManager.py Normal file
View File

@ -0,0 +1,43 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Any, Self, Optional
from Interfaces.Application.AnPInterface import AnPInterface
from Utils.Common import Common
class AIInterpretersManager:
def __init__(self:Self, anp:AnPInterface) -> None:
self.anp:AnPInterface = anp
self.__interpreters:dict[str, Any] = {}
def update(self:Self) -> None:
pass
def reset(self:Self) -> None:
self.__interpreters = {}
self.update()
def close(self:Self) -> None:
self.__interpreters = {}
def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None:
subinputs:dict[str, Any|None]
for subinputs in Common.load_json(inputs, True):
key:str
value:Any|None
for key, value in subinputs.items():
if overwrite or key not in self.__interpreters:
self.__interpreters[key] = value
def remove(self:Self, names:Optional[str|list[str]] = None) -> None:
name:str
for name in Common.get_keys(names) if names else list(self.__interpreters.keys()):
if name in self.__interpreters:
del self.__interpreters[name]

4
Python/Managers/HTTPServersManager.py
View File

@ -19,7 +19,7 @@ class HTTPServersManager:
key:str
for key in ("default_http_server_files", "http_server_files", "default_http_server", "http_server"):
for key in ("default_http_servers_files", "http_servers_files", "default_http_servers", "http_servers"):
self.add(self.anp.settings.get(key), True)
def reset(self:Self) -> None:
@ -60,7 +60,7 @@ class HTTPServersManager:
Class:type[HTTPServersAbstract] = self.anp.models.get(HTTPServersAbstract, _type)
if Class and issubclass(Class, HTTPServersAbstract):
self.__servers[key] = Class(self.anp, value)
self.__servers[key] = Class(self.anp, key, value)
def remove(self:Self, names:str|Sequence[str]) -> None:
for name in Common.get_key(names):

2
Python/Managers/IndexesManager.py
View File

@ -10,7 +10,7 @@ class IndexesManager:
def __init__(self:Self, anp:AnPInterface) -> None:
self.anp:AnPInterface = anp
self.__indexes:list[str] = []
self.__indexes:list[str] = [""]
self.update()

26
Python/Managers/PseudoLoRAsManager.py Normal file
View File

@ -0,0 +1,26 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self, Any, Sequence
from Interfaces.Application.AnPInterface import AnPInterface
from Models.PseudoLoRAModel import PseudoLoRAModel
from Utils.Common import Common
from Utils.Checks import Check
class PseudoLoRAsManager:
def __init__(self:Self, anp:AnPInterface) -> None:
self.anp:AnPInterface = anp
self.__memory_cached:int = 0
self.__loras:list[PseudoLoRAModel] = []
def add(self:Self, inputs:Any|None) -> None:
if isinstance(inputs, PseudoLoRAModel):
self.__loras.append(inputs)
else:
subinputs:dict[str, Any|None]|Sequence[Any|None]
for subinputs in Common.load_json(inputs, False):
if Check.is_array(inputs):
pass

4
Python/Managers/RoutesManager.py
View File

@ -31,12 +31,12 @@ class RoutesManager:
self.update()
def go(self:Self, method:str, path:str, request:RequestModel) -> None:
def go(self:Self, key:str, method:str, path:str, request:RequestModel) -> None:
route:RouteModel
for route in self.__routes:
if route.match(method, path, request):
if route.match(key, method, path, request):
if not request.response:
if route.callback:
route.callback(request)

4
Python/Managers/WebSocketServersManager.py
View File

@ -11,7 +11,7 @@ from Utils.Common import Common
class WebSocketServersManager:
def __init__(self:Self, anp:AnPInterface, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
def __init__(self:Self, anp:Any, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
self.anp:AnPInterface = anp
self.__web_sockets:dict[str, WebSocketServersAbstract] = {}
@ -28,7 +28,7 @@ class WebSocketServersManager:
key:str
for key in ("default_web_sockets_server_files", "web_sockets_server_files", "default_web_sockets_server", "web_sockets_server"):
for key in ("default_web_sockets_servers_files", "web_sockets_servers_files", "default_web_sockets_servers", "web_sockets_servers"):
self.add(self.anp.settings.get(key), True)
def reset(self:Self) -> None:

24
Python/Models/PseudoLoRAModel.py Normal file
View File

@ -0,0 +1,24 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Self, Any, Sequence, Optional
from Utils.Checks import Check
from Utils.Common import Common
class PseudoLoRAModel:
def __init__(self:Self,
title:str,
content:str|dict[str, str|Sequence[str, str|Sequence[Any], Optional[str|Sequence[str]], bool]],
keys:Optional[str|Sequence[str]] = None,
cacheable:bool = False
) -> None:
self.title:str = title
self.path:str|None = content if Check.is_string(content) else None
self.memory:int = 0
self.cache:str = ""
self.i:int = 0
self.keys:list[str] = Common.get_keys(keys)
self.cacheable:bool = cacheable
self.nested:list[PseudoLoRAModel] = [
PseudoLoRAModel(subtitle, subcontent, subkeys, subcacheable) for subtitle, subcontent, subkeys, subcacheable in content
] if Check.is_array(content) else []

33
Python/Models/RouteModel.py
View File

@ -23,6 +23,8 @@ class RouteModel(RouteAbstract):
self.permissions:list[str] = []
self.variables:list[str] = []
self.error:int = 0
self.keys:list[str] = []
self.has_keys:bool = False
if Check.is_string(inputs):
@ -32,13 +34,16 @@ class RouteModel(RouteAbstract):
permissions:str|None
method:str|None
raw_keys:str|None
method, request, self.action, self.controller, self.path, permissions = matches.groups()
raw_keys, method, request, self.action, self.controller, self.path, permissions = matches.groups()
self.method = "get" if method is None else method.lower()
if permissions is not None and (permissions := permissions.strip()) != "":
self.permissions.extend(permissions.split(","))
raw_keys and self.keys.extend(raw_keys.split(","))
elif Check.is_dictionary(inputs):
preaction:str|Callable[[RequestModel], None]|None = Common.get_value("action", inputs)
@ -48,6 +53,7 @@ class RouteModel(RouteAbstract):
self.controller = Common.get_value("controller", inputs)
self.path = Common.get_value("path", inputs)
self.permissions.extend(Common.get_value("permissions", inputs, []))
self.keys.extend(Common.get_value("keys", inputs, []))
if Check.is_function(preaction):
self.callback = preaction
@ -58,13 +64,13 @@ class RouteModel(RouteAbstract):
l:int = len(inputs)
preaction:str|Callable[[RequestModel], None]|None
i:int = 3
i:int = 4
if l < 3:
if l < 4:
self.error = 1 << 1
return
self.method, request, preaction = inputs[:3]
self.keys, self.method, request, preaction = inputs[:4]
while l > i:
if Check.is_key(inputs[i]):
self.controller = inputs[i]
@ -103,9 +109,24 @@ class RouteModel(RouteAbstract):
else:
self.error = 1 << 3
def match(self:Self, method:str, path:str, request:RequestModel) -> bool:
self.has_keys = len(self.keys) > 0
if self.method == method.lower():
def match(self:Self, keys:list[str], method:str, path:str, request:RequestModel) -> bool:
is_in_keys:bool = True
if self.has_keys:
key:str
is_in_keys = False
for key in keys:
if key in self.keys:
is_in_keys = True
break
if is_in_keys and self.method == method.lower():
matches:REMatch = self.request.match(path)

11
Python/Utils/Common.py
View File

@ -14,7 +14,7 @@ from Utils.Checks import Check
from Utils.Patterns import RE
ROOT_PATH:str = absolute_path(directory_name(__file__))
SLASH:str = "/" if "/" in ROOT_PATH else "\\"
SLASH:str = "/" if "/" in ROOT_PATH else "\\\\"
class Common:
@ -174,6 +174,7 @@ class Common:
try:
file:FileIO
absolute_path:str = cls.get_absolute_path(path)
if mode == "r":
for format in (
@ -181,15 +182,15 @@ class Common:
"utf16", "utf16le", "utf16be", "utf8sig", "iso8859_15"
):
try:
with open(cls.get_absolute_path(path), mode, encoding = format) as file:
with open(absolute_path, mode, encoding = format) as file:
return file.read()
except Exception as exception:
pass
elif mode == "rb":
with open(cls.get_absolute_path(path), mode) as file:
with open(absolute_path, mode) as file:
return file.read()
except Exception as exception:
pass
exception
return None
@classmethod
@ -216,7 +217,7 @@ class Common:
try:
json.extend(cls.load_json(json_decode(cls.load_file(data, "r")), only_dictionaries))
except Exception as exception:
pass
pass
elif isinstance(data, (list, tuple)):
if only_dictionaries:

2
Python/Utils/Patterns.py
View File

@ -9,7 +9,7 @@ class RE:
STRING_VARIABLES:REPattern = re_compile(r"\{([a-z_][a-z0-9_]*)\}", RE_IGNORECASE)
SLASHES:REPattern = re_compile(r"[\\\/]+")
NEW_LINES:REPattern = re_compile(r"\r\n|\r|\n")
ROUTE:REPattern = re_compile(r"^(?:([a-z]+)\:)?([^\s]+)\s+(?:([^\s\@]+)\@([^\s]+)|([^\s]+))(?:\s+(.+))?$", RE_IGNORECASE)
ROUTE:REPattern = re_compile(r"^(?:\[((?:[a-z_][a-z0-9_]+,?)+)\])?(?:([a-z]+)\:)?([^\s]+)\s+(?:([^\s\@]+)\@([^\s]+)|([^\s]+))(?:\s+(.+))?$", RE_IGNORECASE)
TO_REGULAR_EXPRESSION:REPattern = re_compile(r'[\(\)\{\}\/\\\.\-\+\*\^\$\?\|\!\<\>\r\n\t]')
ROUTE_KEY:REPattern = re_compile(r'\\\{([a-z_][a-z0-9_]*)\\\}', RE_IGNORECASE)
EXCEPTION:REPattern = re_compile(r'^\s*File "([^"]+)", line ([0-9]+), in ([^\n]+)(.*|[\r\n]*)*$')

2
Python/run.py
View File

@ -5,11 +5,13 @@ from typing import Any
from Application.AnP import AnP
from Controllers.AIController import AIController
from Drivers.WebSocketServerDriver import WebSocketServerDriver
from Drivers.HTTPDriver import HTTPDriver
inputs:dict[str, dict[str, Any|None]] = {
"default_models" : {
"AIController" : AIController,
"WebSocketServerDriver" : WebSocketServerDriver,
"HTTPDriver" : HTTPDriver
}
}