NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

external_messenger.py (6083B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\sovereignty\workforce\client\external_messenger.py
      2 from publicGateway.grpc.exceptions import GenericException
      3 from sovereignty.client.base_external_messenger import BaseClientExternalMessenger
      4 from six.moves import http_client as httplib
      5 from eveProto.generated.eve_public.sovereignty.hub.workforce.api import requests_pb2
      6 from eveProto.generated.eve_public.sovereignty.hub.hub_pb2 import Identifier as HubIdentifier
      7 from eveProto.generated.eve_public.sovereignty.hub.workforce.workforce_pb2 import Configuration, ImportConfiguration, ExportConfiguration
      8 from eveProto.generated.eve_public.solarsystem.solarsystem_pb2 import Identifier as SolarSystemID
      9 import sovereignty.workforce.client.errors as errors
     10 from sovereignty.workforce.client.data_types import WorkforceConfiguration, WorkforceState, WorkforceNetworkedHub
     11 MODE_IDLE = 1
     12 MODE_IMPORT = 2
     13 MODE_EXPORT = 3
     14 MODE_TRANSIT = 4
     15 
     16 class WorkforceExternalMessenger(BaseClientExternalMessenger):
     17 
     18     def get_configuration(self, hub_id):
     19         try:
     20             request = requests_pb2.GetConfigurationRequest(hub=HubIdentifier(sequential=hub_id))
     21             request_wrapper, response_wrapper, response_payload = self._send_request(request, requests_pb2.GetConfigurationResponse)
     22         except GenericException as e:
     23             if e.response_primitive.status_code == httplib.FORBIDDEN:
     24                 raise errors.SovWorkforceGetConfigForbiddenError(e)
     25             if e.response_primitive.status_code == httplib.INTERNAL_SERVER_ERROR:
     26                 raise errors.SovWorkforceGetConfigInternalError(e)
     27             raise errors.SovWorkforceGetConfigError(e)
     28 
     29         configuration = response_payload.configuration
     30         workforceConfiguration = WorkforceConfiguration.create_from_proto(hub_id, configuration)
     31         return workforceConfiguration
     32 
     33     def configure(self, hub_id, workforceConfiguration):
     34         try:
     35             newConfiguration = self._get_new_configuration(workforceConfiguration)
     36             if newConfiguration:
     37                 request = requests_pb2.ConfigureRequest(hub=HubIdentifier(sequential=hub_id), configuration=newConfiguration)
     38                 request_wrapper, response_wrapper, response_payload = self._send_request(request, requests_pb2.ConfigureResponse)
     39                 return True
     40         except GenericException as e:
     41             if e.response_primitive.status_code == httplib.FORBIDDEN:
     42                 raise errors.SovWorkforceConfigureForbiddenError(e)
     43             if e.response_primitive.status_code == httplib.INTERNAL_SERVER_ERROR:
     44                 raise errors.SovWorkforceConfigureInternalError(e)
     45             raise errors.SovWorkforceConfigureError(e)
     46 
     47     def _get_new_configuration(self, workforceConfiguration):
     48         import_configuration = workforceConfiguration.import_configuration
     49         if import_configuration:
     50             source_system_ids = import_configuration.source_system_ids
     51             source_system_ids_list = sorted(source_system_ids)
     52             importSources = [ ImportConfiguration.Source(source=SolarSystemID(sequential=x)) for x in source_system_ids_list ]
     53             return Configuration(import_settings=ImportConfiguration(sources=importSources))
     54         export_configuration = workforceConfiguration.export_configuration
     55         if export_configuration:
     56             if export_configuration.destination_system_id is None:
     57                 return Configuration(export_settings=ExportConfiguration(no_destination=True, amount=export_configuration.amount))
     58             else:
     59                 return Configuration(export_settings=ExportConfiguration(destination_system=SolarSystemID(sequential=export_configuration.destination_system_id), amount=export_configuration.amount))
     60         inactive = workforceConfiguration.inactive
     61         if inactive:
     62             return Configuration(inactive=True)
     63         transit = workforceConfiguration.transit
     64         if transit:
     65             return Configuration(transit=True)
     66 
     67     def get_state(self, hub_id):
     68         try:
     69             request = requests_pb2.GetStateRequest(hub=HubIdentifier(sequential=hub_id))
     70             request_wrapper, response_wrapper, response_payload = self._send_request(request, requests_pb2.GetStateResponse)
     71         except GenericException as e:
     72             if e.response_primitive.status_code == httplib.FORBIDDEN:
     73                 raise errors.SovWorkforceGetStateForbiddenError(e)
     74             if e.response_primitive.status_code == httplib.INTERNAL_SERVER_ERROR:
     75                 raise errors.SovWorkforceGetStateInternalError(e)
     76             raise errors.SovWorkforceGetStateError(e)
     77 
     78         state = response_payload.state
     79         workforceState = WorkforceState.create_from_proto(hub_id, state)
     80         return workforceState
     81 
     82     def get_networkable_hubs(self, hub_id):
     83         try:
     84             request = requests_pb2.GetNetworkableHubsRequest(hub=HubIdentifier(sequential=hub_id))
     85             request_wrapper, response_wrapper, response_payload = self._send_request(request, requests_pb2.GetNetworkableHubsResponse)
     86         except GenericException as e:
     87             raise errors.SovWorkforceGetNetworkableHubsError(e)
     88 
     89         hubs = response_payload.hubs
     90         networkedHubs = [ WorkforceNetworkedHub.create_from_proto(x.hub.sequential, x.system.sequential, x.configuration, x.state) for x in hubs ]
     91         return networkedHubs
     92 
     93 
     94 class FakeWorkforceExternalMessenger(BaseClientExternalMessenger):
     95     workforceByHubID = {}
     96 
     97     def _GetConfiguration(self, hubID):
     98         if hubID in self.workforceByHubID:
     99             return self.workforceByHubID[hubID]
    100         rem = hubID % 2
    101         if rem == 0:
    102             return WorkforceConfiguration(hubID, transit=True)
    103         return WorkforceConfiguration(hubID, inactive=True)
    104 
    105     def get_configuration(self, hub_id):
    106         return self._GetConfiguration(hub_id)
    107 
    108     def configure(self, hub_id, workforceConfiguration):
    109         self.workforceByHubID[hub_id] = workforceConfiguration
    110         return True
    111 
    112     def get_state(self, hub_id):
    113         pass