Cloud

Cloud class

Cloud

 Cloud (truck:tspace.config.vehicles.TruckInCloud,
        can_server:tspace.config.messengers.CANMessenger,
        epi_countdown_time:float=3.0,
        trip_server:Optional[tspace.config.messengers.TripMessenger]=None,
        ui:str='UDP', remotecan:Optional[tspace.conn.remote_can_client.Rem
        oteCanClient]=None, remoteClient_lock:Optional[<built-
        infunctionallocate_lock>]=None,
        driver:tspace.config.drivers.Driver, resume:bool=False,
        data_dir:Optional[pathlib.Path]=None, flash_count:int=0,
        episode_count:int=0, vcu_calib_table_row_start:int=0,
        torque_table_default:Optional[pandas.core.frame.DataFrame]=None,
        torque_table_live:Optional[pandas.core.frame.DataFrame]=None,
        lock_watchdog:<built-infunctionallocate_lock>=<unlocked
        _thread.lock object at 0x7f9a997f6f80>,
        capture_failure_count:int=0, flash_failure_count:int=0,
        logger:Optional[logging.Logger]=None,
        dict_logger:Optional[dict]=None)

*Kvaser is local vehicle interface with Producer(get vehicle status) and Consumer(flasher)

Attributes:

truck: TruckInCloud
    truck type is TruckInCloud
can_server: CANMessenger
    can_server type is CANMessenger
trip_server: Optional[TripMessenger] = None
    trip_server type is TripMessenger
ui: str = "UDP"
    ui must be cloud, local or mobile, not {self.ui}
web_srv = ("rocket_intra",)
    web_srv is a tuple of str
epi_countdown_time: float = 3.0
    epi_countdown_time is a float
remotecan: Optional[RemoteCanClient] = None
    RemoteCanClient type is RemoteCanClient
rmq_consumer: Optional[ClearablePullConsumer] = None
    ClearablePullConsumer type is ClearablePullConsumer
rmq_message_ready: Optional[Message] = None
    Message type is Message
rmq_producer: Optional[Producer] = None
    Producer type is Producer
remoteClient_lock: Optional[Lock] = None
    Lock type is Lock*

source

Cloud.init_cloud

 Cloud.init_cloud ()

initialize cloud interface, set proxy and remote can client and the lock

source

Cloud.init_internal_pipelines

 Cloud.init_internal_pipelines ()

initialize internal pipeline static type for cloud interface

source

Cloud.flash_vehicle

 Cloud.flash_vehicle (torque_table:pandas.core.frame.DataFrame)

flash torque table to the VCU

source

Cloud.hmi_select

 Cloud.hmi_select ()

*select HMI interface according to ui type.

Produce data into the pipeline main entry to the capture thread sub-thread method Callable input parameters example: hmi_pipeline: Pipeline[str],

Return: Callback for the HMI thread with type Callable[[Pipeline[str], Optional[Event]], None]*

source

Cloud.produce

 Cloud.produce (raw_pipeline:tspace.dataflow.pipeline.deque.PipelineDQ[typ
                ing.Union[dict[str,str],dict[str,dict[str,list[typing.Unio
                n[str,list[list[str]]]]]]]], hmi_pipeline:Optional[tspace.
                dataflow.pipeline.queue.Pipeline[str]]=None,
                exit_event:Optional[threading.Event]=None)

Create secondary threading pool for HMI thread and data capture thread

	Type	Default	Details
raw_pipeline	PipelineDQ
hmi_pipeline	Optional	None	PipelineDQ[dict[str, dict[str, list[Union[str, list[str]]]]]],
exit_event	Optional	None

source

Cloud.data_capture_from_remotecan

 Cloud.data_capture_from_remotecan (raw_pipeline:tspace.dataflow.pipeline.
                                    deque.PipelineDQ[typing.Union[dict[str
                                    ,str],dict[str,dict[str,list[typing.Un
                                    ion[str,list[list[str]]]]]]]], exit_ev
                                    ent:Optional[threading.Event]=None)

Callback for the data capture thread

	Type	Default	Details
raw_pipeline	PipelineDQ
exit_event	Optional	None	PipelineDQ[dict[str, dict[str, list[Union[str, list[str]]]]]],

source

Cloud.cloud_ping

 Cloud.cloud_ping ()

utility function for ping test

source

Cloud.cloud_telnet_test

 Cloud.cloud_telnet_test ()

Utility function for telnet test

source

Cloud.hmi_capture_from_udp

 Cloud.hmi_capture_from_udp
                             (hmi_pipeline:tspace.dataflow.pipeline.queue.
                             Pipeline[str],
                             exit_event:Optional[threading.Event]=None)

Callback function for getting HMI message from local UDP

source

Cloud.hmi_capture_from_rmq

 Cloud.hmi_capture_from_rmq
                             (hmi_pipeline:tspace.dataflow.pipeline.queue.
                             Pipeline[str],
                             exit_event:Optional[threading.Event]=None)

Get the hmi message from RocketMQ

source

Cloud.hmi_capture_from_dummy

 Cloud.hmi_capture_from_dummy
                               (hmi_pipeline:tspace.dataflow.pipeline.queu
                               e.Pipeline[str],
                               exit_event:threading.Event)

*Get the hmi status from dummy state management and remote can module

The only way to change the state with dummy mode is through Graceful Killer (Ctrl +C), which is triggered by GracefulKiller in the Cruncher thread and received in hmi_control*

source

Cloud.filter

 Cloud.filter (in_pipeline:tspace.dataflow.pipeline.deque.PipelineDQ[typin
               g.Union[dict[str,str],dict[str,dict[str,list[typing.Union[s
               tr,list[list[str]]]]]]]], out_pipeline:tspace.dataflow.pipe
               line.queue.Pipeline[pandas.core.frame.DataFrame],
               start_event:Optional[threading.Event],
               stop_event:Optional[threading.Event],
               interrupt_event:Optional[threading.Event],
               flash_event:Optional[threading.Event],
               exit_event:Optional[threading.Event])

Callback function for the data filter thread, encapsulating the data into pandas DataFrame

	Type	Details
in_pipeline	PipelineDQ	input pipelineDQ[raw data],
out_pipeline	Pipeline	output pipeline[DataFrame]
start_event	Optional	input event start
stop_event	Optional	not used for cloud
interrupt_event	Optional	not used for cloud
flash_event	Optional
exit_event	Optional	input event exit
Returns	None