gates.py

gates.py

from dataclasses import dataclass
from typing import Optional, Tuple

from samp import (
    KEY_CROUCH,
    GetPlayerFacingAngle,
    GetPlayerPos,
    IsPlayerInAnyVehicle,
    KillTimer,
    SendClientMessage,
    SetTimer,
)

from .cmdparser import cmd
from .elevators import distance, rotate
from .model import Gate as DBGate, session as sessionmaker
from .streamer import (
    CreateDynamicObject,
    DestroyDynamicObject,
    MoveDynamicObject,
    StopDynamicObject,
)
from .tms import admin_level


@dataclass
class GateType:
    name: str
    model_id: int
    closed_offset: Tuple[float, float, float]
    open_offset: Tuple[float, float, float]


@dataclass
class Gate:
    type: GateType
    position: Tuple[float, float, float]
    rotation: float
    object_id: int
    id: Optional[int] = None
    opened: bool = False
    _close_timer_id: Optional[int] = None

    _types = {
        type.name: type
        for type in (
            GateType(
                name='small',
                model_id=19870,
                closed_offset=(-2.37, 0, .8),
                open_offset=(-2.37, 0, -2.75),
            ),
            GateType(
                name='large',
                model_id=976,
                closed_offset=(-4.43, 0, -0.85),
                open_offset=(-4.43, 0, -4.18),
            ),
            GateType(
                name='xl',
                model_id=980,
                closed_offset=(0, 0, 1.75),
                open_offset=(0, 0, -3.78),
            ),
        )
    }

    def toggle(self):
        if not self.opened:
            self.open()
        else:
            self.close()

    def open(self):
        open_offset = self.type.open_offset
        gate_position = tuple(
            xyz + offset_xyz
            for xyz, offset_xyz in zip(
                self.position,
                rotate(open_offset[:2], self.rotation) + open_offset[2:],
            )
        )
        StopDynamicObject(self.object_id)
        MoveDynamicObject(self.object_id, *gate_position, 2.5)
        self._kill_close_timer()
        self._close_timer_id = SetTimer(self.close, 5000, False)
        self.opened = True

    def close(self):
        closed_offset = self.type.closed_offset
        gate_position = tuple(
            xyz + offset_xyz
            for xyz, offset_xyz in zip(
                self.position,
                rotate(closed_offset[:2], self.rotation) + closed_offset[2:],
            )
        )
        StopDynamicObject(self.object_id)
        MoveDynamicObject(self.object_id, *gate_position, 2.5)
        self._kill_close_timer()
        self.opened = False

    def _kill_close_timer(self):
        if self._close_timer_id is not None:
            KillTimer(self._close_timer_id)
            self._close_timer_id = None

    def destroy(self):
        DestroyDynamicObject(self.object_id)


gates = []


def OnGameModeInit():
    with sessionmaker() as session:
        for db_gate in session.query(DBGate):
            gate = add_gate(
                db_gate.type,
                (db_gate.x, db_gate.y, db_gate.z),
                db_gate.rotation
            )
            gate.id = db_gate.id


def OnGameModeExit():
    global gates

    for gate in gates:
        gate.destroy()


def OnPlayerKeyStateChange(playerid, newkeys, oldkeys):
    if not IsPlayerInAnyVehicle(playerid):
        return

    if (newkeys & KEY_CROUCH) and not (oldkeys & KEY_CROUCH):
        gate = find_closest_gate(GetPlayerPos(playerid))

        if not gate:
            return

        gate.toggle()


def add_gate(type: str, position: Tuple[float, float, float], rotation: float):
    if type not in Gate._types:
        raise ValueError(f'Unknown gate type {type}')

    gate_type = Gate._types[type]
    closed_offset = gate_type.closed_offset
    gate_position = tuple(
        xyz + offset_xyz
        for xyz, offset_xyz in zip(
            position,
            rotate(closed_offset[:2], rotation) + closed_offset[2:],
        )
    )

    gate = Gate(
        type=gate_type,
        position=position,
        rotation=rotation,
        object_id=CreateDynamicObject(
            gate_type.model_id,
            *gate_position,
            0,
            0,
            rotation,
        ),
    )

    global gates
    gates.append(gate)

    return gate


def find_closest_gate(from_coords):
    for gate in gates:
        if distance(from_coords, gate.position) < 5:
            return gate


@cmd('addgate', requires=admin_level(5))
def addgate(playerid, type=''):
    if not type:
        types = ', '.join(f'"{type}"' for type in Gate._types)
        SendClientMessage(
            playerid,
            0xFF0000FF,
            'USAGE: /addgate type '
            f'(possible choices: {types})',
        )
        return

    position = GetPlayerPos(playerid)
    rotation = GetPlayerFacingAngle(playerid)

    try:
        gate = add_gate(type, position, rotation)
    except ValueError:
        types = ', '.join(f'"{type}"' for type in Gate._types)
        SendClientMessage(
            playerid,
            0xFF0000FF,
            f'ERROR: Gate type "{type}" does not exist '
            f'(possible choices: {types})',
        )
        return

    with sessionmaker() as session:
        db_gate = DBGate(
            type=type,
            x=position[0],
            y=position[1],
            z=position[2],
            rotation=rotation,
        )
        session.add(db_gate)
        session.commit()
        gate.id = db_gate.id

    SendClientMessage(playerid, 0x4080FFFF, 'Gate was successfully added.')


@cmd('delgate', requires=admin_level(5))
def delgate(playerid):
    gate = find_closest_gate(GetPlayerPos(playerid))

    if not gate:
        SendClientMessage(
            playerid,
            0xFF0000FF,
            'ERROR: Not in range of any gate',
        )
        return

    with sessionmaker() as session:
        db_gate = session.query(DBGate).get(gate.id)

        if db_gate:
            session.delete(db_gate)

        session.commit()

    gate.destroy()

    global gates
    gates.remove(gate)

    SendClientMessage(playerid, 0x4080FFFF, 'Gate was successfully removed.')


@cmd
def open(playerid):
    gate = find_closest_gate(GetPlayerPos(playerid))

    if not gate:
        SendClientMessage(
            playerid,
            0xFF0000FF,
            'ERROR: Not in range of any gate',
        )
        return

    if not gate.opened:
        gate.open()