| #!/usr/bin/env python3 |
| # |
| # Copyright 2017 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Implementation of ChromeOS Factory Shopfloor Service, version 1.0.""" |
| |
| import argparse |
| import logging |
| import socket |
| import socketserver |
| import xmlrpc.server |
| |
| DEFAULT_SERVER_PORT = 8090 |
| DEFAULT_SERVER_ADDRESS = '0.0.0.0' |
| |
| KEY_SERIAL_NUMBER = 'serials.serial_number' |
| KEY_MLB_SERIAL_NUMBER = 'serials.mlb_serial_number' |
| |
| |
| class ShopfloorService: |
| |
| def __init__(self): |
| pass |
| |
| def GetVersion(self): |
| """Returns the version of supported protocol.""" |
| return '1.0' |
| |
| def NotifyStart(self, data, station): |
| """Notifies shopfloor backend that DUT is starting a manufacturing station. |
| |
| Args: |
| data: A FactoryDeviceData instance. |
| station: A string to indicate manufacturing station. |
| |
| Returns: |
| A mapping in DeviceData format. |
| """ |
| logging.info('DUT %s Entering station %s', data.get(KEY_MLB_SERIAL_NUMBER), |
| station) |
| return {} |
| |
| def NotifyEnd(self, data, station): |
| """Notifies shopfloor backend that DUT has finished a manufacturing station. |
| |
| Args: |
| data: A FactoryDeviceData instance. |
| station: A string to indicate manufacturing station. |
| |
| Returns: |
| A mapping in DeviceData format. |
| """ |
| logging.info('DUT %s Leaving station %s', data.get(KEY_MLB_SERIAL_NUMBER), |
| station) |
| return {} |
| |
| def NotifyEvent(self, data, event): |
| """Notifies shopfloor backend that the DUT has performed an event. |
| |
| Args: |
| data: A FactoryDeviceData instance. |
| event: A string to indicate manufacturing event. |
| |
| Returns: |
| A mapping in FactoryDeviceData format. |
| """ |
| assert event in ['Finalize', 'Refinalize'] |
| logging.info('DUT %s sending event %s', data.get(KEY_MLB_SERIAL_NUMBER), |
| event) |
| return {} |
| |
| def GetDeviceInfo(self, data): |
| """Returns information about the device's expected configuration. |
| |
| Args: |
| data: A FactoryDeviceData instance. |
| |
| Returns: |
| A mapping in DeviceData format. |
| """ |
| logging.info('DUT %s requesting device information', |
| data.get(KEY_MLB_SERIAL_NUMBER)) |
| return {'vpd.ro.region': 'us', |
| 'vpd.rw.ubind_attribute': '', |
| 'vpd.rw.gbind_attribute': ''} |
| |
| def ActivateRegCode(self, ubind_attribute, gbind_attribute, hwid): |
| """Notifies shopfloor backend that DUT has deployed a registration code. |
| |
| Args: |
| ubind_attribute: A string for user registration code. |
| gbind_attribute: A string for group registration code. |
| hwid: A string for the HWID of the device. |
| |
| Returns: |
| A mapping in DeviceData format. |
| """ |
| logging.info('DUT <hwid=%s> requesting to activate regcode(u=%s,g=%s)', |
| hwid, ubind_attribute, gbind_attribute) |
| return {} |
| |
| def UpdateTestResult(self, data, test_id, status, details=None): |
| """Sends the specified test result to shopfloor backend. |
| |
| Args: |
| data: A FactoryDeviceData instance. |
| test_id: A string as identifier of the given test. |
| status: A string from TestState; one of PASSED, FAILED, SKIPPED, or |
| FAILED_AND_WAIVED. |
| details: (optional) A mapping to provide more details, including at least |
| 'error_message'. |
| |
| Returns: |
| A mapping in DeviceData format. If 'action' is included, DUT software |
| should follow the value to decide how to proceed. |
| """ |
| logging.info('DUT %s updating test results for <%s> with status <%s> %s', |
| data.get(KEY_MLB_SERIAL_NUMBER), test_id, status, |
| details.get('error_message') if details else '') |
| return {} |
| |
| |
| class ThreadedXMLRPCServer(socketserver.ThreadingMixIn, |
| xmlrpc.server.SimpleXMLRPCServer): |
| """A threaded XML RPC Server.""" |
| |
| |
| def RunAsServer(address, port, instance, logRequest=False): |
| """Starts a XML-RPC server in given address and port. |
| |
| Args: |
| address: Address to bind server. |
| port: Port for server to listen. |
| instance: Server instance for incoming XML RPC requests. |
| logRequests: Boolean to indicate if we should log requests. |
| |
| Returns: |
| Never returns if the server is started successfully, otherwise some |
| exception will be raised. |
| """ |
| server = ThreadedXMLRPCServer((address, port), allow_none=True, |
| logRequests=logRequest) |
| server.register_introspection_functions() |
| server.register_instance(instance) |
| logging.info('Server started: http://%s:%s "%s" version %s', |
| address, port, instance.__class__.__name__, |
| instance.GetVersion()) |
| server.serve_forever() |
| |
| |
| def main(): |
| """Main entry when being invoked by command line.""" |
| parser = argparse.ArgumentParser() |
| parser.add_argument( |
| '-a', '--address', metavar='ADDR', default=DEFAULT_SERVER_ADDRESS, |
| help='address to bind (default: %s)' % DEFAULT_SERVER_ADDRESS) |
| parser.add_argument( |
| '-p', '--port', metavar='PORT', type=int, default=DEFAULT_SERVER_PORT, |
| help='port to bind (default: %s)' % DEFAULT_SERVER_PORT) |
| parser.add_argument('-v', '--verbose', default=False, action='store_true', |
| help='provide verbose logs for debugging.') |
| args = parser.parse_args() |
| |
| log_format = '%(asctime)s %(levelname)s %(message)s' |
| logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, |
| format=log_format) |
| |
| # Disable all DNS lookups, since otherwise the logging code may try to |
| # resolve IP addresses, which may delay request handling. |
| socket.getfqdn = lambda name: name or 'localhost' |
| |
| try: |
| RunAsServer(address=args.address, port=args.port, |
| instance=ShopfloorService(), |
| logRequest=args.verbose) |
| finally: |
| logging.warning('Server stopped.') |
| |
| |
| if __name__ == '__main__': |
| main() |