<commit_before>
###############################################################################
#   Copyright 2015 The University of Texas at Austin                          #
#                                                                             #
#   Licensed under the Apache License, Version 2.0 (the "License");           #
#   you may not use this file except in compliance with the License.          #
#   You may obtain a copy of the License at                                   #
#                                                                             #
#       http://www.apache.org/licenses/LICENSE-2.0                            #
#                                                                             #
#   Unless required by applicable law or agreed to in writing, software       #
#   distributed under the License is distributed on an "AS IS" BASIS,         #
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  #
#   See the License for the specific language governing permissions and       #
#   limitations under the License.                                            #
###############################################################################

import commands
import datetime
import itertools
import os
import re

import ipf.dt
from ipf.error import StepError
from ipf.log import LogFileWatcher

from . import computing_activity
from . import computing_manager
from . import computing_service
from . import computing_share
from . import execution_environment

#######################################################################################################################

class ComputingServiceStep(computing_service.ComputingServiceStep):

    def __init__(self):
        computing_service.ComputingServiceStep.__init__(self)

    def _run(self):
        service = computing_service.ComputingService()
        service.Name = "SLURM"
        service.Capability = ["executionmanagement.jobexecution",
                              "executionmanagement.jobdescription",
                              "executionmanagement.jobmanager",
                              "executionmanagement.executionandplanning",
                              "executionmanagement.reservation",
                              ]
        service.Type = "ipf.SLURM"
        service.QualityLevel = "production"

        return service
        
#######################################################################################################################

class ComputingManagerStep(computing_manager.ComputingManagerStep):

    def __init__(self):
        computing_manager.ComputingManagerStep.__init__(self)

    def _run(self):
        manager = computing_manager.ComputingManager()
        manager.ProductName = "SLURM"
        manager.Name = "SLURM"
        manager.Reservation = True
        #self.BulkSubmission = True

        return manager

#######################################################################################################################

class ComputingActivitiesStep(computing_activity.ComputingActivitiesStep):

    def __init__(self):
        computing_activity.ComputingActivitiesStep.__init__(self)

        self._acceptParameter("scontrol","the path to the SLURM scontrol program (default 'scontrol')",False)

    def _run(self):
        # squeue command doesn't provide submit time
        scontrol = self.params.get("scontrol","scontrol")

        cmd = scontrol + " show job"
        self.debug("running "+cmd)
        status, output = commands.getstatusoutput(cmd)
        if status != 0:
            raise StepError("scontrol failed: "+output+"\n")

        jobs = []
        for job_str in output.split("\n\n"):
            job = _getJob(self,job_str)
            if self._includeQueue(job.Queue):
                jobs.append(job)

        # scontrol doesn't sort jobs, so sort them by priority, job id, and state before returning
        jobs = sorted(jobs,key=lambda job: int(job.LocalIDFromManager))
        jobs = sorted(jobs,key=lambda job: -job.Extension["Priority"])
        jobs = sorted(jobs,key=self._jobStateKey)

        # loop through and set WaitingPosition for waiting jobs
        waiting_pos = 1
        for job in jobs:
            if job.State[0] == computing_activity.ComputingActivity.STATE_PENDING or \
                    job.State[0] == computing_activity.ComputingActivity.STATE_HELD:
                job.WaitingPosition = waiting_pos
                waiting_pos += 1

        return jobs

def _getJob(step, job_str):
    job = computing_activity.ComputingActivity()

    m = re.search("JobId=(\S+)",job_str)
    if m is not None:
        job.LocalIDFromManager = m.group(1)
    m = re.search(" Name=(\S+)",job_str)
    if m is not None:
        job.Name = m.group(1)
    else:
        m = re.search(" JobName=(\S+)",job_str)
        if m is not None:
            job.Name = m.group(1)
    m = re.search("UserId=(\S+)\(",job_str)
    if m is not None:
        job.LocalOwner = m.group(1)
    m = re.search("Account=(\S+)",job_str)
    if m is not None:
        job.Extension["LocalAccount"] = m.group(1)
    m = re.search("Partition=(\S+)",job_str)
    if m is not None:
        job.Queue = m.group(1)
    m = re.search("Reservation=(\S+)",job_str)
    if m is not None and m.group(1) != "(null)":
        job.Queue = m.group(1)  # override
        job.Extension["ReservationName"] = m.group(1)
        job.ResourceID = "urn:glue2:ExecutionEnvironment:%s.%s" % (m.group(1),step.resource_name)
    m = re.search("JobState=(\S+)",job_str)
    if m is not None:
        state = m.group(1)  # see squeue man page for state descriptions
        if state == "CANCELLED":
            job.State = [computing_activity.ComputingActivity.STATE_TERMINATED]<commit_msg>have ComptingActivity.Queue be the same as ComputingShare.MappingQueue for reservations<commit_after>
###############################################################################
#   Copyright 2015 The University of Texas at Austin                          #
#                                                                             #
#   Licensed under the Apache License, Version 2.0 (the "License");           #
#   you may not use this file except in compliance with the License.          #
#   You may obtain a copy of the License at                                   #
#                                                                             #
#       http://www.apache.org/licenses/LICENSE-2.0                            #
#                                                                             #
#   Unless required by applicable law or agreed to in writing, software       #
#   distributed under the License is distributed on an "AS IS" BASIS,         #
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  #
#   See the License for the specific language governing permissions and       #
#   limitations under the License.                                            #
###############################################################################

import commands
import datetime
import itertools
import os
import re

import ipf.dt
from ipf.error import StepError
from ipf.log import LogFileWatcher

from . import computing_activity
from . import computing_manager
from . import computing_service
from . import computing_share
from . import execution_environment

#######################################################################################################################

class ComputingServiceStep(computing_service.ComputingServiceStep):

    def __init__(self):
        computing_service.ComputingServiceStep.__init__(self)

    def _run(self):
        service = computing_service.ComputingService()
        service.Name = "SLURM"
        service.Capability = ["executionmanagement.jobexecution",
                              "executionmanagement.jobdescription",
                              "executionmanagement.jobmanager",
                              "executionmanagement.executionandplanning",
                              "executionmanagement.reservation",
                              ]
        service.Type = "ipf.SLURM"
        service.QualityLevel = "production"

        return service
        
#######################################################################################################################

class ComputingManagerStep(computing_manager.ComputingManagerStep):

    def __init__(self):
        computing_manager.ComputingManagerStep.__init__(self)

    def _run(self):
        manager = computing_manager.ComputingManager()
        manager.ProductName = "SLURM"
        manager.Name = "SLURM"
        manager.Reservation = True
        #self.BulkSubmission = True

        return manager

#######################################################################################################################

class ComputingActivitiesStep(computing_activity.ComputingActivitiesStep):

    def __init__(self):
        computing_activity.ComputingActivitiesStep.__init__(self)

        self._acceptParameter("scontrol","the path to the SLURM scontrol program (default 'scontrol')",False)

    def _run(self):
        # squeue command doesn't provide submit time
        scontrol = self.params.get("scontrol","scontrol")

        cmd = scontrol + " show job"
        self.debug("running "+cmd)
        status, output = commands.getstatusoutput(cmd)
        if status != 0:
            raise StepError("scontrol failed: "+output+"\n")

        jobs = []
        for job_str in output.split("\n\n"):
            job = _getJob(self,job_str)
            if self._includeQueue(job.Queue):
                jobs.append(job)

        # scontrol doesn't sort jobs, so sort them by priority, job id, and state before returning
        jobs = sorted(jobs,key=lambda job: int(job.LocalIDFromManager))
        jobs = sorted(jobs,key=lambda job: -job.Extension["Priority"])
        jobs = sorted(jobs,key=self._jobStateKey)

        # loop through and set WaitingPosition for waiting jobs
        waiting_pos = 1
        for job in jobs:
            if job.State[0] == computing_activity.ComputingActivity.STATE_PENDING or \
                    job.State[0] == computing_activity.ComputingActivity.STATE_HELD:
                job.WaitingPosition = waiting_pos
                waiting_pos += 1

        return jobs

def _getJob(step, job_str):
    job = computing_activity.ComputingActivity()

    m = re.search("JobId=(\S+)",job_str)
    if m is not None:
        job.LocalIDFromManager = m.group(1)
    m = re.search(" Name=(\S+)",job_str)
    if m is not None:
        job.Name = m.group(1)
    else:
        m = re.search(" JobName=(\S+)",job_str)
        if m is not None:
            job.Name = m.group(1)
    m = re.search("UserId=(\S+)\(",job_str)
    if m is not None:
        job.LocalOwner = m.group(1)
    m = re.search("Account=(\S+)",job_str)
    if m is not None:
        job.Extension["LocalAccount"] = m.group(1)
    m = re.search("Partition=(\S+)",job_str)
    if m is not None:
        job.Queue = m.group(1)
    m = re.search("Reservation=(\S+)",job_str)
    if m is not None and m.group(1) != "(null)":
        job.Extension["ReservationName"] = m.group(1)
        job.ResourceID = "urn:glue2:ExecutionEnvironment:%s.%s" % (m.group(1),step.resource_name)
    m = re.search("JobState=(\S+)",job_str)
    if m is not None:
        state = m.group(1)  # see squeue man page for state descriptions
        if state == "CANCELLED":
            job.State = [computing_activity.ComputingActivity.STATE_TERMINATED]
{'fraction_non_alphanumeric': 0.17431745758332615, 'fraction_numerical': 0.004220136077857204, 'mean_word_length': 2.3110921015112633, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1955427', 'n_tokens_mistral': 2948, 'n_tokens_neox': 2839, 'n_words': 736}