generated from opentensor/bittensor-subnet-template
-
Notifications
You must be signed in to change notification settings - Fork 10
/
Copy pathminer.py
176 lines (133 loc) · 8.25 KB
/
miner.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
174
175
176
# The MIT License (MIT)
# Copyright © 2023 Yuma Rao
# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
# documentation files (the “Software”), to deal in the Software without restriction, including without limitation
# the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in all copies or substantial portions of
# the Software.
# THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
# THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import time
import typing
import bittensor as bt
import pytesseract
# Bittensor OCR Miner
import ocr_subnet
# import base miner class which takes care of most of the boilerplate
from ocr_subnet.base.miner import BaseMinerNeuron
class Miner(BaseMinerNeuron):
"""
OCR miner neuron class. You may also want to override the blacklist and priority functions according to your needs.
This class inherits from the BaseMinerNeuron class, which in turn inherits from BaseNeuron. The BaseNeuron class takes care of routine tasks such as setting up wallet, subtensor, metagraph, logging directory, parsing config, etc. You can override any of the methods in BaseNeuron if you need to customize the behavior.
This class provides reasonable default behavior for a miner such as blacklisting unrecognized hotkeys, prioritizing requests based on stake, and forwarding requests to the forward function. If you need to define custom
"""
def __init__(self, config=None):
super(Miner, self).__init__(config=config)
bt.logging.info(f'Miner running tesseract: {pytesseract.get_tesseract_version()} and pytesseract version: {pytesseract.get_version()}')
async def forward(
self, synapse: ocr_subnet.protocol.OCRSynapse
) -> ocr_subnet.protocol.OCRSynapse:
"""
Processes the incoming OCR synapse and attaches the response to the synapse.
Args:
synapse (ocr_subnet.protocol.OCRSynapse): The synapse object containing the image data.
Returns:
ocr_subnet.protocol.OCRSynapse: The synapse object with the 'response' field set to the extracted data.
"""
# Get image data
image = ocr_subnet.utils.image.deserialize(base64_string=synapse.base64_image)
# Use pytesseract to get the data
data = pytesseract.image_to_data(image, output_type=pytesseract.Output.DICT)
response = []
for i in range(len(data['text'])):
if data['text'][i].strip() != '': # This filters out empty text results
x1, y1, width, height = data['left'][i], data['top'][i], data['width'][i], data['height'][i]
if width * height < 10: # This filters out small boxes (likely noise)
continue
x2, y2 = x1 + width, y1 + height
# Here we don't have font information, so we'll omit that.
# Pytesseract does not extract font family or size information.
entry = {
'position': [x1, y1, x2, y2],
'text': data['text'][i]
}
response.append(entry)
# Merge together words into sections, which are on the same line (same y value) and are close together (small distance in x)
response = ocr_subnet.utils.process.group_and_merge_boxes(response)
# Sort sections by y, then sort by x so that they read left to right and top to bottom
response = sorted(response, key=lambda item: (item['position'][1], item['position'][0]))
# Attach response to synapse and return it.
synapse.response = response
return synapse
async def blacklist(
self, synapse: ocr_subnet.protocol.OCRSynapse
) -> typing.Tuple[bool, str]:
"""
Determines whether an incoming request should be blacklisted and thus ignored. Your implementation should
define the logic for blacklisting requests based on your needs and desired security parameters.
Blacklist runs before the synapse data has been deserialized (i.e. before synapse.data is available).
The synapse is instead contructed via the headers of the request. It is important to blacklist
requests before they are deserialized to avoid wasting resources on requests that will be ignored.
Args:
synapse (template.protocol.OCRSynapse): A synapse object constructed from the headers of the incoming request.
Returns:
Tuple[bool, str]: A tuple containing a boolean indicating whether the synapse's hotkey is blacklisted,
and a string providing the reason for the decision.
This function is a security measure to prevent resource wastage on undesired requests. It should be enhanced
to include checks against the metagraph for entity registration, validator status, and sufficient stake
before deserialization of synapse data to minimize processing overhead.
Example blacklist logic:
- Reject if the hotkey is not a registered entity within the metagraph.
- Consider blacklisting entities that are not validators or have insufficient stake.
In practice it would be wise to blacklist requests from entities that are not validators, or do not have
enough stake. This can be checked via metagraph.S and metagraph.validator_permit. You can always attain
the uid of the sender via a metagraph.hotkeys.index( synapse.dendrite.hotkey ) call.
Otherwise, allow the request to be processed further.
"""
# TODO(developer): Define how miners should blacklist requests.
if synapse.dendrite.hotkey not in self.metagraph.hotkeys:
# Ignore requests from unrecognized entities.
bt.logging.trace(
f"Blacklisting unrecognized hotkey {synapse.dendrite.hotkey}"
)
return True, "Unrecognized hotkey"
bt.logging.trace(
f"Not Blacklisting recognized hotkey {synapse.dendrite.hotkey}"
)
return False, "Hotkey recognized!"
async def priority(self, synapse: ocr_subnet.protocol.OCRSynapse) -> float:
"""
The priority function determines the order in which requests are handled. More valuable or higher-priority
requests are processed before others. You should design your own priority mechanism with care.
This implementation assigns priority to incoming requests based on the calling entity's stake in the metagraph.
Args:
synapse (template.protocol.OCRSynapse): The synapse object that contains metadata about the incoming request.
Returns:
float: A priority score derived from the stake of the calling entity.
Miners may recieve messages from multiple entities at once. This function determines which request should be
processed first. Higher values indicate that the request should be processed first. Lower values indicate
that the request should be processed later.
Example priority logic:
- A higher stake results in a higher priority value.
"""
# TODO(developer): Define how miners should prioritize requests.
caller_uid = self.metagraph.hotkeys.index(
synapse.dendrite.hotkey
) # Get the caller index.
prirority = float(
self.metagraph.S[caller_uid]
) # Return the stake as the priority.
bt.logging.trace(
f"Prioritizing {synapse.dendrite.hotkey} with value: ", prirority
)
return prirority
# This is the main function, which runs the miner.
if __name__ == "__main__":
with Miner() as miner:
while True:
bt.logging.info("Miner running...", time.time())
time.sleep(5)