TianWan/Agreement/modus/skio/define.py

import enum
import logging
import typing
from dataclasses import dataclass
# from skio.worker.state import SlotInfo
from typing import Any


class ProtocolType(enum.Enum):
    # beijing & 301
    RCV = 5
    # SMNPC
    SMPXI = 10
    SMHSL = 11
    modbus = 12
    # TXS
    TXS = 12
    TXS_PXI_Dev = 13


class SigType(enum.Enum):
    ALIAS = 1

    STO = 10
    SAI, SAO = 11, 12
    SDI, SDO = 13, 14
    DPO = 15
    SCTR = 16

    HSL_BEAT = 17
    HSLO = 18
    NIS_F, NIS_A = 19, 20

    AI, AO, AIO = 1, 2, 3
    DI, DO, DIO = 4, 5, 6

    SHSL = 7

    TXS, FREG = 21, 22
    FAI, FAO, FDI, FDO = 23, 24, 25, 26
    FAV, FPT100 = 27, 28


class ValType(enum.Enum):
    B1 = 0
    U8, U16, U32, U64 = 1, 2, 3, 4
    I8, I16, I32, I64 = 5, 6, 7, 8
    F32, D64 = 9, 10
    BOOL = 11
    s32 = 12


@dataclass
class BlockInfo(object):
    idx: int
    off: int
    bit: int
    vt: ValType
    length: int


@dataclass
class FVar(object):
    id: int
    name: str
    sig_type: SigType
    val_type: ValType
    length: int = 1
    uri: str = ''
    slot: str = ''
    si: typing.Optional['SlotInfo'] = None
    eu: str = None
    rlo: float = None
    rhi: float = None
    elo: float = None
    ehi: float = None

    channel: int = 1
    trigger: bool = False
    group: list = None
    trans_value: typing.Union[str, int, bool, float] = None

    # 是否使用福清port回读，当硬件置值过来的时候使用
    txs_port: bool = False


@dataclass
class IVar(object):
    id: int
    name: str
    sig_type: SigType
    chr: ValType
    reg: int
    length: int = 1
    slot: str = ''
    si: typing.Optional['SlotInfo'] = None
    engineering_unit: str = None
    rlo: float = None
    rhi: float = None
    elo: float = None
    ehi: float = None


T_Val = typing.Union[float, int, bool]
LOGGER = logging.getLogger('SkIO')


class IDev(object):
    def setup(self, uri):
        raise NotImplementedError

    def read(self, var: IVar) -> T_Val:
        raise NotImplementedError

    def write(self, var: IVar, value: T_Val) -> T_Val:
        raise NotImplementedError


class CmdType(enum.Enum):
    SETUP = 0
    PING, PONG = 1, 2
    READ, WRITE = 3, 4
    FQREAD, FQWRITE = 6, 7

    # 执行 txs 脚本
    TXSPY = 5

    SUCCESS, ERROR = 10, 11


@dataclass
class KCommand(object):
    type: CmdType
    body: Any = None


@dataclass
class KReply(object):
    type: CmdType
    body: Any = None