Skip to content

Debug

Debug

Source code in mfx/debug.py 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class Debug:
    def __init__(self):
        import os
        full_ioc_serverlist = os.popen(f"netconfig search ioc-mfx* --brief").read().splitlines()
        full_daq_serverlist = os.popen(f"netconfig search daq-mfx* --brief").read().splitlines()
        self.ioc_serverlist = [ioc for ioc in full_ioc_serverlist if not ioc.endswith(
            "-ipmi") and not ioc.endswith("-fez") and not ioc.endswith("-ics")]
        self.daq_serverlist = [ioc for ioc in full_daq_serverlist if not ioc.endswith(
            "-ipmi") and not ioc.endswith("-fez") and not ioc.endswith("-ana")]


    def awr(self, hutch='mfx'):
        """
        Checks if the beamline is ready to take beam

        Parameters
        ----------
        hutch: str, optional
            Specify the hutch you want to check. Default is MFX because it is the best
        """
        import os
        import logging
        logging.info(f"{hutch} Beamline Check")
        os.system(f"/cds/group/pcds/pyps/apps/hutch-python/mfx/scripts/awr {hutch}")


    def motor_check(self):
        import os
        import logging
        logging.info(f"Powering up all available motors")
        os.system(f"/cds/group/pcds/pyps/apps/hutch-python/mfx/scripts/mfxpowerup.sh")


    def check_server(self, server):
        """
        Checks the status of an individual server

        Parameters
        ----------
        server: str, required
            Specify the server name to check. Use debug.server_list('all') to see all servers
        """
        import os
        import logging
        status = None
        if str(server) in self.ioc_serverlist or str(server) in self.daq_serverlist:
            logging.info(f"Checking the status of: {server}")
            status = os.popen(f"/reg/g/pcds/engineering_tools/latest-released/scripts/serverStat {server} status").read().splitlines()
        else:
            logging.info(f"The server you are looking for does not exist please select one of the following")
            self.server_list('ioc')
            self.server_list('daq')
        return status


    def cycle_server(self, server):
        """
        Cycles an individual server

        Parameters
        ----------
        server: str, required
            Specify the server name to cycle. Use debug.server_list('all') to see all servers
        """
        import os
        import logging
        if str(server) in self.ioc_serverlist or str(server) in self.daq_serverlist:
            logging.info(f"Power cycling: {server}")
            os.system(f"/reg/g/pcds/engineering_tools/latest-released/scripts/serverStat {server} cycle")
        else:
            logging.info(f"The server you are looking for does not exist please select one of the following")
            self.server_list('ioc')
            self.server_list('daq')


    def check_all_servers(self, server_type):
        """
        Checks the status of all servers local to MFX

        Parameters
        ----------
        server_type: str, required
            Specify the server type input either 'all', 'ioc', or 'daq
        """
        import logging
        self.error_servers = []
        if str(server_type) == 'all':
            logging.info(f"You've decided to check all {len(self.ioc_serverlist) + len(self.daq_serverlist)} servers.")
            for server in self.ioc_serverlist:
                status = self.check_server(str(server))
                if status[0].endswith('on') and status[1].split(", ")[0].endswith('1)') and status[2].endswith('up'):
                    logging.info(f"Server {server} has passed all tests")
                else:
                    logging.error(f"Server {server} has failed one or more tests and is added to the broken list")
                    logging.error(status)
                    self.error_servers.append(server)

            for server in self.daq_serverlist:
                status = self.check_server(str(server))
                if status[0].endswith('on') and status[1].split(", ")[0].endswith('1)') and status[2].endswith('up'):
                    logging.info(f"Server {server} has passed all tests")
                else:
                    logging.error(f"Server {server} has failed one or more tests and is added to the broken list")
                    logging.error(status)
                    self.error_servers.append(server)

        elif str(server_type) == 'ioc':
            logging.info(f"You've decided to check all {len(self.ioc_serverlist)} ioc servers.")
            for server in self.ioc_serverlist:
                status = self.check_server(str(server))
                if status[0].endswith('on') and status[1].endswith('1)') and status[2].endswith('up'):
                    logging.info(f"Server {server} has passed all tests")
                else:
                    logging.error(f"Server {server} has failed one or more tests and is added to the broken list")
                    self.error_servers.append(server)

        elif str(server_type) == 'daq':
            logging.info(f"You've decided to check all {len(self.daq_serverlist)} daq servers.")
            for server in self.daq_serverlist:
                status = self.check_server(str(server))
                if status[0].endswith('on') and status[1].endswith('1)') and status[2].endswith('up'):
                    logging.info(f"Server {server} has passed all tests")
                else:
                    logging.error(f"Server {server} has failed one or more tests and is added to the broken list")
                    self.error_servers.append(server)
        else:
            logging.warning(f"There is no server of the type you requested. Please use either ioc or daq or all.")

        if len(self.error_servers) != 0:
            logging.warning(f"There is something wrong with the following servers.")
            for server in self.error_servers:
                print(f'{server}')
            cycle = input("\nWould you like to power cycle all error servers? (y/n)? ")

            if cycle.lower() == "y":
                logging.info(f"You've decided to cycle all {len(self.error_servers)} broken servers.")
                for server in self.error_servers:
                    self.cycle_server(str(server))
            else:
                logging.info(f"You've decided not to cycle {len(self.error_servers)} broken servers.")
        else:
            logging.info(f"All {len(self.error_servers)} servers are ready to rock.")

        return self.error_servers


    def server_list(self, server_type):
        """
        Lists servers local to MFX

        Parameters
        ----------
        server_type: str, required
            Specify the server type input either 'all', 'ioc', or 'daq
        """
        import logging
        if str(server_type) == 'all':
            print('IOC SERVERS\n#########################')
            for server in self.ioc_serverlist:
                print(f'{server}')
            print('\nDAQ SERVERS\n#########################')
            for server in self.daq_serverlist:
                print(f'{server}')
        elif str(server_type) == 'ioc':
            print('IOC SERVERS\n#########################')
            for server in self.ioc_serverlist:
                print(f'{server}')
        elif str(server_type) == 'daq':
            print('DAQ SERVERS\n#########################')
            for server in self.daq_serverlist:
                print(f'{server}')
        else:
            logging.warning(f"There is no server of the type you requested. Please use either ioc or daq.")

awr(hutch='mfx')

Checks if the beamline is ready to take beam

Parameters:

Name Type Description Default
hutch

Specify the hutch you want to check. Default is MFX because it is the best

 'mfx'
Source code in mfx/debug.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
def awr(self, hutch='mfx'):
    """
    Checks if the beamline is ready to take beam

    Parameters
    ----------
    hutch: str, optional
        Specify the hutch you want to check. Default is MFX because it is the best
    """
    import os
    import logging
    logging.info(f"{hutch} Beamline Check")
    os.system(f"/cds/group/pcds/pyps/apps/hutch-python/mfx/scripts/awr {hutch}")

check_all_servers(server_type)

Checks the status of all servers local to MFX

Parameters:

Name Type Description Default
server_type

Specify the server type input either 'all', 'ioc', or 'daq

 required
Source code in mfx/debug.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def check_all_servers(self, server_type):
    """
    Checks the status of all servers local to MFX

    Parameters
    ----------
    server_type: str, required
        Specify the server type input either 'all', 'ioc', or 'daq
    """
    import logging
    self.error_servers = []
    if str(server_type) == 'all':
        logging.info(f"You've decided to check all {len(self.ioc_serverlist) + len(self.daq_serverlist)} servers.")
        for server in self.ioc_serverlist:
            status = self.check_server(str(server))
            if status[0].endswith('on') and status[1].split(", ")[0].endswith('1)') and status[2].endswith('up'):
                logging.info(f"Server {server} has passed all tests")
            else:
                logging.error(f"Server {server} has failed one or more tests and is added to the broken list")
                logging.error(status)
                self.error_servers.append(server)

        for server in self.daq_serverlist:
            status = self.check_server(str(server))
            if status[0].endswith('on') and status[1].split(", ")[0].endswith('1)') and status[2].endswith('up'):
                logging.info(f"Server {server} has passed all tests")
            else:
                logging.error(f"Server {server} has failed one or more tests and is added to the broken list")
                logging.error(status)
                self.error_servers.append(server)

    elif str(server_type) == 'ioc':
        logging.info(f"You've decided to check all {len(self.ioc_serverlist)} ioc servers.")
        for server in self.ioc_serverlist:
            status = self.check_server(str(server))
            if status[0].endswith('on') and status[1].endswith('1)') and status[2].endswith('up'):
                logging.info(f"Server {server} has passed all tests")
            else:
                logging.error(f"Server {server} has failed one or more tests and is added to the broken list")
                self.error_servers.append(server)

    elif str(server_type) == 'daq':
        logging.info(f"You've decided to check all {len(self.daq_serverlist)} daq servers.")
        for server in self.daq_serverlist:
            status = self.check_server(str(server))
            if status[0].endswith('on') and status[1].endswith('1)') and status[2].endswith('up'):
                logging.info(f"Server {server} has passed all tests")
            else:
                logging.error(f"Server {server} has failed one or more tests and is added to the broken list")
                self.error_servers.append(server)
    else:
        logging.warning(f"There is no server of the type you requested. Please use either ioc or daq or all.")

    if len(self.error_servers) != 0:
        logging.warning(f"There is something wrong with the following servers.")
        for server in self.error_servers:
            print(f'{server}')
        cycle = input("\nWould you like to power cycle all error servers? (y/n)? ")

        if cycle.lower() == "y":
            logging.info(f"You've decided to cycle all {len(self.error_servers)} broken servers.")
            for server in self.error_servers:
                self.cycle_server(str(server))
        else:
            logging.info(f"You've decided not to cycle {len(self.error_servers)} broken servers.")
    else:
        logging.info(f"All {len(self.error_servers)} servers are ready to rock.")

    return self.error_servers

check_server(server)

Checks the status of an individual server

Parameters:

Name Type Description Default
server

Specify the server name to check. Use debug.server_list('all') to see all servers

 required
Source code in mfx/debug.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def check_server(self, server):
    """
    Checks the status of an individual server

    Parameters
    ----------
    server: str, required
        Specify the server name to check. Use debug.server_list('all') to see all servers
    """
    import os
    import logging
    status = None
    if str(server) in self.ioc_serverlist or str(server) in self.daq_serverlist:
        logging.info(f"Checking the status of: {server}")
        status = os.popen(f"/reg/g/pcds/engineering_tools/latest-released/scripts/serverStat {server} status").read().splitlines()
    else:
        logging.info(f"The server you are looking for does not exist please select one of the following")
        self.server_list('ioc')
        self.server_list('daq')
    return status

cycle_server(server)

Cycles an individual server

Parameters:

Name Type Description Default
server

Specify the server name to cycle. Use debug.server_list('all') to see all servers

 required
Source code in mfx/debug.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def cycle_server(self, server):
    """
    Cycles an individual server

    Parameters
    ----------
    server: str, required
        Specify the server name to cycle. Use debug.server_list('all') to see all servers
    """
    import os
    import logging
    if str(server) in self.ioc_serverlist or str(server) in self.daq_serverlist:
        logging.info(f"Power cycling: {server}")
        os.system(f"/reg/g/pcds/engineering_tools/latest-released/scripts/serverStat {server} cycle")
    else:
        logging.info(f"The server you are looking for does not exist please select one of the following")
        self.server_list('ioc')
        self.server_list('daq')

server_list(server_type)

Lists servers local to MFX

Parameters:

Name Type Description Default
server_type

Specify the server type input either 'all', 'ioc', or 'daq

 required
Source code in mfx/debug.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def server_list(self, server_type):
    """
    Lists servers local to MFX

    Parameters
    ----------
    server_type: str, required
        Specify the server type input either 'all', 'ioc', or 'daq
    """
    import logging
    if str(server_type) == 'all':
        print('IOC SERVERS\n#########################')
        for server in self.ioc_serverlist:
            print(f'{server}')
        print('\nDAQ SERVERS\n#########################')
        for server in self.daq_serverlist:
            print(f'{server}')
    elif str(server_type) == 'ioc':
        print('IOC SERVERS\n#########################')
        for server in self.ioc_serverlist:
            print(f'{server}')
    elif str(server_type) == 'daq':
        print('DAQ SERVERS\n#########################')
        for server in self.daq_serverlist:
            print(f'{server}')
    else:
        logging.warning(f"There is no server of the type you requested. Please use either ioc or daq.")