AURUMVORAX/PyG2O
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Добавлена авторизация через UDP пакеты

Browse Source
This commit is contained in:
AURUMVORXX
2025-08-27 13:55:15 +05:00
parent 77bfd7492a
commit 7447973428
18 changed files with 249 additions and 3645 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
include/client/main.nut Normal file
View File

@@ -0,0 +1,9 @@
CLIENT_PASSWORD <- "";
addEventHandler("onPacket", function(data){
local id = data.readUInt8();
if (id == 250)
{
CLIENT_PASSWORD = data.readString();
}
})

10
include/events.nut → include/server/events.nut
View File

@@ -450,9 +450,17 @@ addEventHandler("onPlayerEquipSpell", function(playerid, slotId, instance)
addEventHandler("onPlayerJoin", function(playerid) addEventHandler("onPlayerJoin", function(playerid)
{ {
client_password = _globalInstance.generateClientPassword();
local packet = Packet();
packet.writeUInt8(250);
packet.writeString(client_password);
packet.send(playerid, RELIABLE);
local data = { local data = {
event = "onPlayerJoin", event = "onPlayerJoin",
playerid = playerid playerid = playerid,
password = client_password
} }
if (_globalInstance != -1) if (_globalInstance != -1)

14
include/main.nut → include/server/main.nut
View File

@@ -133,4 +133,18 @@ class PyG2O
_message_call.bindenv(this)(request["data"]); _message_call.bindenv(this)(request["data"]);
} }
function generateClientPassword()
{
local chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!@#$%^&*()";
local result = "";
local length = 32;
for (local i = 0; i < length; i++) {
local randomIndex = rand() % chars.len();
result += chars.slice(randomIndex, randomIndex + 1);
}
return result;
}
} }

0
include/messages.nut → include/server/messages.nut
View File

319
src/pyg2o/__init__.py
View File

@@ -1,158 +1,4 @@
from .server import PythonWebsocketServer
from .functions.chat import sendMessageToAll
from .functions.chat import sendMessageToPlayer
from .functions.chat import sendPlayerMessageToAll
from .functions.chat import sendPlayerMessageToPlayer
from .functions.math import getDistance2d
from .functions.math import getDistance3d
from .functions.math import getVectorAngle
from .functions.game import getHostname
from .functions.game import getMaxSlots
from .functions.game import getServerPublic
from .functions.game import getPlayersCount
from .functions.game import exit
from .functions.game import getDayLength
from .functions.game import getServerDescription
from .functions.game import getServerWorld
from .functions.game import getTime
from .functions.game import serverLog
from .functions.game import setDayLength
from .functions.game import setServerDescription
from .functions.game import setServerWorld
from .functions.game import setServerPublic
from .functions.game import setTime
from .functions.npc import clearNpcActions
from .functions.npc import createNpc
from .functions.npc import destroyNpc
from .functions.npc import getNpcAction
from .functions.npc import getNpcActions
from .functions.npc import getNpcActionsCount
from .functions.npc import getNpcHostPlayer
from .functions.npc import getNpcLastActionId
from .functions.npc import isNpc
from .functions.npc import isNpcActionFinished
from .functions.npc import npcAttackMelee
from .functions.npc import npcAttackRanged
from .functions.npc import npcSpellCast
from .functions.npc import npcUseClosestMob
from .functions.npc import setNpcHostPlayer
from .functions.player import addBan
from .functions.player import applyPlayerOverlay
from .functions.player import ban
from .functions.player import drawWeapon
from .functions.player import equipItem
from .functions.player import getPlayerAmulet
from .functions.player import getPlayerAngle
from .functions.player import getPlayerAni
from .functions.player import getPlayerOverlays
from .functions.player import getPlayerArmor
from .functions.player import getPlayerAtVector
from .functions.player import getPlayerBelt
from .functions.player import getPlayerCameraPosition
from .functions.player import getPlayerCollision
from .functions.player import getPlayerColor
from .functions.player import getPlayerContext
from .functions.player import getPlayerDexterity
from .functions.player import getPlayerFaceAnis
from .functions.player import getPlayerFatness
from .functions.player import getPlayerFocus
from .functions.player import getPlayerHealth
from .functions.player import getPlayerHelmet
from .functions.player import getPlayerIP
from .functions.player import getPlayerInstance
from .functions.player import getPlayerInvisible
from .functions.player import getPlayerMacAddr
from .functions.player import getPlayerMana
from .functions.player import getPlayerMaxHealth
from .functions.player import getPlayerMaxMana
from .functions.player import getPlayerMeleeWeapon
from .functions.player import getPlayerName
from .functions.player import getPlayerPing
from .functions.player import getPlayerPosition
from .functions.player import getPlayerRangedWeapon
from .functions.player import getPlayerRespawnTime
from .functions.player import getPlayerRing
from .functions.player import getPlayerScale
from .functions.player import getPlayerSerial
from .functions.player import getPlayerShield
from .functions.player import getPlayerSkillWeapon
from .functions.player import getPlayerSpell
from .functions.player import getPlayerStrength
from .functions.player import getPlayerTalent
from .functions.player import getPlayerVirtualWorld
from .functions.player import getPlayerVisual
from .functions.player import getPlayerWeaponMode
from .functions.player import getPlayerWorld
from .functions.player import getPlayerUID
from .functions.player import giveItem
from .functions.player import hitPlayer
from .functions.player import isPlayerConnected
from .functions.player import isPlayerDead
from .functions.player import isPlayerSpawned
from .functions.player import isPlayerUnconscious
from .functions.player import kick
from .functions.player import playAni
from .functions.player import playFaceAni
from .functions.player import fadeOutAni
from .functions.player import readySpell
from .functions.player import removeItem
from .functions.player import removePlayerOverlay
from .functions.player import removeWeapon
from .functions.player import setPlayerAngle
from .functions.player import setPlayerCollision
from .functions.player import setPlayerColor
from .functions.player import setPlayerContext
from .functions.player import setPlayerDexterity
from .functions.player import setPlayerFatness
from .functions.player import setPlayerHealth
from .functions.player import setPlayerInstance
from .functions.player import setPlayerInvisible
from .functions.player import setPlayerMana
from .functions.player import setPlayerMaxHealth
from .functions.player import setPlayerMaxMana
from .functions.player import setPlayerName
from .functions.player import setPlayerPosition
from .functions.player import setPlayerRespawnTime
from .functions.player import setPlayerScale
from .functions.player import setPlayerSkillWeapon
from .functions.player import setPlayerStrength
from .functions.player import setPlayerTalent
from .functions.player import setPlayerVirtualWorld
from .functions.player import setPlayerVisual
from .functions.player import setPlayerWeaponMode
from .functions.player import setPlayerWorld
from .functions.player import spawnPlayer
from .functions.player import stopAni
from .functions.player import stopFaceAni
from .functions.player import unequipItem
from .functions.player import unreadySpell
from .functions.player import unspawnPlayer
from .functions.player import useItem
from .functions.player import useItemToState
from .functions.streamer import findNearbyPlayers
from .functions.streamer import getStreamedPlayersByPlayer
from .functions.streamer import getSpawnedPlayersForPlayer
from .functions.waypoint import getNearestWaypoint
from .functions.waypoint import getWaypoint
from .functions.event import addEvent
from .functions.event import callEvent
from .functions.event import event
from .functions.event import removeEventHandler
from .functions.event import toggleEvent
from .functions.event import removeEvent
from .functions.pyg2o import call_squirrel_function
from .functions.pyg2o import execute_squirrel_code
from .constants import Constant from .constants import Constant
from .classes.daedalus import Daedalus from .classes.daedalus import Daedalus
@@ -161,168 +7,3 @@ from .classes.items import ItemGround
from .classes.items import ItemsGround from .classes.items import ItemsGround
from .classes.mds import Mds from .classes.mds import Mds
from .classes.sky import Sky from .classes.sky import Sky
__all__ = [
"PythonWebsocketServer",
"sendMessageToAll",
"sendMessageToPlayer",
"sendPlayerMessageToAll",
"sendPlayerMessageToPlayer",
"getDistance2d",
"getDistance3d",
"getVectorAngle",
"getHostname",
"getMaxSlots",
"getServerPublic",
"getPlayersCount",
"exit",
"getDayLength",
"getServerDescription",
"getServerWorld",
"getTime",
"serverLog",
"setDayLength",
"setServerDescription",
"setServerWorld",
"setServerPublic",
"setTime",
"clearNpcActions",
"createNpc",
"destroyNpc",
"getNpcAction",
"getNpcActions",
"getNpcActionsCount",
"getNpcHostPlayer",
"getNpcLastActionId",
"isNpc",
"isNpcActionFinished",
"npcAttackMelee",
"npcAttackRanged",
"npcSpellCast",
"npcUseClosestMob",
"setNpcHostPlayer",
"addBan",
"applyPlayerOverlay",
"ban",
"drawWeapon",
"equipItem",
"getPlayerAmulet",
"getPlayerAngle",
"getPlayerAni",
"getPlayerOverlays",
"getPlayerArmor",
"getPlayerAtVector",
"getPlayerBelt",
"getPlayerCameraPosition",
"getPlayerCollision",
"getPlayerColor",
"getPlayerContext",
"getPlayerDexterity",
"getPlayerFaceAnis",
"getPlayerFatness",
"getPlayerFocus",
"getPlayerHealth",
"getPlayerHelmet",
"getPlayerIP",
"getPlayerInstance",
"getPlayerInvisible",
"getPlayerMacAddr",
"getPlayerMana",
"getPlayerMaxHealth",
"getPlayerMaxMana",
"getPlayerMeleeWeapon",
"getPlayerName",
"getPlayerPing",
"getPlayerPosition",
"getPlayerRangedWeapon",
"getPlayerRespawnTime",
"getPlayerRing",
"getPlayerScale",
"getPlayerSerial",
"getPlayerShield",
"getPlayerSkillWeapon",
"getPlayerSpell",
"getPlayerStrength",
"getPlayerTalent",
"getPlayerVirtualWorld",
"getPlayerVisual",
"getPlayerWeaponMode",
"getPlayerWorld",
"getPlayerUID",
"giveItem",
"hitPlayer",
"isPlayerConnected",
"isPlayerDead",
"isPlayerSpawned",
"isPlayerUnconscious",
"kick",
"playAni",
"playFaceAni",
"fadeOutAni",
"readySpell",
"removeItem",
"removePlayerOverlay",
"removeWeapon",
"setPlayerAngle",
"setPlayerCollision",
"setPlayerColor",
"setPlayerContext",
"setPlayerDexterity",
"setPlayerFatness",
"setPlayerHealth",
"setPlayerInstance",
"setPlayerInvisible",
"setPlayerMana",
"setPlayerMaxHealth",
"setPlayerMaxMana",
"setPlayerName",
"setPlayerPosition",
"setPlayerRespawnTime",
"setPlayerScale",
"setPlayerSkillWeapon",
"setPlayerStrength",
"setPlayerTalent",
"setPlayerVirtualWorld",
"setPlayerVisual",
"setPlayerWeaponMode",
"setPlayerWorld",
"spawnPlayer",
"stopAni",
"stopFaceAni",
"unequipItem",
"unreadySpell",
"unspawnPlayer",
"useItem",
"useItemToState",
"findNearbyPlayers",
"getStreamedPlayersByPlayer",
"getSpawnedPlayersForPlayer",
"getNearestWaypoint",
"getWaypoint",
"addEvent",
"callEvent",
"event",
"removeEventHandler",
"toggleEvent",
"removeEvent",
"call_squirrel_function",
"execute_squirrel_code",
"Constant",
"Daedalus",
"DamageDescription",
"ItemGround",
"ItemsGround",
"Mds",
"Sky",
]

13
src/pyg2o/call_repr.py
View File

@@ -1,13 +0,0 @@
import inspect
def get_call_repr():
frame = inspect.currentframe().f_back
func_name = frame.f_code.co_name
args_info = inspect.getargvalues(frame)
args_str = []
for arg in args_info.args:
val = args_info.locals[arg]
args_str.append(f"{val!r}" if not isinstance(val, str) else f"'{val}'")
return f"{func_name}({', '.join(args_str)})"

0
src/pyg2o/event.py Normal file
View File

98
src/pyg2o/functions/chat.py
View File

@@ -1,98 +0,0 @@
from ..server import PythonWebsocketServer
from ..call_repr import get_call_repr
async def sendMessageToAll(r : int, g : int, b : int, text : str):
"""
This function will send a chat message to every connected player.
Sending a message triggers client side event [onPlayerMessage](../../defaultEvents/player/onPlayerMessage.md) with playerid set as `-1`.
Original: [sendMessageToAll](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/chat/sendMessageToAll/)
## Declaration
```python
async def sendMessageToAll(r : int, g : int, b : int, text : str)
```
## Parameters
* `int` **r**: the red color component in RGB model.
* `int` **g**: the green color component in RGB model.
* `int` **b**: the blue color component in RGB model.
* `str` **text**: that will be send.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def sendMessageToPlayer(playerid : int, r : int, g : int, b : int, text : str):
"""
This function will send a chat message to specific player.
Sending a message triggers client side event [onPlayerMessage](../../defaultEvents/player/onPlayerMessage.md) with playerid set as `-1`.
Original: [sendMessageToPlayer](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/chat/sendMessageToPlayer/)
## Declaration
```python
async def sendMessageToPlayer(playerid : int, r : int, g : int, b : int, text : str)
```
## Parameters
* `int` **playerid**: the id of the player which will receive a message.
* `int` **r**: the red color component in RGB model.
* `int` **g**: the green color component in RGB model.
* `int` **b**: the blue color component in RGB model.
* `str` **text**: that will be send.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def sendPlayerMessageToAll(senderid : int, r : int, g : int, b : int, text : str):
"""
This function will send a chat message from one player to every player. Sending a message
Sending a message triggers client side event [onPlayerMessage](../../defaultEvents/player/onPlayerMessage.md) with playerid set as **senderid**.
Original: [sendPlayerMessageToAll](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/chat/sendPlayerMessageToAll/)
## Declaration
```python
async def sendPlayerMessageToAll(senderid : int, r : int, g : int, b : int, text : str)
```
## Parameters
* `int` **senderid**: the id of the player which will send a message.
* `int` **r**: the red color component in RGB model.
* `int` **g**: the green color component in RGB model.
* `int` **b**: the blue color component in RGB model.
* `str` **text**: that will be send.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def sendPlayerMessageToPlayer(senderid : int, receiverid : int, r : int, g : int, b : int, text : str):
"""
This function will send a chat message from one player to another player.
Sending a message triggers client side event [onPlayerMessage](../../defaultEvents/player/onPlayerMessage.md) with playerid set as **senderid**.
Original: [sendPlayerMessageToPlayer](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/chat/sendPlayerMessageToPlayer/)
## Declaration
```python
async def sendPlayerMessageToPlayer(senderid : int, receiverid : int, r : int, g : int, b : int, text : str)
```
## Parameters
* `int` **senderid**: the id of the player which will send a message.
* `int` **receiverid**: the id of the player which will receive a message.
* `int` **r**: the red color component in RGB model.
* `int` **g**: the green color component in RGB model.
* `int` **b**: the blue color component in RGB model.
* `str` **text**: that will be send.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result

230
src/pyg2o/functions/event.py
View File

@@ -1,230 +0,0 @@
from functools import wraps
eventList = {}
disabledEventList = []
async def callEvent(evtName : str, **kwargs : list):
"""
This function will notify (call) every handler bound to specified event.
Original: [callEvent](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/event/callEvent/)
## Declaration
```python
def callEvent(evtName : str, **kwargs : list)
```
## Parameters
* `str` **name**: the name of the event
* `**dict` **kwargs**: the variable number of arguments.
## Usage
```python
import g2o
g2o.addEvent('testEvt')
@g2o.event('testEvt')
def onTestEvent(**kwargs):
print(f'{kwargs['name']} called my beautiful test event')
g2o.callEvent('testEvt', name = 'Diego')
```
"""
if evtName in eventList and evtName not in disabledEventList:
for event in eventList[evtName]:
event['function'].eventName = evtName
result = await event['function'](**kwargs)
def addEvent(name : str):
"""
This function will register a new event with specified name.
Events can be used to notify function(s) when something will happen, like player joins the server, etc.
Original: [addEvent](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/event/addEvent/)
## Declaration
```python
def addEvent(name)
```
## Parameters
* `str` **name**: the name of the event
## Usage
```python
import g2o
g2o.addEvent('testEvt')
```
"""
if name not in eventList:
eventList[name] = []
def event(event_name: str, priority: int = 9999) -> None:
def inlineEvt(func):
if event_name not in eventList:
addEvent(event_name)
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
eventList[event_name].append({'function': wrapper, 'priority': priority})
eventList[event_name].sort(key = lambda x: x['priority'])
return wrapper
return inlineEvt
def removeEventHandler(name : str, func : object):
"""
This function will unbind function from specified event.
Original: [removeEventHandler](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/event/removeEventHandler/)
## Declaration
```python
def removeEventHandler(name : str, func : object)
```
## Parameters
* `str` **name**: the name of the event
* `object` **func**: the reference to a function which is currently bound to specified event.
## Usage
```python
import g2o
@g2o.event('onTime')
def onTimeEvt(**kwargs):
print('Calling only once')
g2o.removeEventHandler('onTime', onTimeEvt)
```
"""
if not name in eventList:
pass
for index, item in enumerate(eventList[name]):
if item['function'] == func:
del eventList[name][index]
def toggleEvent(name : str, toggle : bool):
'''
!!! note
By default every event is toggled `on` (enabled).
This function will toggle event (enable or disable it globally). By toggling event off, you can completely disable certain event from calling it's handlers.
Original: [toggleEvent](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/event/toggleEvent/)
## Declaration
```python
def toggleEvent(name : str, toggle : bool)
```
## Parameters
* `str` **name**: the name of the event
* `bool` **toggle**: `false` if you want to disable the event, otherwise true.
## Usage
```python
import g2o
@g2o.event('onTime')
def onTimeEvt(**kwargs):
print('Calling only once')
g2o.toggleEvent('onTime', false)
```
'''
if not toggle and name not in disabledEventList:
disabledEventList.append(name)
elif toggle and name in disabledEventList:
disabledEventList.remove(name)
def removeEvent(name : str):
'''
!!! warning
Removing an event also cause all event handlers to unregister.
This function will unregister an event with specified name.
Original: [removeEvent](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/event/removeEvent/)
## Declaration
```python
def removeEvent(name : str)
```
## Parameters
* `str` **name**: the name of the event
## Usage
```python
import g2o
@g2o.event('onTime')
def onTimeEvt(**kwargs):
print('Calling only once')
g2o.removeEvent('onTime')
```
'''
if name in eventList:
eventList.pop(name)
## registering default events
addEvent('onInit')
addEvent('onExit')
addEvent('onTick')
addEvent('onTime')
addEvent('onBan')
addEvent('onUnban')
addEvent('onPlayerChangeColor')
addEvent('onPlayerChangeFocus')
addEvent('onPlayerChangeHealth')
addEvent('onPlayerChangeMana')
addEvent('onPlayerChangeMaxHealth')
addEvent('onPlayerChangeMaxMana')
addEvent('onPlayerChangeWeaponMode')
addEvent('onPlayerChangeWorld')
addEvent('onPlayerCommand')
addEvent('onPlayerDamage')
addEvent('onPlayerDead')
addEvent('onPlayerDisconnect')
addEvent('onPlayerDropItem')
addEvent('onPlayerEnterWorld')
addEvent('onPlayerJoin')
addEvent('onPlayerMessage')
addEvent('onPlayerMobInteract')
addEvent('onPlayerRespawn')
addEvent('onPlayerShoot')
addEvent('onPlayerSpellCast')
addEvent('onPlayerSpellSetup')
addEvent('onPlayerTakeItem')
addEvent('onPlayerTeleport')
addEvent('onPlayerToggleFaceAni')
addEvent('onPlayerEquipAmulet')
addEvent('onPlayerEquipArmor')
addEvent('onPlayerEquipBelt')
addEvent('onPlayerEquipHandItem')
addEvent('onPlayerEquipHelmet')
addEvent('onPlayerEquipMeleeWeapon')
addEvent('onPlayerEquipRangedWeapon')
addEvent('onPlayerEquipRing')
addEvent('onPlayerEquipShield')
addEvent('onPlayerEquipSpell')
addEvent('onPlayerSpawnForPlayer')
addEvent('onPlayerUnspawnForPlayer')
addEvent('onPacket')
addEvent('onPlayerUseCheat')
addEvent('onNpcActionFinished')
addEvent('onNpcActionSent')
addEvent('onNpcChangeHostPlayer')
addEvent('onNpcCreated')
addEvent('onNpcDestroyed')
addEvent('onWebsocketConnect')
addEvent('onWebsocketDisconnect')

320
src/pyg2o/functions/game.py
View File

@@ -1,320 +0,0 @@
from ..server import PythonWebsocketServer
from ..call_repr import get_call_repr
async def getHostname() -> str:
"""
This function will get the hostname of the server.
Original: [getHostname](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/game/getHostname/)
## Declaration
```python
async def getHostname() -> str
```
## Returns
`str`: Server hostname.
## Usage
```python
import g2o
@g2o.event('onInit')
def evtInit(**kwargs):
print('Server hostname:', g2o.getHostname())
```
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getOnlinePlayers():
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getMaxSlots() -> int:
"""
This function will get the max number of slots available on the server.
Original: [getMaxSlots](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/game/getMaxSlots/)
## Declaration
```python
async def getMaxSlots() -> int
```
## Returns
`int`: Max slots number on the server.
## Usage
```python
import g2o
@g2o.event('onInit')
def evtInit(**kwargs):
print('Server max slots:', g2o.getMaxSlots())
```
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getPlayersCount() -> int:
"""
This function will get the max number of slots available on the server.
Original: [getPlayersCount](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/game/getPlayersCount/)
## Declaration
```python
async def getPlayersCount() -> int
```
## Returns
`int`: Number of players on the server.
## Usage
```python
import g2o
@g2o.event('onInit')
def evtInit(**kwargs):
print('Players online:', g2o.getPlayersCount())
```
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getServerPublic() -> bool:
"""
This function will get the publicity state of the server.
## Declaration
```python
async def getServerPublic() -> bool
```
## Returns
`bool`: ``true`` if server is publicly available, otherwise ``false``
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def exit(exitCode : int = 0):
"""
This function will close the server with specified exit code.
Original: [exit](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/exit/)
## Declaration
```python
async def exit(exitCode : int = 0)
```
## Parameters
* `int` **exitCode**: exit status for g2o server.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getDayLength() -> float:
"""
The function is used to get the day length in miliseconds.
Original: [getDayLength](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/getDayLength/)
## Declaration
```python
async def getDayLength() -> float
```
## Returns
`float`: the current day length in miliseconds.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getServerDescription() -> str:
"""
This function will get the description of the server.
Original: [getServerDescription](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/getServerDescription/)
## Declaration
```python
async def getServerDescription() -> str
```
## Returns
`str`: Server description.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getServerWorld() -> str:
"""
The function is used to get the path of the default world on the server.
Original: [getServerWorld](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/getServerWorld/)
## Declaration
```python
async def getServerWorld() -> str
```
## Returns
`str`: The world path name.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getTime() -> tuple:
"""
The function is used to get the path of the default world on the server.
Original: [getTime](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/getTime/)
## Declaration
```python
async def getTime() -> tuple
```
## Returns
`tuple (day, hour, min)`: The current time in the game.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return (result['day'], result['hour'], result['min'])
async def serverLog(text : str):
"""
This function will log the text into server.log file.
Original: [serverLog](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/serverLog/)
## Declaration
```python
async def serverLog(text : str)
```
## Parameters
`str` **text**: the text message that you want to append to server.log file.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def setDayLength(miliseconds : float):
"""
!!! note
Day length can't be smaller than 10 000 miliseconds.
This function will set the day length in miliseconds.
Original: [setDayLength](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/setDayLength/)
## Declaration
```python
async def setDayLength(miliseconds : float)
```
## Parameters
`float` **miliseconds**: day length in miliseconds.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def setServerDescription(description : str):
"""
This function will set the description of the server.
Original: [setServerDescription](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/setServerDescription/)
## Declaration
```python
async def setServerDescription(description : str)
```
## Parameters
`str` **description**: the server description.
## Returns
`bool`: `true` if server description was set successfully, otherwise `false`.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def setServerWorld(world : str):
"""
!!! note
The server world limit is set to 32 characters.
!!! note
If the target world path is written with backslashes instead of normal slashes, you need to escape it with another backslashes e.g. "NEWWORLD\\NEWWORLD.ZEN".
This function will change the default world to which players will enter after joining.
Original: [setServerWorld](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/setServerWorld/)
## Declaration
```python
async def setServerWorld(world : str)
```
## Parameters
`str` **world**: the path to the target world.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def setServerPublic(public : str):
"""
This function will change the publicity state of the server.
## Declaration
```python
async def setServerPublic(public : str)
```
## Parameters
`bool` **public**: server public state.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def setTime(hour : int, min : int, day : int = 0):
"""
!!! note
This functions supports ``pass_exception: bool`` optional argument for manual handling exceptions.
This function will set the current time in the game to the given time, for all the players.
Original: [setTime](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/game/setTime/)
## Declaration
```python
async def setTime(hour : int, min : int, day : int = 0)
```
## Parameters
`int` **hour**: the hour of new time (in the range between 0-23) or subtract value from hour (hour < 0).
`int` **mins**: the minute of new time (in the range between 0-59) or subtract value from mins (mins < 0).
`int` **day**: the day of new time or subtract value from day (day < 0).
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result

91
src/pyg2o/functions/math.py
View File

@@ -1,91 +0,0 @@
from ..server import PythonWebsocketServer
from ..call_repr import get_call_repr
async def getDistance2d(x1 : float, y1: float, x2 : float, y2 : float) -> float:
"""
!!! note
This functions supports ``pass_exception: bool`` optional argument for manual handling exceptions.
This function will get the 2d distance between two points.
Original: [getDistance2d](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/math/getDistance2d/)
## Declaration
```python
async def getDistance2d(x1 : float, y1: float, x2 : float, y2 : float) -> float
```
## Parameters
* `float` **x1**: the position on X axis of the first point.
* `float` **y1**: the position on Y axis of the first point.
* `float` **x2**: the position on X axis of the second point.
* `float` **y2**: the position on Y axis of the second point.
**OR**
* `dict[str, float]` **first**: the poistion on XY axis of the first point.
* `dict[str, float]` **second**: the position of XY axis of the second point.
## Returns
`float`: Returns the calculated 2d distance between two points as floating point number.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getDistance3d(x1 : float, y1: float, z1 : float, x2 : float, y2 : float, z2 : float) -> float:
"""
!!! note
This functions supports ``pass_exception: bool`` optional argument for manual handling exceptions.
This function will get the 3d distance between two points.
Original: [getDistance3d](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/math/getDistance3d/)
## Declaration
```python
async def getDistance3d(x1 : float, y1: float, z1 : float, x2 : float, y2 : float, z2 : float) -> float
```
## Parameters
* `float` **x1**: the position on X axis of the first point.
* `float` **y1**: the position on Y axis of the first point.
* `float` **z1**: the position on Z axis of the first point.
* `float` **x2**: the position on X axis of the second point.
* `float` **y2**: the position on Y axis of the second point.
* `float` **z2**: the position on Z axis of the second point.
**OR**
* `dict[str, float]` **first**: the position on XYZ axis of the first point.
* `dict[str, float]` **second**: the position on XYZ axic of the second point.
## Returns
`float`: Returns the calculated 3d distance between two points as floating point number.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getVectorAngle(x1 : float, y1: float, x2 : float, y2 : float) -> float:
"""
!!! note
This functions supports ``pass_exception: bool`` optional argument for manual handling exceptions.
This function will get angle on Y axis directed towards the second point.
Original: [getVectorAngle](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/shared-functions/math/getVectorAngle/)
## Declaration
```python
async def getVectorAngle(x1 : float, y1: float, x2 : float, y2 : float) -> float
```
## Parameters
* `float` **x1**: the position on X axis of the first point.
* `float` **y1**: the position on Y axis of the first point.
* `float` **x2**: the position on X axis of the second point.
* `float` **y2**: the position on Y axis of the second point.
**OR**
* `dict[str, float]` **first**: the poistion on XY axis of the first point.
* `dict[str, float]` **second**: the position of XY axis of the second point.
## Returns
`float`: Returns the angle on Y axis directed towards the second point.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result

308
src/pyg2o/functions/npc.py
View File

@@ -1,308 +0,0 @@
from ..server import PythonWebsocketServer
from ..call_repr import get_call_repr
async def clearNpcActions(npc_id : int):
"""
This function clears remote NPC actions queue. Remote NPCs uses actions queue to execute thier tasks.
Original: [clearNpcActions](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/clearNpcActions/)
## Declaration
```python
async def clearNpcActions(npc_id : int)
```
## Parameters
`int` **npc_id**: the npc identifier.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def createNpc(name : str, instance : str = 'PC_HERO') -> int:
"""
!!! note
By default npcs won't be added to world. In order to do that, you have to call [spawnPlayer](../player/spawnPlayer.md).
!!! note
Remote NPC id will always begins from max slots value.
This function creates remote NPC.
Original: [createNpc](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/createNpc/)
## Declaration
```python
async def createNpc(name : str, instance : str = 'PC_HERO') -> int
```
## Parameters
`str` **name**: the displayed name of the npc.
`str` **instance**: the instance name of for the npc.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def destroyNpc(npc_id : int) -> bool:
"""
This function destroys remote NPC.
Original: [destroyNpc](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/destroyNpc/)
## Declaration
```python
async def destroyNpc(npc_id : int) -> bool
```
## Parameters
`int` **npc_id**: the identifier of npc.
## Returns
`bool`: `true` when npc was successfully destroyed, otherwise false`.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getNpcAction(npc_id : int, index : int) -> dict:
"""
This function gets information about element on specified index in NPC action queue.
Original: [getNpcAction](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/getNpcAction/)
## Declaration
```python
async def getNpcAction(npc_id : int, index : int) -> dict
```
## Parameters
`int` **npc_id**: the identifier of npc.
`int` **index**: the index of element in the queue.
## Returns
`dict {type, id, status}`: The table containing information about selected element.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getNpcActions(npc_id : int) -> list:
"""
This function gets informations about elements in NPC action queue.
Original: [getNpcActions](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/getNpcActions/)
## Declaration
```python
async def getNpcActions(npc_id : int) -> list
```
## Parameters
`int` **npc_id**: the identifier of npc.
## Returns
`list [{type, id}]`: The array containing information about queue elements.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getNpcActionsCount(npc_id : int) -> int:
"""
This function gets elements count in NPC action queue.
Original: [getNpcActionsCount](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/getNpcActionsCount/)
## Declaration
```python
async def getNpcActionsCount(npc_id : int) -> int
```
## Parameters
`int` **npc_id**: the identifier of npc.
## Returns
`int`: The count of elements inside queue, otherwise `-1`.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getNpcHostPlayer(npc_id : int) -> int:
"""
This function gets NPC host player id.
Original: [getNpcHostPlayer](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/getNpcHostPlayer/)
## Declaration
```python
async def getNpcHostPlayer(npc_id : int) -> int
```
## Parameters
`int` **npc_id**: the identifier of npc.
## Returns
`int`: the host player identifier. If there is no host player `-1` is returned instead.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getNpcLastActionId(npc_id : int) -> int:
"""
This function gets last action identifier, that was enqued to the NPC action queue. Every action in queue has associated unique id, by which can be identified.
Original: [getNpcLastActionId](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/getNpcLastActionId/)
## Declaration
```python
async def getNpcLastActionId(npc_id : int) -> int
```
## Parameters
`int` **npc_id**: the identifier of npc.
## Returns
`int`: The last finished action identifier, otherwise `-1`.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def isNpc(npc_id : int) -> bool:
"""
This function checks whether id related to given object is remote NPC.
Original: [isNpc](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/isNpc/)
## Declaration
```python
async def isNpc(npc_id : int) -> bool
```
## Parameters
`int` **npc_id**: the identifier of npc.
## Returns
`bool`: `true` when object is NPC, otherwise `false`.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def isNpcActionFinished(npc_id : int, action_id : int) -> bool:
"""
This function checks whether specified NPC action was finished.
Original: [isNpcActionFinished](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/isNpcActionFinished/)
## Declaration
```python
async def isNpcActionFinished(npc_id : int, action_id : int) -> bool
```
## Parameters
`int` **npc_id**: the identifier of npc.
`int` **action_id**: the unique action identifier.
## Returns
`bool`: `true` if specified action identifier was already finished, otherwise `false`.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def npcAttackMelee(attacker_id : int, enemy_id : int, attack_type : int, combo : int):
"""
!!! note
Combo is internal Gothic value. Its behaviour can be sometimes undefined. For example -1 value doesn't work for not humanoid NPCs.
This function enqueues attack melee action to the remote NPC action queue.
Original: [npcAttackMelee](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/npcAttackMelee/)
## Declaration
```python
async def npcAttackMelee(attacker_id : int, enemy_id : int, attack_type : int, combo : int)
```
## Parameters
`int` **attacker_id**: the remote npc id.
`int` **enemy_id**: the remote npc or player id.
`int` **attack_type**: the type of attack.
`int` **combol**: the combo sequence. For `-1` execute next command immediately.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def npcAttackRanged(attacker_id : int, enemy_id : int):
"""
This function enqueues attack ranged action to the remote NPC action queue.
Original: [npcAttackRanged](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/npcAttackRanged/)
## Declaration
```python
async def npcAttackRanged(attacker_id : int, enemy_id : int)
```
## Parameters
`int` **attacker_id**: the remote npc id.
`int` **enemy_id**: the remote npc or player id.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def npcSpellCast(attacker_id : int, enemy_id : int):
"""
This function enqueues spell cast action to the remote NPC action queue.
Original: [npcSpellCast](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/npcSpellCast/)
## Declaration
```python
async def npcSpellCast(attacker_id : int, enemy_id : int)
```
## Parameters
`int` **attacker_id**: the remote npc id.
`int` **enemy_id**: the remote npc or player id.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def npcUseClosestMob(npc_id : int, sceme : str, target_state : int):
"""
This function enqueues use closest mob action to the remote NPC action queue.
Original: [npcUseClosestMob](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/npcUseClosestMob/)
## Declaration
```python
async def npcUseClosestMob(npc_id : int, sceme : str, target_state : int)
```
## Parameters
`int` **npc_id**: the npc identifier.
`str` **sceme**: the animation sceme name, e.g: `"BENCH"` when you want to interact with bench.
`int` **target_state**: the target state, use `1` if you want to start interaction and `-1` to end it.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def setNpcHostPlayer(npc_id : int, host_id : int) -> bool:
"""
This function sets new NPC host player.
Original: [setNpcHostPlayer](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/npc/setNpcHostPlayer/)
## Declaration
```python
async def setNpcHostPlayer(npc_id : int, host_id : int) -> bool
```
## Parameters
`int` **npc_id**: the npc identifier.
`int` **host_id**: the player host identifier.
## Returns
`bool`: `true` if host was successfully changed, otherwise `false`.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result

1939
src/pyg2o/functions/player.py
View File

File diff suppressed because it is too large Load Diff

16
src/pyg2o/functions/pyg2o.py
View File

@@ -1,16 +0,0 @@
from ..server import PythonWebsocketServer
async def call_squirrel_function(function_name: str, *args):
args_str = map(str, args)
arg_list = ', '.join(args_str)
data = f'return {function_name}({arg_list})'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def execute_squirrel_code(data: str):
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result

65
src/pyg2o/functions/streamer.py
View File

@@ -1,65 +0,0 @@
from ..server import PythonWebsocketServer
from ..call_repr import get_call_repr
async def findNearbyPlayers(position : dict, radius : int, world : str, virtual_world : int = 0) -> list:
"""
This function will search for nearest players, that matches given query arguments.
Original: [findNearbyPlayers](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/streamer/findNearbyPlayers/)
## Declaration
```python
async def findNearbyPlayers(position : dict, radius : int, world : str, virtual_world : int = 0) -> list
```
## Parameters
`dict {x, y, z}` **position**: the centroid position.
`int` **radius**: the maximum radius to search from centroid.
`str` **world**: the world used to find players.
`int` **virtual_world**: the virtual world used to find players.
## Returns
`list [int]`: ids of nearby players.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getSpawnedPlayersForPlayer(id : int) -> list:
"""
This function is used to retrieve currently spawned players for given player.
Original: [getSpawnedPlayersForPlayer](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/streamer/getSpawnedPlayersForPlayer/)
## Declaration
```python
async def getSpawnedPlayersForPlayer(id : int) -> list
```
## Parameters
`int` **id**: the player id.
## Returns
`list [int]`: ids of spawned players.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result
async def getStreamedPlayersByPlayer(id : int) -> list:
"""
This function is used to retrieve currently streamed players by given player. More details: Streamed players are basically clients, that has spawned given player in their game. Please notice, that player can be spawned only one way. Which means that there are situation were player 1 is spawned for player 2, but not the other way arount. Simple examples: - Invisible players cannot be seen, but they can see everyone nearby. - Flying around world using camera.
Original: [getStreamedPlayersByPlayer](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/streamer/getStreamedPlayersByPlayer/)
## Declaration
```python
async def getStreamedPlayersByPlayer(id : int) -> list
```
## Parameters
`int` **id**: the player id.
## Returns
`list [int]`: ids of streamed players.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return result

47
src/pyg2o/functions/waypoint.py
View File

@@ -1,47 +0,0 @@
from ..server import PythonWebsocketServer
from ..call_repr import get_call_repr
from typing import Optional
async def getNearestWaypoint(world : str, x : int, y : int, z : int, distance: int = -1) -> Optional[tuple]:
"""
This function is used to retrieve the information about nearest waypoint from the specified position.
Original: [getNearestWaypoint](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/waypoint/getNearestWaypoint/)
## Declaration
```python
async def getNearestWaypoint(world : str, x : int, y : int, z : int) -> Optional[tuple]
```
## Parameters
`str` **world**: the world name in which the waypoint exists.
`int` **x**: the position in the world on the x axis.
`int` **y**: the position in the world on the y axis.
`int` **z**: the position in the world on the z axis.
## Returns
`tuple (name, x, y, z)`: Waypoint information.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return (result['name'], result['x'], result['y'], result['z']) if result is not None else (None, None, None)
async def getWaypoint(world : str, name : str) -> Optional[tuple]:
"""
This function is used to retrieve the position of specified waypoint.
Original: [getWaypoint](https://gothicmultiplayerteam.gitlab.io/docs/0.3.0/script-reference/server-functions/waypoint/getWaypoint/)
## Declaration
```python
async def getWaypoint(world : str, name : str) -> Optional[tuple]
```
## Parameters
`str` **world**: the world name in which the waypoint exists.
`str` **name**: the name of the waypoint.
## Returns
`dict {x, y, z}`: The position of waypoint.
"""
data = f'return {get_call_repr()}'
server = await PythonWebsocketServer.get_server()
result = await server.make_request(data)
return (result['x'], result['y'], result['z']) if result is not None else (None, None, None)

279
src/pyg2o/server.py
View File

@@ -1,162 +1,143 @@
import websockets from __future__ import annotations
import asyncio
import json import json
import uuid import logging
from typing import Optional import asyncio
from .constants import Constant from weakref import WeakValueDictionary
from .functions.event import callEvent from fastapi import WebSocket, FastAPI, Depends, HTTPException, WebSocketDisconnect, WebSocketException
from .serialize import _deserialize from fastapi.security import HTTPBasic, HTTPBasicCredentials
from loguru import logger from uuid import uuid4
class Server:
_current_server: Server | None = None
class PythonWebsocketServer:
_current_server = None
def __init__(self, host: str, port: int, whitelist: list[str], ping_interval: int = 30):
self.host: str = host
self.port: int = port
self.ping_interval: int = ping_interval
self.whitelist = whitelist
self._messageHandlers: dict[str, callable] = dict()
self._requests_list: dict[str, asyncio.Future] = dict()
self._stop_event: asyncio.Event = asyncio.Event()
self._connected_socket: Optional[websockets.ClientConnection] = None
self._registerMessage('event', self._message_event)
self._registerMessage('init_constants', self._message_init_constants)
self._registerMessage('result', self._message_call_result)
@classmethod @classmethod
async def get_server(cls): def get_current_server(cls) -> Server:
if cls._current_server is None:
raise ConnectionError('PyG2O сервер не подключен')
return cls._current_server return cls._current_server
def _registerMessage(self, type: str, handler: callable): def __init__(self, *, app: FastAPI, server_username: str, server_password: str, client_password: str):
if type in self._messageHandlers: Server._current_server = self
return self._security = HTTPBasic()
self._server_token: str = ''
self._server_username = server_username
self._server_password = server_password
self._client_password = client_password
self._messageHandlers[type] = handler self._logger = logging.getLogger(__name__)
self._logger.addHandler(logging.NullHandler())
self._server_connection: WebSocket | None = None
self._requests: WeakValueDictionary[str, asyncio.Future] = WeakValueDictionary()
self._register_routes(app)
@classmethod
async def server_call(cls, message: str):
return await cls.get_current_server()._call(cls.get_current_server()._server_connection, message)
async def _call(self, socket: WebSocket | None, message: str):
if socket is None:
raise ConnectionError('PyG2O сервер не подключен')
request, data = self._make_request()
data['data'] = message
data = json.dumps(data)
# Меняем синтаксис под Squirrel
data = data.replace("'", '\\"').replace('True', 'true').replace('False', 'false')
async def _callMessage(self, type: str, data: dict): await socket.send_text(message)
if type not in self._messageHandlers: return request
return
def _make_request(self):
await self._messageHandlers[type](data) request_id = str(uuid4())
request = asyncio.Future()
async def start(self): self._requests[request_id] = request
async with websockets.serve(
self.handle_connection, data = {
host=self.host,
port=self.port,
ping_interval=self.ping_interval,
):
logger.success(f'Server is started at ws://{self.host}:{self.port}')
PythonWebsocketServer._current_server = self
asyncio.create_task(callEvent('onInit', **{}))
await self._stop_event.wait()
async def stop(self):
PythonWebsocketServer._current_server = None
self._connected_socket = None
self._stop_event.set()
async def make_request(self, data: str):
if (self._connected_socket is None):
return None
request_id = str(uuid.uuid4())
self._requests_list[request_id] = asyncio.get_running_loop().create_future()
request = {
'type': 'call',
'uuid': request_id, 'uuid': request_id,
'data': data, 'data': None,
} }
request = json.dumps(request)
request = request.replace("'", '\\"') return request, data
request = request.replace('True', 'true')
request = request.replace('False', 'false')
await self._connected_socket.send(request)
result = await asyncio.wait_for(
self._requests_list[request_id],
timeout=30
)
return result
async def _message_event(self, data: dict): def _register_routes(self, app):
if (not isinstance(data['data'], dict) or @app.websocket('/pyg2o/server')
'event' not in data['data']): async def pyg2o_main(websocket: WebSocket):
await self._handle_server_connection(websocket)
@app.websocket('/pyg2o/client/{playerid}')
async def pyg2o_client(websocket: WebSocket, playerid: int):
await self._handle_client_connection(websocket, playerid)
# Я потратил примерно 2ч чтобы понять, почему pyright игнорирует type: ignore
# Я сдаюсь, мне пришлось это добавить
_ = pyg2o_main
_ = pyg2o_client
async def _verify_token(self, credentials: HTTPBasicCredentials):
username = credentials.username
password = credentials.password
if username == self._server_username and password == self._server_password:
token = self._create_server_token()
if token is None:
raise HTTPException(status_code=403)
return token
elif password == self._client_password:
...
return None
def _create_server_token(self) -> str | None:
self._server_token = str(uuid4())
return self._server_token
async def _handle_auth_connection(self, credentials: HTTPBasicCredentials):
response: str | None = await self._verify_token(credentials)
if response is None:
raise HTTPException(status_code=401)
return {'token': response}
async def _handle_server_connection(self, websocket: WebSocket):
headers = websocket.headers
uuid = headers.get('Authorization')
if uuid != self._server_token:
# Закрытие до принятия подключения выбрасывает 403 (Forbidden) код, так что не нужны доп сообщения
await websocket.close()
return return
eventName = data['data']['event'] if self._server_connection is not None:
del data['data']['event'] await self._server_connection.close()
if 'desc' in data['data']: await websocket.accept()
obj_name = data['data']['desc']['obj_name'] self._server_connection = websocket
obj_data = data['data']['desc']['obj_data'] self._logger.info('PyG2O сервер подключился')
data['data']['desc'] = _deserialize(obj_name, obj_data)
elif 'itemGround' in data['data']:
obj_name = data['data']['itemGround']['obj_name']
obj_data = data['data']['itemGround']['obj_data']
data['data']['itemGround'] = _deserialize(obj_name, obj_data)
asyncio.create_task(callEvent(eventName, **data['data']))
async def _message_init_constants(self, data: dict):
if data['data'] is not dict:
return
Constant._update(data['data'])
async def _message_call_result(self, data: dict):
if data['uuid'] not in self._requests_list:
return
result = data['data']
if (isinstance(data['data'], dict) and
'obj_name' in data['data'] and
'obj_data' in data['data']):
result = _deserialize(result['obj_name'], result['obj_data'])
self._requests_list[data['uuid']].set_result(result)
del self._requests_list[data['uuid']]
async def handle_connection(self, websocket: websockets.ClientConnection):
if len(self.whitelist) != 0 and websocket.remote_address[0] not in self.whitelist:
await websocket.close(4000, 'Connection denied (whitelist)')
return
if self._connected_socket is not None:
await websocket.close(4000, 'Connection denied (already_connected)')
return
self._connected_socket = websocket
self.is_connected = websocket
logger.info(f'Client connected: {websocket.remote_address}')
asyncio.create_task(callEvent('onWebsocketConnect', **{}))
try: try:
async for message in websocket: while True:
try: try:
message_json = json.loads(message) data = await websocket.receive_text()
if not all(key in message_json for key in ('type', 'uuid', 'data')): message_data = json.loads(data)
logger.error(f'Expected message with (type, uuid, data) fields, got: {message_json}') self._logger.info(f'Сообщение сервера: {message_data}')
continue
await self._callMessage(message_json['type'], message_json)
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
logger.exception(f'JSON Exception: {e}') self._logger.exception(f'Ошибка декодирования JSON: {e}')
continue except WebSocketDisconnect:
except Exception as e: self._logger.info('PyG2O сервер отключился')
logger.exception(f'Exception: {e}') except WebSocketException as e:
continue self._logger.exception(f'Ошибка подключения PyG2O сервера: {e}')
except websockets.exceptions.ConnectionClosedError:
pass async def _process_server_message(self, message: dict):
finally: match message:
logger.info('Client disconnected') case {'uuid': id, 'data': data}:
self.is_connected = None ...
self._connected_socket = None case {'data': data}:
asyncio.create_task(callEvent('onWebsocketDisconnect', **{})) ...
case _:
raise ValueError(f'Неподдерживаемый тип PyG2O Server сообщения: {message}')
async def _handle_client_connection(self, websocket: WebSocket, playerid: int):
...

136
src/pyg2o/server_v3.py
View File

@@ -1,136 +0,0 @@
import json
import logging
import asyncio
from weakref import WeakValueDictionary
from fastapi import WebSocket, FastAPI, Depends, HTTPException, WebSocketDisconnect, WebSocketException
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from uuid import uuid4
class Server:
def __init__(self, *, app: FastAPI, server_username: str, server_password: str, client_password: str):
self._security = HTTPBasic()
self._server_token: str = ''
self._server_username = server_username
self._server_password = server_password
self._client_password = client_password
self._logger = logging.getLogger(__name__)
self._logger.addHandler(logging.NullHandler())
self._server_connection: WebSocket | None = None
self._requests: WeakValueDictionary[str, asyncio.Future] = WeakValueDictionary()
self._register_routes(app)
async def server_call(self, message: str):
return await self._call(self._server_connection, message)
async def _call(self, socket: WebSocket | None, message: str):
if socket is None:
raise ConnectionError('PyG2O сервер не подключен')
request, data = self._make_request()
data['data'] = message
data = json.dumps(data)
# Меняем синтаксис под Squirrel
data = data.replace("'", '\\"').replace('True', 'true').replace('False', 'false')
await socket.send_text(message)
return request
def _make_request(self):
request_id = str(uuid4())
request = asyncio.Future()
self._requests[request_id] = request
data = {
'uuid': request_id,
'data': None,
}
return request, data
def _register_routes(self, app):
@app.get('/pyg2o/auth')
async def pyg2o_auth(credentials: HTTPBasicCredentials = Depends(self._security)):
return await self._handle_auth_connection(credentials)
@app.websocket('/pyg2o/server')
async def pyg2o_main(websocket: WebSocket):
await self._handle_server_connection(websocket)
@app.websocket('/pyg2o/client/{playerid}')
async def pyg2o_client(websocket: WebSocket, playerid: int):
await self._handle_client_connection(websocket, playerid)
# Я потратил примерно 2ч чтобы понять, почему pyright игнорирует type: ignore
# Я сдаюсь, мне пришлось это добавить
_ = pyg2o_auth
_ = pyg2o_main
_ = pyg2o_client
async def _verify_token(self, credentials: HTTPBasicCredentials):
username = credentials.username
password = credentials.password
if username == self._server_username and password == self._server_password:
token = self._create_server_token()
if token is None:
raise HTTPException(status_code=403)
return token
elif password == self._client_password:
...
return None
def _create_server_token(self) -> str | None:
self._server_token = str(uuid4())
return self._server_token
async def _handle_auth_connection(self, credentials: HTTPBasicCredentials):
response: str | None = await self._verify_token(credentials)
if response is None:
raise HTTPException(status_code=401)
return {'token': response}
async def _handle_server_connection(self, websocket: WebSocket):
headers = websocket.headers
uuid = headers.get('Authorization')
if uuid != self._server_token:
# Закрытие до принятия подключения выбрасывает 403 (Forbidden) код, так что не нужны доп сообщения
await websocket.close()
return
if self._server_connection is not None:
await self._server_connection.close()
await websocket.accept()
self._server_connection = websocket
self._logger.info('PyG2O сервер подключился')
try:
while True:
try:
data = await websocket.receive_text()
message_data = json.loads(data)
self._logger.info(f'Сообщение сервера: {message_data}')
except json.JSONDecodeError as e:
self._logger.exception(f'Ошибка декодирования JSON: {e}')
except WebSocketDisconnect:
self._logger.info('PyG2O сервер отключился')
except WebSocketException as e:
self._logger.exception(f'Ошибка подключения PyG2O сервера: {e}')
async def _process_server_message(self, message: dict):
match message:
case {'uuid': id, 'data': data}:
...
case {'data': data}:
...
case _:
raise ValueError(f'Неподдерживаемый тип PyG2O Server сообщения: {message}')
async def _handle_client_connection(self, websocket: WebSocket, playerid: int):
...