pyscenario

Classes

`NetworkConfiguration`([host, tcp_port, ...])	A class that represents the default network configuration.
`Protocol`(*values)	An enum that represents the supported protocols.
`QueueManager`(send_queue, receive_queue)	A class that manages queues.
`TaskManager`([send_task, receive_task])	A class that manages tasks.

Modules

`client`	IFSEI Telnet Client Module.
`config_schema`	Validation schema for config file.
`const`	Constants for the Scenario IFSEI.
`ifsei`	IFSEI Module.
`manager`	Device Manager Module.

Documentation

class pyscenario.Protocol(*values)[source]

Bases: Enum

An enum that represents the supported protocols.

TCP = 1

UDP = 2

class pyscenario.NetworkConfiguration(host: str = '192.168.1.20', tcp_port: int = 28000, udp_port: int = 25200, protocol: Protocol = Protocol.TCP, reconnect: bool = True, reconnect_delay: float = 10.0)[source]

Bases: object

A class that represents the default network configuration.

host: str = '192.168.1.20'

tcp_port: int = 28000

udp_port: int = 25200

protocol: Protocol = 1

reconnect: bool = True

reconnect_delay: float = 10.0

__annotations__ = {'host': <class 'str'>, 'protocol': <enum 'Protocol'>, 'reconnect': <class 'bool'>, 'reconnect_delay': <class 'float'>, 'tcp_port': <class 'int'>, 'udp_port': <class 'int'>}

__dataclass_fields__ = {'host': Field(name='host',type=<class 'str'>,default='192.168.1.20',default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'protocol': Field(name='protocol',type=<enum 'Protocol'>,default=<Protocol.TCP: 1>,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'reconnect': Field(name='reconnect',type=<class 'bool'>,default=True,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'reconnect_delay': Field(name='reconnect_delay',type=<class 'float'>,default=10.0,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'tcp_port': Field(name='tcp_port',type=<class 'int'>,default=28000,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'udp_port': Field(name='udp_port',type=<class 'int'>,default=25200,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}

__dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False,match_args=True,kw_only=False,slots=False,weakref_slot=False)

__eq__(other): Return self==value.

__hash__ = None

__init__(host: str = '192.168.1.20', tcp_port: int = 28000, udp_port: int = 25200, protocol: Protocol = Protocol.TCP, reconnect: bool = True, reconnect_delay: float = 10.0) → None

__match_args__ = ('host', 'tcp_port', 'udp_port', 'protocol', 'reconnect', 'reconnect_delay')

__repr__(): Return repr(self).

class pyscenario.QueueManager(send_queue: Queue, receive_queue: Queue)[source]

Bases: object

A class that manages queues.

send_queue: Queue

receive_queue: Queue

__annotations__ = {'receive_queue': <class 'asyncio.queues.Queue'>, 'send_queue': <class 'asyncio.queues.Queue'>}

__dataclass_fields__ = {'receive_queue': Field(name='receive_queue',type=<class 'asyncio.queues.Queue'>,default=<dataclasses._MISSING_TYPE object>,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'send_queue': Field(name='send_queue',type=<class 'asyncio.queues.Queue'>,default=<dataclasses._MISSING_TYPE object>,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}

__dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False,match_args=True,kw_only=False,slots=False,weakref_slot=False)

__eq__(other): Return self==value.

__hash__ = None

__init__(send_queue: Queue, receive_queue: Queue) → None

__match_args__ = ('send_queue', 'receive_queue')

__repr__(): Return repr(self).

class pyscenario.TaskManager(send_task: Task | None = None, receive_task: Task | None = None)[source]

Bases: object

A class that manages tasks.

send_task: Task | None = None

receive_task: Task | None = None

__annotations__ = {'receive_task': _asyncio.Task | None, 'send_task': _asyncio.Task | None}

__dataclass_fields__ = {'receive_task': Field(name='receive_task',type=_asyncio.Task | None,default=None,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'send_task': Field(name='send_task',type=_asyncio.Task | None,default=None,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}

__dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False,match_args=True,kw_only=False,slots=False,weakref_slot=False)

__eq__(other): Return self==value.

__hash__ = None

__init__(send_task: Task | None = None, receive_task: Task | None = None) → None

__match_args__ = ('send_task', 'receive_task')

__repr__(): Return repr(self).