EquipmentHandler¶

class secsgem.gem.equipmenthandler.GemEquipmentHandler(connection: secsgem.common.protocol.Protocol, initial_control_state: str = 'ATTEMPT_ONLINE', initial_online_control_state: str = 'REMOTE')[source]¶

Bases: secsgem.gem.handler.GemHandler

Baseclass for creating equipment models. Inherit from this class and override required functions.

control_state¶: Get control state.

control_switch_online()[source]¶: Operator switches to online control state.

control_switch_offline()[source]¶: Operator switches to offline control state.

control_switch_online_local()[source]¶: Operator switches to the local online control state.

control_switch_online_remote()[source]¶: Operator switches to the local online control state.

data_values¶

Get list of the data values.

Returns:	Data value list
Return type:	list of `secsgem.gem.DataValue`

on_dv_value_request(data_value_id: secsgem.secs.variables.base.Base, data_value: secsgem.gem.data_value.DataValue) → secsgem.secs.variables.base.Base[source]¶

Get the data value depending on its configuation.

Override in inherited class to provide custom data value request handling.

Parameters:	dvid (`secsgem.secs.variables.Base`) – Id of the data value encoded in the corresponding type dv (`secsgem.gem.DataValue`) – The data value requested
Returns:	The value encoded in the corresponding type
Return type:	`secsgem.secs.variables.Base`

status_variables¶

Get list of the status variables.

Returns:	Status variable list
Return type:	list of `secsgem.gem.StatusVariables`

on_sv_value_request(svid: secsgem.secs.variables.base.Base, status_variable: secsgem.gem.status_variable.StatusVariable) → secsgem.secs.variables.base.Base[source]¶

Get the status variable value depending on its configuation.

Override in inherited class to provide custom status variable request handling.

Parameters:	svid (`secsgem.secs.variables.Base`) – Id of the status variable encoded in the corresponding type sv (`secsgem.gem.StatusVariable`) – The status variable requested
Returns:	The value encoded in the corresponding type
Return type:	`secsgem.secs.variables.Base`

collection_events¶

Get list of the collection events.

Returns:	Collection event list
Return type:	list of `secsgem.gem.CollectionEvent`

registered_reports¶

Get list of the subscribed reports.

Returns:	Collection event report list
Return type:	dictionary of subscribed reports

registered_collection_events¶

Get list of the subscribed collection events.

Returns:	Collection event list
Return type:	dictionary of `secsgem.gem.CollectionEventLink`

trigger_collection_events(ceids: List[Union[str, int]])[source]¶

Triggers the supplied collection events.

Parameters:	ceids (list of various) – List of collection events

equipment_constants¶

The list of the equipments contstants.

Returns:	Equipment constant list
Return type:	list of `secsgem.gem.EquipmentConstant`

on_ec_value_request(equipment_constant_id: secsgem.secs.variables.base.Base, equipment_constant: secsgem.gem.equipment_constant.EquipmentConstant) → secsgem.secs.variables.base.Base[source]¶

Get the equipment constant value depending on its configuation.

Override in inherited class to provide custom equipment constant request handling.

Parameters:	ecid (`secsgem.secs.variables.Base`) – Id of the equipment constant encoded in the corresponding type ec (`secsgem.gem.EquipmentConstant`) – The equipment constant requested
Returns:	The value encoded in the corresponding type
Return type:	`secsgem.secs.variables.Base`

on_ec_value_update(equipment_constant_id: secsgem.secs.variables.base.Base, equipment_constant: secsgem.gem.equipment_constant.EquipmentConstant, value: Union[int, float])[source]¶

Set the equipment constant value depending on its configuation.

Override in inherited class to provide custom equipment constant update handling.

Parameters:	ecid (`secsgem.secs.variables.Base`) – Id of the equipment constant encoded in the corresponding type ec (`secsgem.gem.EquipmentConstant`) – The equipment constant to be updated value (`secsgem.secs.variables.Base`) – The value encoded in the corresponding type

alarms¶

Get the list of the alarms.

Returns:	Alarms list
Return type:	list of `secsgem.gem.Alarm`

set_alarm(alid: Union[int, str])[source]¶

Set the list of the alarms.

Parameters:	alid (str/int) – Alarm id

clear_alarm(alid: Union[int, str])[source]¶

Clear the list of the alarms.

Parameters:	alid (str/int) – Alarm id

remote_commands¶

Get list of the remote commands.

Returns:	Remote command list
Return type:	list of `secsgem.gem.RemoteCommand`

on_connection_closed(connection)[source]¶: Handle connection was closed event.

are_you_there()¶: Check if remote is still replying.

callbacks¶: Property for callback handling.

communication_state¶: Get the communication state model.

disable() → None¶: Disable the connection.

disable_ceid_reports()¶: Disable all Collection Event Reports.

disable_ceids()¶: Disable all Collection Events.

enable() → None¶: Enable the connection.

events¶: Wrapper for connections events.

get_ceid_name(ceid)¶

Get the name of a collection event.

Parameters:	ceid (integer) – ID of collection event
Returns:	Name of the event or empty string if not found
Return type:	string

get_dvid_name(dvid)¶

Get the name of a data value.

Parameters:	dvid (integer) – ID of data value
Returns:	Name of the event or empty string if not found
Return type:	string

classmethod hsms(address, port, active, session_id, name, custom_connection_handler=None, **kwargs) → secsgem.secs.handler.SecsHandler¶

Initialize a secs handler using a hsms connection.

All arguments will be passed to the HSMS handler.

list_ecs(ecs=None)¶

Get list of available Equipment Constants.

Returns:	available Equipment Constants
Return type:	list

list_svs(svs=None)¶

Get list of available Service Variables.

Returns:	available Service Variables
Return type:	list

on_commack_requested() → int¶

Get the acknowledgement code for the connection request.

override to accept or deny connection request

Returns:	0 when connection is accepted, 1 when connection is denied
Return type:	integer

protocol¶: Get the connection for the handler.

register_stream_function(stream: int, function: int, callback)¶

Register the function callback for stream and function.

Parameters:	stream (integer) – stream to register callback for function (integer) – function to register callback for callback (def callback(connection)) – method to call when stream and functions is received

request_ec(ec_id)¶

Request contents of one Equipment Constant.

Parameters:	ec_id (int) – id of Equipment Constant
Returns:	value of requested Equipment Constant
Return type:	various

request_ecs(ecs)¶

Request contents of supplied Equipment Constants.

Parameters:	ecs (list) – Equipment Constants to request
Returns:	values of requested Equipment Constants
Return type:	list

request_process_program(ppid: Union[int, str]) → Tuple[Union[int, str], str]¶

Request a process program.

Parameters:	ppid (string) – Transferred process programs ID

request_sv(sv_id)¶

Request contents of one Service Variable.

Parameters:	sv_id (int) – id of Service Variable
Returns:	value of requested Service Variable
Return type:	various

request_svs(svs)¶

Request contents of supplied Service Variables.

Parameters:	svs (list) – Service Variables to request
Returns:	values of requested Service Variables
Return type:	list

secs_decode(packet)¶

Get object of decoded stream and function class, or None if no class is available.

Parameters:	packet (`secsgem.common.Packet`) – packet to get object for
Returns:	matching stream and function object
Return type:	secsSxFx object

send_and_waitfor_response(*args, **kwargs)¶: Wrapper for connections send_and_waitfor_response function.

send_equipment_terminal(terminal_id, text)¶

Set text to equipment terminal.

Parameters:	terminal_id (int) – ID of terminal text (string) – text to send

send_process_program(ppid: Union[int, str], ppbody: str)¶

Send a process program.

Parameters:	ppid (string) – Transferred process programs ID ppbody (string) – Content of process program

send_response(*args, **kwargs)¶: Wrapper for connections send_response function.

send_stream_function(*args, **kwargs)¶: Wrapper for connections send_stream_function function.

serialize_data() → Dict[str, Any]¶

Get serialized data.

Returns:	data to serialize for this object
Return type:	dict

set_ec(ec_id, value)¶

Set contents of one Equipment Constant.

Parameters:	ec_id (int) – id of Equipment Constant value (various) – new content of Equipment Constant

set_ecs(ecs)¶

Set contents of supplied Equipment Constants.

Parameters:	ecs (list) – list containing list of id / value pairs

stream_function(stream, function)¶

Get class for stream and function.

Parameters:	stream (int) – stream to get function for function (int) – function to get
Returns:	matching stream and function class
Return type:	secsSxFx class

unregister_stream_function(stream, function)¶

Unregister the function callback for stream and function.

Parameters:	stream (integer) – stream to unregister callback for function (integer) – function to register callback for

waitfor_communicating(timeout: Optional[float] = None) → bool¶

Wait until connection gets into communicating state. Returns immediately if state is communicating.

Parameters:	timeout (float) – seconds to wait before aborting
Returns:	True if state is communicating, False if timed out
Return type:	bool

class secsgem.gem.equipmenthandler.DataValue(dvid: Union[int, str], name: str, value_type: Type[secsgem.secs.variables.base.Base], use_callback: bool = True, **kwargs)[source]¶

Bases: object

Data value definition.

class secsgem.gem.equipmenthandler.StatusVariable(svid, name, unit, value_type, use_callback=True, **kwargs)[source]¶

Bases: object

Status variable definition.

class secsgem.gem.equipmenthandler.CollectionEvent(ceid: Union[int, str], name: str, data_values: List[Union[str, int]], **kwargs)[source]¶

Bases: object

Collection event definition.

class secsgem.gem.equipmenthandler.CollectionEventLink(collection_event: secsgem.gem.collection_event.CollectionEvent, reports: List[secsgem.gem.collection_event_report.CollectionEventReport], **kwargs)[source]¶

Bases: object

Representation for registered/linked collection event.

collection_event¶: Get the associated collection event.

reports¶

Get list of the data values.

Returns:	List of linked reports
Return type:	list of `gem.CollectionEventReport`

class secsgem.gem.equipmenthandler.CollectionEventReport(rptid: Union[int, str], variables: List[Union[str, int]], **kwargs)[source]¶

Bases: object

Report definition for registered collection events.

class secsgem.gem.equipmenthandler.EquipmentConstant(ecid: Union[int, str], name: str, min_value: Union[int, float], max_value: Union[int, float], default_value: Union[int, float], unit: str, value_type: Type[secsgem.secs.variables.base.Base], use_callback: bool = True, **kwargs)[source]¶

Bases: object

Equipment constant definition.