mine.models.status.ProgramStatus.is_running() - Code Metrics - jacebrowning/mine - Measure and Improve Code Quality continuously with Scrutinizer

mine.models.status.ProgramStatus.is_running() A
last analyzed 2020-12-17 23:19 UTC

↳ Parent: mine.models.status

Complexity

Conditions

Size

Total Lines	11
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	5

Importance

Changes

Metric	Value
cc	5
eloc	8
nop	3
dl	0
loc	11
ccs	7
cts	7
cp	1
crap	5
rs	9.3333
c	0
b	0
f	0

"""Data structures for application/computer status."""

import functools

import log
import yorm

from .timestamp import Timestamp


def log_running(func):
    @functools.wraps(func)
    def wrapped(self, application, computer):
        running = func(self, application, computer)
        log.debug(
            "%s marked as %s on: %s",
            application,
            "started" if running else "stopped",
            computer,
        )
        return running

    return wrapped


def log_starting(func):
    @functools.wraps(func)
    def wrapped(self, application, computer):
        log.debug("Marking %s as started on %s...", application, computer)
        result = func(self, application, computer)
        log.debug("%s marked as started on: %s", application, computer)
        return result

    return wrapped


def log_stopping(func):
    @functools.wraps(func)
    def wrapped(self, application, computer):
        log.debug("Marking %s as stopped on %s...", application, computer)
        result = func(self, application, computer)
        log.debug("%s marked as stopped on: %s", application, computer)
        return result

    return wrapped


@yorm.attr(computer=yorm.types.String)
@yorm.attr(timestamp=Timestamp)
class State(yorm.types.AttributeDictionary):
    """Dictionary of computer state."""

    def __init__(self, computer=None, timestamp=None):
        super().__init__()
        self.computer = computer
        self.timestamp = timestamp or Timestamp()

    def __str__(self):
        return str(self.computer)

    def __lt__(self, other):
        return str(self.computer).lower() < str(other.computer).lower()


@yorm.attr(all=State)
class StateList(yorm.types.SortedList):
    """List of computer states for an application."""


@yorm.attr(application=yorm.types.String)
@yorm.attr(computers=StateList)
@yorm.attr(next=yorm.types.NullableString)
class Status(yorm.types.AttributeDictionary):
    """Dictionary of computers using an application."""

    def __init__(
        self, application=None, computers=None, next=None
    ):  # pylint: disable=redefined-builtin
        super().__init__()
        self.application = application
        self.computers = computers or StateList()
        self.next = next

    def __str__(self):
        return str(self.application)

    def __lt__(self, other):
        return str(self.application).lower() < str(other.application).lower()


@yorm.attr(all=Status)
class StatusList(yorm.types.SortedList):
    """List of application statuses."""


@yorm.attr(applications=StatusList)
@yorm.attr(counter=yorm.types.Integer)
class ProgramStatus(yorm.types.AttributeDictionary):
    """Dictionary of current program status."""

    def __init__(self, applications=None, counter=0):
        super().__init__()
        self.applications = applications or StatusList()
        self.counter = counter

    def find(self, application):
        """Return the application status for an application."""
        for app_status in self.applications:
            if app_status.application == application.name:
                break
        else:
            app_status = Status(application.name)
            self.applications.append(app_status)
        return app_status

    def get_latest(self, application):
        """Get the last computer's name logged as running an application."""
        for status in self.applications:
            if status.application == application.name:
                states = [s for s in status.computers if s.timestamp.active]
                if states:
                    states.sort(key=lambda s: s.timestamp, reverse=True)
                    log.debug(
                        "%s marked as started on: %s",
                        application,
                        ', '.join(str(s) for s in states),
                    )
                    # TODO: consider returning the computer instance?
                    return states[0].computer

        log.debug("marked as started on: nothing")
        return None

    @log_running
    def is_running(self, application, computer):
        """Determine if an application is logged as running on a computer."""
        for status in self.applications:
            if status.application == application.name:
                for state in status.computers:
                    if state.computer == computer.name:
                        return state.timestamp.active

        # Status not found, assume the application is not running
        return False

    def queue(self, application, computer):
        """Record an application as queued for launch on a computer."""
        status = self.find(application)
        status.next = computer.name

    @log_starting

    def start(self, application, computer):
        """Record an application as running on a computer."""
        for status in self.applications:
            if status.application == application.name:
                for state in status.computers:
                    if state.computer == computer.name:
                        self.counter += 1
                        state.timestamp.started = self.counter
                        return
                break
        else:
            status = None

        # Status not found, add the application/computer as started
        self.counter += 1
        state = State(computer.name)
        state.timestamp.started = self.counter
        if status is None:
            status = Status(application.name)
            status.computers.append(state)
            self.applications.append(status)
        else:
            status.computers.append(state)

    @log_stopping

    def stop(self, application, computer):
        """Record an application as no longer running on a computer."""
        for status in self.applications:
            if status.application == application.name:
                for state in status.computers:
                    if state.computer == computer.name:
                        self.counter += 1
                        state.timestamp.stopped = self.counter
                        return
                break
        else:
            status = None

        # Status not found, add the application/computer as stopped
        self.counter += 1
        state = State(computer.name)
        state.timestamp.stopped = self.counter
        if status is None:
            status = Status(application.name)
            status.computers.append(state)
            self.applications.append(status)
        else:
            status.computers.append(state)


1			"""Data structures for application/computer status."""
2
3	1		import functools
4	1
5			import log
6	1		import yorm
7
8	1		from .timestamp import Timestamp
9
10	1
11			def log_running(func):
12			@functools.wraps(func)
13	1		def wrapped(self, application, computer):
14			running = func(self, application, computer)
15	1		log.debug(
16			"%s marked as %s on: %s",
17			application,
18	1		"started" if running else "stopped",
19	1		computer,
20			)
21	1		return running
22	1
23			return wrapped
24
25	1
26			def log_starting(func):
27	1		@functools.wraps(func)
28			def wrapped(self, application, computer):
29			log.debug("Marking %s as started on %s...", application, computer)
30	1		result = func(self, application, computer)
31	1		log.debug("%s marked as started on: %s", application, computer)
32	1		return result
33	1
34	1		return wrapped
35
36
37	1		def log_stopping(func):
38			@functools.wraps(func)
39	1		def wrapped(self, application, computer):
40			log.debug("Marking %s as stopped on %s...", application, computer)
41			result = func(self, application, computer)
42	1		log.debug("%s marked as stopped on: %s", application, computer)
43	1		return result
44	1
45	1		return wrapped
46	1
47
48			@yorm.attr(computer=yorm.types.String)
49	1		@yorm.attr(timestamp=Timestamp)
50	1		class State(yorm.types.AttributeDictionary):
51	1		"""Dictionary of computer state."""
52
53			def __init__(self, computer=None, timestamp=None):
54	1		super().__init__()
55	1		self.computer = computer
56	1		self.timestamp = timestamp or Timestamp()
57	1
58			def __str__(self):
59	1		return str(self.computer)
60	1
61			def __lt__(self, other):
62	1		return str(self.computer).lower() < str(other.computer).lower()
63	1
64
65			@yorm.attr(all=State)
66	1		class StateList(yorm.types.SortedList):
67	1		"""List of computer states for an application."""
68
69
70			@yorm.attr(application=yorm.types.String)
71	1		@yorm.attr(computers=StateList)
72	1		@yorm.attr(next=yorm.types.NullableString)
73	1		class Status(yorm.types.AttributeDictionary):
74	1		"""Dictionary of computers using an application."""
75
76			def __init__(
77	1		self, application=None, computers=None, next=None
78	1		): # pylint: disable=redefined-builtin
79	1		super().__init__()
80	1		self.application = application
81	1		self.computers = computers or StateList()
82			self.next = next
83	1
84	1		def __str__(self):
85			return str(self.application)
86	1
87	1		def __lt__(self, other):
88			return str(self.application).lower() < str(other.application).lower()
89
90	1
91	1		@yorm.attr(all=Status)
92			class StatusList(yorm.types.SortedList):
93			"""List of application statuses."""
94
95	1
96	1		@yorm.attr(applications=StatusList)
97	1		@yorm.attr(counter=yorm.types.Integer)
98			class ProgramStatus(yorm.types.AttributeDictionary):
99			"""Dictionary of current program status."""
100	1
101	1		def __init__(self, applications=None, counter=0):
102	1		super().__init__()
103	1		self.applications = applications or StatusList()
104			self.counter = counter
105	1
106			def find(self, application):
107	1		"""Return the application status for an application."""
108	1		for app_status in self.applications:
109	1		if app_status.application == application.name:
110			break
111	1		else:
112	1		app_status = Status(application.name)
113	1		self.applications.append(app_status)
114			return app_status
115	1
116			def get_latest(self, application):
117	1		"""Get the last computer's name logged as running an application."""
118	1		for status in self.applications:
119	1		if status.application == application.name:
120	1		states = [s for s in status.computers if s.timestamp.active]
121	1		if states:
122	1		states.sort(key=lambda s: s.timestamp, reverse=True)
123			log.debug(
124			"%s marked as started on: %s",
125	1		application,
126			', '.join(str(s) for s in states),
127	1		)
128	1		# TODO: consider returning the computer instance?
129			return states[0].computer
130	1
131			log.debug("marked as started on: nothing")
132			return None
133	1
134	1		@log_running
135	1		def is_running(self, application, computer):
136	1		"""Determine if an application is logged as running on a computer."""
137	1		for status in self.applications:
138			if status.application == application.name:
139			for state in status.computers:
140	1		if state.computer == computer.name:
141			return state.timestamp.active
142	1
143			# Status not found, assume the application is not running
144	1		return False
145	1
146			def queue(self, application, computer):
147	1		"""Record an application as queued for launch on a computer."""
148			status = self.find(application)
149			status.next = computer.name
150	1
151	1	View Code Duplication	@log_starting
			0 ignored issues – show Duplication introduced 2016-05-26 18:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
152	1		def start(self, application, computer):
153	1		"""Record an application as running on a computer."""
154	1		for status in self.applications:
155	1		if status.application == application.name:
156	1		for state in status.computers:
157	1		if state.computer == computer.name:
158			self.counter += 1
159	1		state.timestamp.started = self.counter
160			return
161			break
162	1		else:
163	1		status = None
164	1
165	1		# Status not found, add the application/computer as started
166	1		self.counter += 1
167	1		state = State(computer.name)
168	1		state.timestamp.started = self.counter
169			if status is None:
170	1		status = Status(application.name)
171			status.computers.append(state)
172	1		self.applications.append(status)
173			else:
174			status.computers.append(state)
175	1
176	1	View Code Duplication	@log_stopping
			0 ignored issues – show Duplication introduced 2016-05-26 18:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
177	1		def stop(self, application, computer):
178	1		"""Record an application as no longer running on a computer."""
179	1		for status in self.applications:
180	1		if status.application == application.name:
181	1		for state in status.computers:
182	1		if state.computer == computer.name:
183			self.counter += 1
184	1		state.timestamp.stopped = self.counter
185			return
186			break
187	1		else:
188	1		status = None
189	1
190	1		# Status not found, add the application/computer as stopped
191	1		self.counter += 1
192	1		state = State(computer.name)
193	1		state.timestamp.stopped = self.counter
194			if status is None:
195	1		status = Status(application.name)
196			status.computers.append(state)
197			self.applications.append(status)
198			else:
199			status.computers.append(state)
200

jacebrowning / mine

mine.models.status.ProgramStatus.is_running() A last analyzed 2020-12-17 23:19 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

mine.models.status.ProgramStatus.is_running() A
last analyzed 2020-12-17 23:19 UTC